Split the EchoServer implementation

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

@@ -53,6 +53,11 @@ public static SocketChannel newUnixDomainSocket(final String pathName, final boo
     return result;
   }
 
+  /** Utility function to read until newline from a blocking channel. */
+  public static String readLine(SocketChannel channel) throws IOException {
+    return readLine(channel, 0);
+  }
+
   /** Utility function to read until newline from a non-blocking channel. */
   public static String readLine(SocketChannel channel, int readTimeoutMilis) throws IOException {
     int readBytes;
@@ -65,8 +70,10 @@ public static String readLine(SocketChannel channel, int readTimeoutMilis) throw
         buf.rewind();
         int numOfKeys = -1;
         if (isJava17Plus() && !ServerSocketChannels.isWin) {
-          channel.register(sel, SelectionKey.OP_READ);
-          numOfKeys = sel.select(readTimeoutMilis);
+          if (!channel.isBlocking()) {
+            channel.register(sel, SelectionKey.OP_READ);
+            numOfKeys = sel.select(readTimeoutMilis);
+          }
         } else {
           if (readTimeoutMilis > 0) {
             throw new IOException("timeout is unsupported on JDK 8");
@@ -96,6 +103,11 @@ public static String readLine(SocketChannel channel, int readTimeoutMilis) throw
     return new String(buf2.array(), StandardCharsets.UTF_8).replace("\n", "").replace("\r", "");
   }
 
+  /** Utility function to read all buffer from a blocking channel. */
+  public static ByteBuffer readAll(SocketChannel channel) throws IOException {
+    return readAll(channel, 0);
+  }
+
   /** Utility function to read all buffer from a non-blocking channel. */
   public static ByteBuffer readAll(SocketChannel channel, int readTimeoutMilis) throws IOException {
     int readBytes;
@@ -107,8 +119,10 @@ public static ByteBuffer readAll(SocketChannel channel, int readTimeoutMilis) th
         buf.rewind();
         int numOfKeys = -1;
         if (isJava17Plus() && !ServerSocketChannels.isWin) {
-          channel.register(sel, SelectionKey.OP_READ);
-          numOfKeys = sel.select(readTimeoutMilis);
+          if (!channel.isBlocking()) {
+            channel.register(sel, SelectionKey.OP_READ);
+            numOfKeys = sel.select(readTimeoutMilis);
+          }
         } else {
           if (readTimeoutMilis > 0) {
             throw new IOException("timeout is unsupported on JDK 8");

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

@@ -0,0 +1,36 @@
+package org.scalasbt.ipcsocket;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.util.concurrent.CompletableFuture;
+
+public class BlockingEchoServer {
+  private final ServerSocketChannel serverSocketChannel;
+
+  public BlockingEchoServer(ServerSocketChannel serverSocketChannel) {
+    this.serverSocketChannel = serverSocketChannel;
+  }
+
+  public void run() throws IOException {
+    while (true) {
+      SocketChannel clientChannel = serverSocketChannel.accept();
+      System.out.println("accepted: " + clientChannel.toString());
+      CompletableFuture.supplyAsync(
+          () -> {
+            try {
+              clientChannel.configureBlocking(true);
+              ByteBuffer inBytes = SocketChannels.readAll(clientChannel);
+              final String line =
+                  new String(inBytes.array(), "UTF-8").replace("\n", "").replace("\r", "");
+              System.out.println("server: " + line);
+              clientChannel.write(inBytes);
+            } catch (IOException e) {
+              e.printStackTrace();
+            }
+            return true;
+          });
+    }
+  }
+}

src/test/java/org/scalasbt/ipcsocket/EchoServer.java → src/test/java/org/scalasbt/ipcsocket/NonBlockingEchoServer.java

@@ -1,22 +1,17 @@
 package org.scalasbt.ipcsocket;
 
-import java.io.File;
 import java.io.IOException;
 import java.net.SocketTimeoutException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ServerSocketChannel;
 import java.nio.channels.SocketChannel;
 import java.util.concurrent.CompletableFuture;
 
-public class EchoServer {
-  static boolean isJava17Plus() {
-    return SocketChannels.isJava17Plus();
-  }
-
+public class NonBlockingEchoServer {
   private final ServerSocketChannel serverSocketChannel;
   private final int READ_TIMEOUT_MILI = 5000;
 
-  public EchoServer(ServerSocketChannel serverSocketChannel) {
+  public NonBlockingEchoServer(ServerSocketChannel serverSocketChannel) {
     this.serverSocketChannel = serverSocketChannel;
   }
 
@@ -29,15 +24,14 @@ public void run() throws IOException {
             try {
               clientChannel.configureBlocking(false);
               try {
-                ByteBuffer inBytes =
-                    SocketChannels.readAll(
-                        clientChannel,
-                        isJava17Plus() && !ServerSocketChannels.isWin ? READ_TIMEOUT_MILI : 0);
+                ByteBuffer inBytes = SocketChannels.readAll(clientChannel, READ_TIMEOUT_MILI);
                 final String line =
                     new String(inBytes.array(), "UTF-8").replace("\n", "").replace("\r", "");
                 System.out.println("server: " + line);
                 clientChannel.write(inBytes);
               } catch (SocketTimeoutException e) {
+                // if readAll doesn't complete in READ_TIMEOUT_MILI,
+                // SocketTimeoutException is thrown
                 System.out.println("server read timeout");
                 return false;
               }

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

@@ -14,36 +14,53 @@ static boolean isJava17Plus() {
     return SocketChannels.isJava17Plus();
   }
 
+  /** Test the non-blocking echo server using JDK 17 Unix Domain Socket. */
   @Test
-  public void testEchoServer() throws IOException, InterruptedException {
-    System.out.println("SocketChannelTest#testEchoServer(" + Boolean.toString(useJNI()) + ")");
+  public void testNonBlockingEchoServer() throws IOException, InterruptedException {
+    System.out.println(
+        "SocketChannelTest#testNonBlockingEchoServer(" + Boolean.toString(useJNI()) + ")");
     withSocket(
         sock -> {
-          String line = echoServerTest(sock, ServerSocketChannels.isWin ? 0 : 100);
-          assertEquals("echo did not return the content", "hello", line);
+          if (isJava17Plus() && !ServerSocketChannels.isWin) {
+            String line = nonBlockingEchoServerTest(sock, 100);
+            assertEquals("echo did not return the content", "hello", line);
+          }
         });
   }
 
+  /** Test the non-blocking echo server using JDK 17 Unix Domain Socket. */
   @Test
   public void testTimeout() throws IOException, InterruptedException {
     System.out.println("SocketChannelTest#testTimeout(" + Boolean.toString(useJNI()) + ")");
     withSocket(
         sock -> {
           if (isJava17Plus() && !ServerSocketChannels.isWin) {
-            String line = echoServerTest(sock, 6000);
+            String line = nonBlockingEchoServerTest(sock, 6000);
             assertEquals("echo did not timeout", "<timeout>", line);
           }
         });
   }
 
-  private String echoServerTest(String sock, int sleepBeforeSend)
+  /** Test the blocking echo server. */
+  @Test
+  public void testBlockingEchoServer() throws IOException, InterruptedException {
+    System.out.println(
+        "SocketChannelTest#testBlockingEchoServer(" + Boolean.toString(useJNI()) + ")");
+    withSocket(
+        sock -> {
+          String line = blockingEchoServerTest(sock);
+          assertEquals("echo did not return the content", "hello", line);
+        });
+  }
+
+  private String nonBlockingEchoServerTest(String sock, int sleepBeforeSend)
       throws IOException, InterruptedException {
     ServerSocketChannel serverSocket = ServerSocketChannels.newServerSocketChannel(sock, useJNI());
     CompletableFuture<Boolean> server =
         CompletableFuture.supplyAsync(
             () -> {
               try {
-                EchoServer echo = new EchoServer(serverSocket);
+                NonBlockingEchoServer echo = new NonBlockingEchoServer(serverSocket);
                 echo.run();
               } catch (IOException e) {
                 // e.printStackTrace();
@@ -59,8 +76,7 @@ private String echoServerTest(String sock, int sleepBeforeSend)
     client.configureBlocking(false);
     String line;
     try {
-      line =
-          SocketChannels.readLine(client, isJava17Plus() && !ServerSocketChannels.isWin ? 500 : 0);
+      line = SocketChannels.readLine(client, 500);
     } catch (SocketTimeoutException e) {
       line = "<timeout>";
     }
@@ -69,4 +85,29 @@ private String echoServerTest(String sock, int sleepBeforeSend)
     serverSocket.close();
     return line;
   }
+
+  private String blockingEchoServerTest(String sock) throws IOException, InterruptedException {
+    ServerSocketChannel serverSocket = ServerSocketChannels.newServerSocketChannel(sock, useJNI());
+    CompletableFuture<Boolean> server =
+        CompletableFuture.supplyAsync(
+            () -> {
+              try {
+                BlockingEchoServer echo = new BlockingEchoServer(serverSocket);
+                echo.run();
+              } catch (IOException e) {
+                // e.printStackTrace();
+              }
+              return true;
+            });
+    Thread.sleep(100);
+    SocketChannel client = SocketChannels.newSocketChannel(sock.toString(), useJNI());
+    System.out.println("client: " + client.toString());
+    client.write(ByteBuffer.wrap("hello\n".getBytes("UTF-8")));
+    client.configureBlocking(false);
+    String line = SocketChannels.readLine(client);
+    client.close();
+    server.cancel(true);
+    serverSocket.close();
+    return line;
+  }
 }