Named Pipe connection behavior matching Unix Stream behavior

src/main/java/com/timgroup/statsd/NamedPipeClientChannel.java

@@ -1,42 +1,81 @@
 package com.timgroup.statsd;
 
-import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.RandomAccessFile;
 import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
 import java.nio.channels.FileChannel;
 
+/**
+ * A ClientChannel for Windows named pipes.
+ *
+ * <p>The pipe is opened on the first write, not when the client is built, so the client can start
+ * before the server creates the pipe. After a failed write the pipe is closed and the next write
+ * opens it again, so the client recovers when the server restarts. Once {@link #close()} is called,
+ * the channel stays closed and writes throw {@link ClosedChannelException}.
+ */
 class NamedPipeClientChannel implements ClientChannel {
-    private final RandomAccessFile randomAccessFile;
-    private final FileChannel fileChannel;
     private final String pipe;
+    private RandomAccessFile randomAccessFile;
+    private FileChannel fileChannel;
+    private boolean closed;
 
     /**
      * Creates a new NamedPipeClientChannel with the given address.
      *
      * @param address Location of named pipe
-     * @throws FileNotFoundException if pipe does not exist
      */
-    NamedPipeClientChannel(NamedPipeSocketAddress address) throws FileNotFoundException {
+    NamedPipeClientChannel(NamedPipeSocketAddress address) {
         pipe = address.getPipe();
-        randomAccessFile = new RandomAccessFile(pipe, "rw");
-        fileChannel = randomAccessFile.getChannel();
     }
 
     @Override
-    public boolean isOpen() {
-        return fileChannel.isOpen();
+    public synchronized boolean isOpen() {
+        return !closed;
     }
 
     @Override
-    public int write(ByteBuffer src) throws IOException {
-        return fileChannel.write(src);
+    public synchronized int write(ByteBuffer src) throws IOException {
+        if (closed) {
+            throw new ClosedChannelException();
+        }
+        connectIfNeeded();
+        try {
+            return fileChannel.write(src);
+        } catch (IOException e) {
+            // The pipe is broken, for example because the server restarted. Reopen on next write.
+            try {
+                disconnect();
+            } catch (IOException ignored) {
+                // Report the write error, not the close error.
+            }
+            throw e;
+        }
+    }
+
+    private void connectIfNeeded() throws IOException {
+        if (fileChannel == null) {
+            randomAccessFile = new RandomAccessFile(pipe, "rw");
+            fileChannel = randomAccessFile.getChannel();
+        }
+    }
+
+    private void disconnect() throws IOException {
+        if (randomAccessFile != null) {
+            try {
+                // closing the file also closes the channel
+                randomAccessFile.close();
+            } finally {
+                randomAccessFile = null;
+                fileChannel = null;
+            }
+        }
     }
 
     @Override
-    public void close() throws IOException {
-        // closing the file also closes the channel
-        randomAccessFile.close();
+    public synchronized void close() throws IOException {
+        closed = true;
+        disconnect();
     }
 
     @Override

src/test/java/com/timgroup/statsd/DummyStatsDServer.java

@@ -24,34 +24,42 @@ protected void listen() {
                         new Runnable() {
                             @Override
                             public void run() {
-                                final ByteBuffer packet =
-                                        ByteBuffer.allocate(DEFAULT_UDS_MAX_PACKET_SIZE_BYTES);
-
-                                while (isOpen() && !Thread.interrupted()) {
-                                    if (freeze) {
-                                        try {
-                                            Thread.sleep(10);
-                                        } catch (InterruptedException e) {
-                                        }
-                                    } else {
-                                        try {
-                                            ((Buffer) packet)
-                                                    .clear(); // Cast necessary to handle Java9
-                                            // covariant return types
-                                            // see: https://jira.mongodb.org/browse/JAVA-2559 for
-                                            // ref.
-                                            receive(packet);
-                                            handlePacket(packet);
-                                        } catch (IOException e) {
-                                        }
-                                    }
+                                try {
+                                    readUntilClosed();
+                                } finally {
+                                    onListenStopped();
                                 }
                             }
                         });
         thread.setDaemon(true);
         thread.start();
     }
 
+    private void readUntilClosed() {
+        final ByteBuffer packet = ByteBuffer.allocate(DEFAULT_UDS_MAX_PACKET_SIZE_BYTES);
+
+        while (isOpen() && !Thread.interrupted()) {
+            if (freeze) {
+                try {
+                    Thread.sleep(10);
+                } catch (InterruptedException e) {
+                }
+            } else {
+                try {
+                    // Cast necessary to handle Java9 covariant return types
+                    // see: https://jira.mongodb.org/browse/JAVA-2559 for ref.
+                    ((Buffer) packet).clear();
+                    receive(packet);
+                    handlePacket(packet);
+                } catch (IOException e) {
+                }
+            }
+        }
+    }
+
+    /** Called on the listening thread after it stops reading. */
+    protected void onListenStopped() {}
+
     protected boolean sleepIfFrozen() {
         if (freeze) {
             try {

src/test/java/com/timgroup/statsd/NamedPipeClientChannelTest.java

@@ -0,0 +1,34 @@
+package com.timgroup.statsd;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
+import org.junit.Test;
+import org.junit.function.ThrowingRunnable;
+
+public class NamedPipeClientChannelTest {
+
+    @Test
+    public void write_after_close_throws_without_opening_the_pipe() throws Exception {
+        // The pipe doesn't exist, so a write that tried to open it would throw
+        // FileNotFoundException instead.
+        final NamedPipeClientChannel channel =
+                new NamedPipeClientChannel(new NamedPipeSocketAddress("missing-dir/pipe"));
+        assertTrue(channel.isOpen());
+
+        channel.close();
+
+        assertFalse(channel.isOpen());
+        assertThrows(
+                ClosedChannelException.class,
+                new ThrowingRunnable() {
+                    @Override
+                    public void run() throws Exception {
+                        channel.write(ByteBuffer.wrap(new byte[] {1}));
+                    }
+                });
+    }
+}

src/test/java/com/timgroup/statsd/NamedPipeDummyStatsDServer.java

@@ -79,9 +79,21 @@ protected void receive(ByteBuffer packet) throws IOException {
         packet.position(bytesRead.getValue());
     }
 
+    /**
+     * Stops the server. The listening thread closes the pipe once its current read returns, so a
+     * client that is still connected must send something before the pipe closes.
+     *
+     * <p>The handle isn't closed here because the pipe uses synchronous I/O, so {@code CloseHandle}
+     * from this thread can wait for the listening thread's pending {@code ReadFile}, which returns
+     * only when the client writes or disconnects.
+     */
     @Override
     public void close() throws IOException {
         isOpen = false;
+    }
+
+    @Override
+    protected void onListenStopped() {
         Kernel32.INSTANCE.CloseHandle(hNamedPipe);
     }
 }

src/test/java/com/timgroup/statsd/NamedPipeTest.java

@@ -73,4 +73,68 @@ public void sends_to_statsd() {
         }
         assertThat(lastException.getMessage(), nullValue());
     }
+
+    @Test(timeout = 5000L)
+    public void sends_to_pipe_created_after_build() throws Exception {
+        String pipeName = "testPipe-late-" + random.nextInt(10000);
+        NonBlockingStatsDClient lateClient = newClient(pipeName, new RecordingErrorHandler());
+        DummyStatsDServer lateServer = null;
+        try {
+            lateServer = new NamedPipeDummyStatsDServer(pipeName);
+            lateClient.gauge("late", 1);
+            lateServer.waitForMessage();
+            assertThat(lateServer.messagesReceived(), contains("my.prefix.late:1|g"));
+        } finally {
+            lateClient.stop();
+            if (lateServer != null) {
+                lateServer.close();
+            }
+        }
+    }
+
+    @Test(timeout = 10000L)
+    public void reconnects_after_server_restart() throws Exception {
+        String pipeName = "testPipe-restart-" + random.nextInt(10000);
+        RecordingErrorHandler errorHandler = new RecordingErrorHandler();
+        DummyStatsDServer firstServer = new NamedPipeDummyStatsDServer(pipeName);
+        NonBlockingStatsDClient restartClient = newClient(pipeName, errorHandler);
+        DummyStatsDServer secondServer = null;
+        try {
+            restartClient.gauge("before", 1);
+            firstServer.waitForMessage();
+            firstServer.close();
+
+            // The server closes its pipe after its next read, so send until the client sees the
+            // broken pipe and closes its end.
+            while (errorHandler.getExceptions().isEmpty()) {
+                restartClient.gauge("during", 1);
+                Thread.sleep(10);
+            }
+
+            secondServer = new NamedPipeDummyStatsDServer(pipeName);
+            // The queue holds one message, so keep sending until one arrives.
+            while (!secondServer.messagesReceived().contains("my.prefix.after:1|g")) {
+                restartClient.gauge("after", 1);
+                Thread.sleep(10);
+            }
+        } finally {
+            restartClient.stop();
+            firstServer.close();
+            if (secondServer != null) {
+                secondServer.close();
+            }
+        }
+    }
+
+    private static NonBlockingStatsDClient newClient(
+            String pipeName, StatsDClientErrorHandler errorHandler) {
+        return new NonBlockingStatsDClientBuilder()
+                .withEnvironmentVariables(new HashMap<String, String>())
+                .prefix("my.prefix")
+                .namedPipe(pipeName)
+                .queueSize(1)
+                .enableAggregation(false)
+                .errorHandler(errorHandler)
+                .build();
+    }
 }

src/test/java/com/timgroup/statsd/NonBlockingStatsDClientBuilderTest.java

@@ -5,13 +5,41 @@
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
+import java.io.FileNotFoundException;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Random;
 import org.junit.Test;
 import org.junit.function.ThrowingRunnable;
 
 public class NonBlockingStatsDClientBuilderTest {
 
+    @Test(timeout = 5000L)
+    public void missing_named_pipe_does_not_fail_build() throws Exception {
+        // A pipe inside a missing directory fails to open on every platform without creating a
+        // file.
+        final Map<String, String> env = new HashMap<>();
+        env.put(
+                NonBlockingStatsDClient.DD_NAMED_PIPE_ENV_VAR,
+                "missing-dir-" + new Random().nextInt(1000000) + "/pipe");
+        final RecordingErrorHandler errorHandler = new RecordingErrorHandler();
+        final NonBlockingStatsDClient client =
+                new NonBlockingStatsDClientBuilder()
+                        .withEnvironmentVariables(env)
+                        .enableAggregation(false)
+                        .errorHandler(errorHandler)
+                        .build();
+        try {
+            client.gauge("value", 42);
+            while (errorHandler.getExceptions().isEmpty()) {
+                Thread.sleep(10);
+            }
+            assertTrue(errorHandler.getExceptions().get(0) instanceof FileNotFoundException);
+        } finally {
+            client.stop();
+        }
+    }
+
     @Test(timeout = 5000L)
     public void origin_detection_env_false() throws Exception {
         final Map<String, String> env = new HashMap<>();