View Javadoc
1   /*
2    * junixsocket
3    *
4    * Copyright 2009-2026 Christian Kohlschütter
5    *
6    * Licensed under the Apache License, Version 2.0 (the "License");
7    * you may not use this file except in compliance with the License.
8    * You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
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   * A selectable channel for stream-oriented connecting sockets.
41   *
42   * @param <A> The concrete {@link AFSocketAddress} that is supported by this type.
43   * @author Christian Kohlschütter
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     * Creates a new socket channel for the given socket, using the given {@link SelectorProvider}.
52     *
53     * @param socket The socket.
54     * @param sp The {@link SelectorProvider}.
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     * Returns the corresponding {@link AFSocket}.
65     *
66     * @return The corresponding socket.
67     */
68    protected final AFSocket<A> getAFSocket() {
69      return afSocket;
70    }
71  
72    /**
73     * A reference to a method that provides an {@link AFSocket} instance.
74     *
75     * @param <A> The concrete {@link AFSocketAddress} that is supported by this type.
76     */
77    @FunctionalInterface
78    protected interface AFSocketSupplier<A extends AFSocketAddress> {
79      /**
80       * Returns a new {@link AFSocket} instance.
81       *
82       * @return The instance.
83       * @throws IOException on error.
84       */
85      AFSocket<A> newInstance() throws IOException;
86    }
87  
88    /**
89     * Opens a socket channel.
90     *
91     * @param <A> The concrete {@link AFSocketAddress} that is supported by this type.
92     * @param supplier The AFSocketChannel constructor.
93     *
94     * @return The new channel
95     * @throws IOException on error.
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    * Opens a socket channel, connecting to the given socket address.
104    *
105    * @param <A> The concrete {@link AFSocketAddress} that is supported by this type.
106    * @param remote The socket address to connect to.
107    * @param supplier The AFSocketChannel constructor.
108    * @return The new channel
109    * @throws IOException on error.
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) { // NOPMD
118       try {
119         sc.close();
120       } catch (Throwable suppressed) { // NOPMD
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( // NOPMD.PreserveStackTrace
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( // NOPMD.PreserveStackTrace
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( // NOPMD.PreserveStackTrace
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     // FIXME support more than one buffer for scatter-gather access
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     // FIXME support more than one buffer for scatter-gather access
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( // NOPMD.PreserveStackTrace
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    * Opens a socket channel. The {@code family} parameter specifies the {@link ProtocolFamily
366    * protocol family} of the channel's socket.
367    * <p>
368    * If the {@link ProtocolFamily} is of an {@link AFProtocolFamily}, or {@code UNIX}, the
369    * corresponding junixsocket implementation is used. In all other cases, the call is delegated to
370    * {@link SocketChannel#open()}.
371    *
372    * @param family The protocol family.
373    * @return The new {@link SocketChannel}.
374    * @throws IOException on error.
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 }