Add duplex test

.github/workflows/ci.yml

@@ -1,7 +1,5 @@
 name: CI
-on:
-  pull_request:
-  push:
+on: [ pull_request, push ]
 
 jobs:
   linux-test:

.gitignore

@@ -1 +1,15 @@
 **/target
+
+# Eclipse/VsCode
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+
+# VsCode
+.vscode
+
+# Metals
+.metals
+

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

@@ -0,0 +1,49 @@
+package org.scalasbt.ipcsocket;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.Random;
+
+public abstract class BaseSocketSetup {
+
+  final boolean isWin = System.getProperty("os.name", "").toLowerCase().startsWith("win");
+
+  Random rand = new Random();
+
+  boolean useJNI() {
+    return false;
+  }
+
+  public static interface MayThrow {
+    void accept(String string) throws IOException, InterruptedException;
+  }
+
+  protected void withSocket(final MayThrow consumer) throws IOException, InterruptedException {
+    Path tempDir = isWin ? null : Files.createTempDirectory("ipcsocket");
+    Path socketPath = tempDir != null ? tempDir.resolve("foo" + rand.nextInt() + ".sock") : null;
+    String socket =
+        socketPath != null ? socketPath.toString() : "\\\\.\\pipe\\ipcsockettest" + rand.nextInt();
+    try {
+      consumer.accept(socket);
+    } finally {
+      if (socketPath != null) Files.deleteIfExists(socketPath);
+      if (tempDir != null) Files.deleteIfExists(socketPath);
+    }
+  }
+
+  protected ServerSocket newServerSocket(String socketName) throws IOException {
+    return isWin
+        ? new Win32NamedPipeServerSocket(socketName, useJNI(), Win32SecurityLevel.LOGON_DACL)
+        : new UnixDomainServerSocket(socketName, useJNI());
+  }
+
+  protected Socket newClientSocket(String socketName) throws IOException {
+    return isWin
+        ? new Win32NamedPipeSocket(socketName, useJNI())
+        : new UnixDomainSocket(socketName, useJNI());
+  }
+}

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

@@ -17,44 +17,8 @@
 import java.net.Socket;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Consumer;
-import java.util.Random;
 
