1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.newsclub.net.unix;
19
20 import static java.util.Objects.requireNonNull;
21
22 import java.io.FileDescriptor;
23 import java.io.IOException;
24 import java.net.ProtocolFamily;
25 import java.net.SocketAddress;
26 import java.net.SocketOption;
27 import java.net.StandardProtocolFamily;
28 import java.nio.ByteBuffer;
29 import java.nio.channels.SocketChannel;
30 import java.nio.channels.spi.SelectorProvider;
31 import java.util.Objects;
32 import java.util.Set;
33 import java.util.concurrent.atomic.AtomicBoolean;
34
35 import org.eclipse.jdt.annotation.NonNull;
36
37 import com.kohlschutter.annotations.compiletime.SuppressFBWarnings;
38
39
40
41
42
43
44
45 public abstract class AFSocketChannel<A extends AFSocketAddress> extends SocketChannel implements
46 AFSomeSocket, AFSocketExtensions, AFSomeSocketChannel {
47 private final @NonNull AFSocket<A> afSocket;
48 private final AtomicBoolean connectPending = new AtomicBoolean(false);
49
50
51
52
53
54
55
56 @SuppressWarnings("all")
57 @SuppressFBWarnings("CT_CONSTRUCTOR_THROW")
58 protected AFSocketChannel(AFSocket<A> socket, AFSelectorProvider<A> sp) {
59 super(sp);
60 this.afSocket = Objects.requireNonNull(socket);
61 }
62
63
64
65
66
67
68 protected final AFSocket<A> getAFSocket() {
69 return afSocket;
70 }
71
72
73
74
75
76
77 @FunctionalInterface
78 protected interface AFSocketSupplier<A extends AFSocketAddress> {
79
80
81
82
83
84
85 AFSocket<A> newInstance() throws IOException;
86 }
87
88
89
90
91
92
93
94
95
96
97 protected static final <A extends AFSocketAddress> AFSocketChannel<A> open(
98 AFSocketSupplier<A> supplier) throws IOException {
99 return supplier.newInstance().getChannel();
100 }
101
102
103
104
105
106
107
108
109
110
111 protected static final <A extends AFSocketAddress> AFSocketChannel<A> open(
112 AFSocketSupplier<A> supplier, SocketAddress remote) throws IOException {
113 @SuppressWarnings("resource")
114 AFSocketChannel<A> sc = open(supplier);
115 try {
116 sc.connect(remote);
117 } catch (Throwable x) {
118 try {
119 sc.close();
120 } catch (Throwable suppressed) {
121 x.addSuppressed(suppressed);
122 }
123 throw x;
124 }
125 assert sc.isConnected();
126 return sc;
127 }
128
129 @SuppressWarnings("unchecked")
130 @Override
131 public final <T> T getOption(SocketOption<T> name) throws IOException {
132 if (name instanceof AFSocketOption<?>) {
133 return getAFCore().getOption((AFSocketOption<T>) name);
134 }
135 Integer optionId = SocketOptionsMapper.resolve(name);
136 if (optionId == null) {
137 throw new UnsupportedOperationException("unsupported option");
138 } else {
139 return (T) afSocket.getAFImpl().getOption(optionId);
140 }
141 }
142
143 @Override
144 public final <T> AFSocketChannel<A> setOption(SocketOption<T> name, T value) throws IOException {
145 if (name instanceof AFSocketOption<?>) {
146 getAFCore().setOption((AFSocketOption<T>) name, value);
147 return this;
148 }
149 Integer optionId = SocketOptionsMapper.resolve(name);
150 if (optionId == null) {
151 throw new UnsupportedOperationException("unsupported option");
152 } else {
153 afSocket.getAFImpl().setOption(optionId, value);
154 }
155 return this;
156 }
157
158 @Override
159 public final Set<SocketOption<?>> supportedOptions() {
160 return SocketOptionsMapper.SUPPORTED_SOCKET_OPTIONS;
161 }
162
163 @Override
164 public final AFSocketChannel<A> bind(SocketAddress local) throws IOException {
165 afSocket.bind(local);
166 return this;
167 }
168
169 @Override
170 public final AFSocketChannel<A> shutdownInput() throws IOException {
171 afSocket.getAFImpl().shutdownInput();
172 return this;
173 }
174
175 @Override
176 public final AFSocketChannel<A> shutdownOutput() throws IOException {
177 afSocket.getAFImpl().shutdownOutput();
178 return this;
179 }
180
181 @Override
182 @SuppressFBWarnings("EI_EXPOSE_REP")
183 public final AFSocket<A> socket() {
184 return afSocket;
185 }
186
187 @Override
188 public final boolean isConnected() {
189 boolean connected = afSocket.isConnected();
190 if (connected) {
191 connectPending.set(false);
192 }
193 return connected;
194 }
195
196 @Override
197 public final boolean isConnectionPending() {
198 return connectPending.get();
199 }
200
201 @Override
202 public final boolean connect(SocketAddress remote) throws IOException {
203 boolean complete = false;
204 Exception exception = null;
205 try {
206 begin();
207 boolean connected = afSocket.connect0(remote, 0);
208 if (!connected) {
209 connectPending.set(true);
210 }
211 complete = true;
212 return connected;
213 } catch (IOException e) {
214 throw InterruptibleChannelUtil.ioExceptionOrThrowRuntimeException(
215 (exception = InterruptibleChannelUtil.handleException(this, e)));
216 } finally {
217 InterruptibleChannelUtil.endInterruptable(this, this::end, complete, exception);
218 }
219 }
220
221 @Override
222 public final boolean finishConnect() throws IOException {
223 if (isConnected()) {
224 return true;
225 } else if (!isConnectionPending()) {
226 return false;
227 }
228
229 boolean complete = false;
230 Exception exception = null;
231 try {
232 begin();
233 boolean connected = NativeUnixSocket.finishConnect(afSocket.getFileDescriptor())
234 || isConnected();
235 if (connected) {
236 connectPending.set(false);
237 }
238 complete = true;
239 return connected;
240 } catch (IOException e) {
241 throw InterruptibleChannelUtil.ioExceptionOrThrowRuntimeException(
242 (exception = InterruptibleChannelUtil.handleException(this, e)));
243 } finally {
244 InterruptibleChannelUtil.endInterruptable(this, this::end, complete, exception);
245 }
246 }
247
248 @Override
249 public final A getRemoteAddress() throws IOException {
250 return getRemoteSocketAddress();
251 }
252
253 @Override
254 public final A getRemoteSocketAddress() {
255 return afSocket.getRemoteSocketAddress();
256 }
257
258 @Override
259 public final int read(ByteBuffer dst) throws IOException {
260 boolean complete = false;
261 Exception exception = null;
262 try {
263 begin();
264 int read = afSocket.getAFImpl().read(dst, null);
265 complete = true;
266 return read;
267 } catch (IOException e) {
268 throw InterruptibleChannelUtil.ioExceptionOrThrowRuntimeException(
269 (exception = InterruptibleChannelUtil.handleException(this, e)));
270 } finally {
271 InterruptibleChannelUtil.endInterruptable(this, this::end, complete, exception);
272 }
273 }
274
275 @Override
276 public final long read(ByteBuffer[] dsts, int offset, int length) throws IOException {
277 if (length == 0) {
278 return 0;
279 }
280
281 return read(dsts[offset]);
282 }
283
284 @Override
285 public final long write(ByteBuffer[] srcs, int offset, int length) throws IOException {
286 if (length == 0) {
287 return 0;
288 }
289
290 return write(srcs[offset]);
291 }
292
293 @Override
294 public final int write(ByteBuffer src) throws IOException {
295 boolean complete = false;
296 Exception exception = null;
297 try {
298 begin();
299 int written = afSocket.getAFImpl().write(src);
300 complete = true;
301 return written;
302 } catch (IOException e) {
303 throw InterruptibleChannelUtil.ioExceptionOrThrowRuntimeException(
304 (exception = InterruptibleChannelUtil.handleException(this, e)));
305 } finally {
306 InterruptibleChannelUtil.endInterruptable(this, this::end, complete, exception);
307 }
308 }
309
310 @Override
311 public final A getLocalAddress() throws IOException {
312 return getLocalSocketAddress();
313 }
314
315 @Override
316 public final A getLocalSocketAddress() {
317 return afSocket.getLocalSocketAddress();
318 }
319
320 @Override
321 protected final void implCloseSelectableChannel() throws IOException {
322 afSocket.close();
323 }
324
325 @Override
326 protected final void implConfigureBlocking(boolean block) throws IOException {
327 getAFCore().implConfigureBlocking(block);
328 }
329
330 @Override
331 public final int getAncillaryReceiveBufferSize() {
332 return afSocket.getAncillaryReceiveBufferSize();
333 }
334
335 @Override
336 public final void setAncillaryReceiveBufferSize(int size) {
337 afSocket.setAncillaryReceiveBufferSize(size);
338 }
339
340 @Override
341 public final void ensureAncillaryReceiveBufferSize(int minSize) {
342 afSocket.ensureAncillaryReceiveBufferSize(minSize);
343 }
344
345 final AFSocketCore getAFCore() {
346 return afSocket.getAFImpl().getCore();
347 }
348
349 @Override
350 public final FileDescriptor getFileDescriptor() throws IOException {
351 return afSocket.getFileDescriptor();
352 }
353
354 @Override
355 public final String toString() {
356 return super.toString() + afSocket.toStringSuffix();
357 }
358
359 @Override
360 public void setShutdownOnClose(boolean enabled) {
361 getAFCore().setShutdownOnClose(enabled);
362 }
363
364
365
366
367
368
369
370
371
372
373
374
375
376 @SuppressFBWarnings("HSM_HIDING_METHOD")
377 public static SocketChannel open(ProtocolFamily family) throws IOException {
378 requireNonNull(family);
379
380 if (family instanceof AFProtocolFamily) {
381 return ((AFProtocolFamily) family).openSocketChannel();
382 } else if ("UNIX".equals(family.name())) {
383 return AFUNIXSocketChannel.open();
384 } else if (family instanceof StandardProtocolFamily) {
385 return SocketChannel.open();
386 } else {
387 throw new UnsupportedOperationException("Protocol family not supported");
388 }
389 }
390 }