Native async

Author: eed3si9n

src/main/java/org/scalasbt/ipcsocket/JNIUnixDomainSocketLibraryProvider.java

@@ -46,10 +46,24 @@ public int shutdown(int fd, int how) throws NativeErrorException {
 
   public int available(int fd) throws NativeErrorException {
     int result = availableNative(fd);
-    if (result < 0) {
-      return returnOrThrow(pollReadNative(fd, 0), 0);
+    if (result >= 0) {
+      return result;
+    } else if (pollRead(fd, 0)) {
+      return 1;
     } else {
-      return returnOrThrow(result, 0);
+      return 0;
+    }
+  }
+
+  public boolean pollRead(int fd, int timeout) throws NativeErrorException {
+    int result = pollReadNative(fd, timeout);
+    if (result == 0) {
+      return false;
+    } else if (result > 0) {
+      return true;
+    } else {
+      returnOrThrow(result, 0);
+      return false;
     }
   }
 

src/main/java/org/scalasbt/ipcsocket/SocketChannels.java

@@ -99,14 +99,8 @@ public static String readLine(SocketChannel channel, int readTimeoutMillis) thro
             numOfKeys = sel.select(readTimeoutMillis);
           }
         } else {
-          if (readTimeoutMillis > 0) {
-            if (ServerSocketChannels.isWin) {
-              if (channel.supportedOptions().contains(SO_TIMEOUT)) {
-                channel.setOption(SO_TIMEOUT, Integer.valueOf(readTimeoutMillis));
-              }
-            } else {
-              throw new IOException("timeout requires JDK 17 or Windows");
-            }
+          if (channel.supportedOptions().contains(SO_TIMEOUT)) {
+            channel.setOption(SO_TIMEOUT, Integer.valueOf(readTimeoutMillis));
           }
         }
         if (numOfKeys == 0) {
@@ -153,16 +147,13 @@ public static ByteBuffer readAll(SocketChannel channel, int readTimeoutMillis)
           }
         } else {
           if (readTimeoutMillis > 0) {
-            // The following operation gets blocked on JDK 8
-            // channel.register(sel, SelectionKey.OP_READ);
-            // numOfKeys = sel.select(readTimeoutMillis);
-            if (ServerSocketChannels.isWin) {
-              if (channel.supportedOptions().contains(SO_TIMEOUT)) {
-                channel.setOption(SO_TIMEOUT, Integer.valueOf(readTimeoutMillis));
-              }
-            } else {
-              throw new IOException("timeout requires JDK 17 or Windows");
-            }
+            throw new IOException("timeout requires JDK 17 and non-Windows");
+          }
+          // The following operation gets blocked on JDK 8
+          // channel.register(sel, SelectionKey.OP_READ);
+          // numOfKeys = sel.select(readTimeoutMilis);
+          if (channel.supportedOptions().contains(SO_TIMEOUT)) {
+            channel.setOption(SO_TIMEOUT, Integer.valueOf(readTimeoutMillis));
           }
         }
         if (numOfKeys == 0) {

src/main/java/org/scalasbt/ipcsocket/UnixDomainSocket.java

@@ -24,6 +24,7 @@
 import java.nio.ByteBuffer;
 
 import java.net.Socket;
+import java.net.SocketTimeoutException;
 
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -137,7 +138,7 @@ public int available() throws IOException {
     public int read() throws IOException {
       byte[] buf = new byte[1];
       int result;
-      if (doRead(buf, 0, 1) == 0) {
+      if (doRead(buf, 0, 1, getSoTimeout()) == 0) {
         result = -1;
       } else {
         // Make sure to & with 0xFF to avoid sign extension
@@ -152,7 +153,7 @@ public int read(byte[] b, int off, int len) throws IOException {
       }
       int socketFd = fd.acquire();
       try {
-        int result = doRead(b, off, len);
+        int result = doRead(b, off, len, getSoTimeout());
         if (result == 0) {
           try {
             provider.close(socketFd);
@@ -168,12 +169,17 @@ public int read(byte[] b, int off, int len) throws IOException {
       }
     }
 
-    private int doRead(byte[] buf, int offset, int len) throws IOException {
+    private int doRead(byte[] buf, int offset, int len, int timeoutMillis) throws IOException {
       try {
         int fdToRead = fd.acquire();
         if (fdToRead == -1) {
           return -1;
         }
+        if (timeoutMillis > 0) {
+          if (!provider.pollRead(fdToRead, timeoutMillis)) {
+            throw new SocketTimeoutException("read timed out");
+          }
+        }
         return provider.read(fdToRead, buf, offset, len);
       } catch (NativeErrorException e) {
         throw new IOException(e);

src/main/java/org/scalasbt/ipcsocket/UnixDomainSocketLibrary.java

@@ -259,19 +259,28 @@ public int available(int fd) throws NativeErrorException {
       UnixDomainSocketLibrary.ioctl(fd, op, len);
       return len.getValue();
     } catch (final LastErrorException e) {
-      try {
-        UnixDomainSocketLibrary.PollfdUn pollfd =
-            new UnixDomainSocketLibrary.PollfdUn(
-                fd, (short) UnixDomainSocketLibrary.POLL_IN, (short) 0);
-        UnixDomainSocketLibrary.poll(pollfd, 1, 0);
-        if ((((int) pollfd.revents) & UnixDomainSocketLibrary.POLL_IN) != 0) {
-          return 1;
-        } else {
-          return 0;
-        }
-      } catch (final LastErrorException e2) {
-        throw new NativeErrorException(e2.getErrorCode(), e2.getMessage());
+      if (pollRead(fd, 0)) {
+        return 1;
+      } else {
+        return 0;
+      }
+    }
+  }
+
+  @Override
+  public boolean pollRead(int fd, int timeout) throws NativeErrorException {
+    try {
+      UnixDomainSocketLibrary.PollfdUn pollfd =
+          new UnixDomainSocketLibrary.PollfdUn(
+              fd, (short) UnixDomainSocketLibrary.POLL_IN, (short) 0);
+      UnixDomainSocketLibrary.poll(pollfd, 1, timeout);
+      if ((((int) pollfd.revents) & UnixDomainSocketLibrary.POLL_IN) != 0) {
+        return true;
+      } else {
+        return false;
       }
+    } catch (final LastErrorException e2) {
+      throw new NativeErrorException(e2.getErrorCode(), e2.getMessage());
     }
   }
 

src/main/java/org/scalasbt/ipcsocket/UnixDomainSocketLibraryProvider.java

@@ -21,6 +21,8 @@ public interface UnixDomainSocketLibraryProvider {
 
   int available(int fd) throws NativeErrorException;
 
+  boolean pollRead(int fd, int timeout) throws NativeErrorException;
+
   int maxSocketLength();
 
   static UnixDomainSocketLibraryProvider get(boolean useJNI) {

src/test/java/org/scalasbt/ipcsocket/SocketChannelTest.java

@@ -14,7 +14,7 @@ static boolean isJava17Plus() {
     return SocketChannels.isJava17Plus();
   }
 
-  /** Test the non-blocking echo server using JDK 17 Unix Domain Socket. */
+  /** Test the non-blocking echo server. */
   @Test
   public void testNonBlockingEchoServer() throws IOException, InterruptedException {
     System.out.println(
@@ -28,13 +28,13 @@ public void testNonBlockingEchoServer() throws IOException, InterruptedException
         });
   }
 
-  /** Test the non-blocking echo server using JDK 17 Unix Domain Socket. */
+  /** Test the non-blocking echo server. */
   @Test
   public void testTimeout() throws IOException, InterruptedException {
     System.out.println("SocketChannelTest#testTimeout(" + Boolean.toString(useJNI()) + ")");
     withSocket(
         sock -> {
-          if (isJava17Plus() || ServerSocketChannels.isWin) {
+          if (isJava17Plus() && !ServerSocketChannels.isWin) {
             String line = nonBlockingEchoServerTest(sock, 6000, 600);
             assertEquals("echo did not timeout", "<unavailable>", line);
           }