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 diff --git a/src/test/java/com/timgroup/statsd/DummyStatsDServer.java b/src/test/java/com/timgroup/statsd/DummyStatsDServer.java index 5d87168e..e80d1fae 100644 --- a/src/test/java/com/timgroup/statsd/DummyStatsDServer.java +++ b/src/test/java/com/timgroup/statsd/DummyStatsDServer.java @@ -24,27 +24,10 @@ 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(); } } }); @@ -52,6 +35,31 @@ public void run() { 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 { diff --git a/src/test/java/com/timgroup/statsd/NamedPipeClientChannelTest.java b/src/test/java/com/timgroup/statsd/NamedPipeClientChannelTest.java new file mode 100644 index 00000000..83a909b8 --- /dev/null +++ b/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})); + } + }); + } +} diff --git a/src/test/java/com/timgroup/statsd/NamedPipeDummyStatsDServer.java b/src/test/java/com/timgroup/statsd/NamedPipeDummyStatsDServer.java index 19cf7dae..78307592 100644 --- a/src/test/java/com/timgroup/statsd/NamedPipeDummyStatsDServer.java +++ b/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. + * + *
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);
}
}
diff --git a/src/test/java/com/timgroup/statsd/NamedPipeTest.java b/src/test/java/com/timgroup/statsd/NamedPipeTest.java
index 65b0e1de..a1119162 100644
--- a/src/test/java/com/timgroup/statsd/NamedPipeTest.java
+++ b/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