-public class SocketTest {
-  boolean useJNI() {
-    return false;
-  }
-
-  final boolean isWin = System.getProperty("os.name", "").toLowerCase().startsWith("win");
-
-  interface MayThrow {
-    void accept(String string) throws IOException, InterruptedException;
-  }
-
-  final void withSocket(final MayThrow consumer) throws IOException, InterruptedException {
-    Random rand = new Random();
-    Path tempDir = isWin ? null : Files.createTempDirectory("ipcsocket");
-    Path socketPath = tempDir != null ? tempDir.resolve("foo" + rand.nextInt() + ".sock") : null;
-    String socket =
-        socketPath != null ? socketPath.toString() : "\\\\.\\pipe\\ipcsockettest" + rand.nextInt();
-    try {
-      consumer.accept(socket);
-    } finally {
-      if (socketPath != null) Files.deleteIfExists(socketPath);
-      if (tempDir != null) Files.deleteIfExists(socketPath);
-    }
-  }
-
-  ServerSocket newServerSocket(String socketName) throws IOException {
-    return isWin
-        ? new Win32NamedPipeServerSocket(socketName, useJNI(), Win32SecurityLevel.LOGON_DACL)
-        : new UnixDomainServerSocket(socketName, useJNI());
-  }
-
-  Socket newClientSocket(String socketName) throws IOException {
-    return isWin
-        ? new Win32NamedPipeSocket(socketName, useJNI())
-        : new UnixDomainSocket(socketName, useJNI());
-  }
+public class SocketTest extends BaseSocketSetup {
 
   @Test
   public void testAssertEquals() throws IOException, InterruptedException {

src/test/java/org/scalasbt/ipcsocket/duplex/DuplexClient.java

@@ -0,0 +1,37 @@
+package org.scalasbt.ipcsocket.duplex;
+
+import java.io.IOException;
+import java.net.Socket;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ExecutorService;
+import org.scalasbt.ipcsocket.BaseSocketSetup;
+
+public class DuplexClient extends BaseSocketSetup {
+
+  private final String pipeName;
+  private final int sendMessages;
+
+  public Sender sender;
+  public Receiver receiver;
+
+  public DuplexClient(String pipeName, int sendMessages) {
+    this.pipeName = pipeName;
+    this.sendMessages = sendMessages;
+  }
+
+  public void startAndAwait() {
+    ExecutorService pool = Executors.newFixedThreadPool(2);
+
+    try (Socket socket = newClientSocket(pipeName)) {
+      sender = new Sender("client", socket, sendMessages);
+      receiver = new Receiver("client", socket);
+      pool.execute(sender);
+      pool.execute(receiver);
+
+      Thread.sleep(sendMessages * 1000);
+      pool.shutdownNow();
+    } catch (InterruptedException | IOException e) {
+      e.printStackTrace();
+    }
+  }
+}

src/test/java/org/scalasbt/ipcsocket/duplex/DuplexServer.java

@@ -0,0 +1,41 @@
+package org.scalasbt.ipcsocket.duplex;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ExecutorService;
+import org.scalasbt.ipcsocket.BaseSocketSetup;
+
+public class DuplexServer extends BaseSocketSetup {
+
+  private final String pipeName;
+  private final int sendMessages;
+
+  public Sender sender;
+  public Receiver receiver;
+
+  public DuplexServer(String pipeName, int sendMessages) {
+    this.pipeName = pipeName;
+    this.sendMessages = sendMessages;
+  }
+
+  public void startAndAwait() {
+    ExecutorService pool = Executors.newFixedThreadPool(10);
+
+    System.out.println("DuplexServer started. Waiting for client...");
+    try (ServerSocket serverSocket = newServerSocket(pipeName);
+        Socket socket = serverSocket.accept()) {
+
+      sender = new Sender("server", socket, sendMessages);
+      receiver = new Receiver("server", socket);
+      pool.execute(sender);
+      pool.execute(receiver);
+
+      Thread.sleep((sendMessages + 2) * 1000);
+      pool.shutdownNow();
+    } catch (InterruptedException | IOException e) {
+      e.printStackTrace();
+    }
+  }
+}

src/test/java/org/scalasbt/ipcsocket/duplex/DuplexTest.java

@@ -0,0 +1,38 @@
+package org.scalasbt.ipcsocket.duplex;
+
+import static org.junit.Assert.assertEquals;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.junit.Test;
+import org.scalasbt.ipcsocket.BaseSocketSetup;
+
+public class DuplexTest extends BaseSocketSetup {
+
+  @Test
+  public void testDuplexCommunication() throws Exception {
+    withSocket(
+        socketName -> {
+          ExecutorService pool = Executors.newFixedThreadPool(2);
+
+          // start server
+          int serverSendMessages = 15;
+          DuplexServer server = new DuplexServer(socketName, serverSendMessages);
+          pool.execute(() -> server.startAndAwait());
+
+          // wait for pipe to be instantiated
+          Thread.sleep(2000);
+
+          // start client
+          int clientSendMessages = 7;
+          DuplexClient client = new DuplexClient(socketName, clientSendMessages);
+          pool.execute(() -> client.startAndAwait());
+
+          // wait client and server to terminate
+          Thread.sleep((Math.max(serverSendMessages, clientSendMessages) + 1) * 1000);
+          pool.shutdown();
+
+          assertEquals(serverSendMessages, client.receiver.receivedMessages);
+          assertEquals(clientSendMessages, server.receiver.receivedMessages);
+        });
+  }
+}

src/test/java/org/scalasbt/ipcsocket/duplex/Receiver.java

@@ -0,0 +1,34 @@
+package org.scalasbt.ipcsocket.duplex;
+
+import java.io.IOException;
+import java.net.Socket;
+import java.util.Random;
+import java.util.Scanner;
+
+public class Receiver implements Runnable {
+
+  private final String name;
+  private final Socket socket;
+  public volatile int receivedMessages;
+  private final Random random = new Random();
+
+  public Receiver(String name, Socket socket) {
+    this.name = name;
+    this.socket = socket;
+  }
+
+  @Override
+  public void run() {
+    try (Scanner in = new Scanner(socket.getInputStream())) {
+      while (true) {
+        while (in.hasNextLine()) {
+          System.out.println("[" + name + "] got a message: " + in.nextLine());
+          receivedMessages++;
+        }
+        Thread.sleep(Math.abs(random.nextInt(1000)));
+      }
+    } catch (IOException | InterruptedException e) {
+      e.printStackTrace();
+    }
+  }
+}

src/test/java/org/scalasbt/ipcsocket/duplex/Sender.java

@@ -0,0 +1,33 @@
+package org.scalasbt.ipcsocket.duplex;
+
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.net.Socket;
+import java.util.Random;
+
+public class Sender implements Runnable {
+
+  private final String name;
+  private final Socket socket;
+  private final int sendMessages;
+  private final Random random = new Random();
+
+  public Sender(String name, Socket socket, int sendMessages) {
+    this.name = name;
+    this.socket = socket;
+    this.sendMessages = sendMessages;
+  }
+
+  @Override
+  public void run() {
+    try (PrintWriter out = new PrintWriter(socket.getOutputStream(), true)) {
+      for (int i = 0; i < sendMessages; i++) {
+        System.out.println("[" + name + "] sending msg: " + i);
+        out.println("hello" + i);
+        Thread.sleep(Math.abs(random.nextInt(1000)));
+      }
+    } catch (IOException | InterruptedException e) {
+      e.printStackTrace();
+    }
+  }
+}