SocketTextStreamFunction implements SourceFunction<String> { @Override public void cancel() { isRunning = false; Socket theSocket = this.currentSocket; if (theSocket != null) { IOUtils.closeSocket(theSocket); } } SocketTextStreamFunction(String hostname, int port, char delimiter, long maxNumRetries);  SocketTextStreamFunction(String hostname, int port, char delimiter, long maxNumRetries, long delayBetweenRetries); @Override void run(SourceContext<String> ctx); @Override void cancel();  }
@Test public void testSocketSourceSimpleOutput() throws Exception { ServerSocket server = new ServerSocket(0); Socket channel = null; try { SocketTextStreamFunction source = new SocketTextStreamFunction(LOCALHOST, server.getLocalPort(), '\n', 0); SocketSourceThread runner = new SocketSourceThread(source, "test1", "check"); runner.start(); channel = server.accept(); OutputStreamWriter writer = new OutputStreamWriter(channel.getOutputStream()); writer.write("test1\n"); writer.write("check\n"); writer.flush(); runner.waitForNumElements(2); runner.cancel(); runner.interrupt(); runner.waitUntilDone(); channel.close(); } finally { if (channel != null) { IOUtils.closeQuietly(channel); } IOUtils.closeQuietly(server); } }
@Test public void testSocketSourceOutputWithRetries() throws Exception { ServerSocket server = new ServerSocket(0); Socket channel = null; try { SocketTextStreamFunction source = new SocketTextStreamFunction(LOCALHOST, server.getLocalPort(), '\n', 10, 100); SocketSourceThread runner = new SocketSourceThread(source, "test1", "check"); runner.start(); channel = server.accept(); channel.close(); channel = server.accept(); OutputStreamWriter writer = new OutputStreamWriter(channel.getOutputStream()); writer.write("test1\n"); writer.close(); channel.close(); channel = server.accept(); channel.close(); channel = server.accept(); writer = new OutputStreamWriter(channel.getOutputStream()); writer.write("check\n"); writer.flush(); runner.waitForNumElements(2); runner.cancel(); runner.waitUntilDone(); } finally { if (channel != null) { IOUtils.closeQuietly(channel); } IOUtils.closeQuietly(server); } }
@Test public void testSocketSourceOutputInfiniteRetries() throws Exception { ServerSocket server = new ServerSocket(0); Socket channel = null; try { SocketTextStreamFunction source = new SocketTextStreamFunction(LOCALHOST, server.getLocalPort(), '\n', -1, 100); SocketSourceThread runner = new SocketSourceThread(source, "test1", "check"); runner.start(); channel = server.accept(); channel.close(); channel = server.accept(); OutputStreamWriter writer = new OutputStreamWriter(channel.getOutputStream()); writer.write("test1\n"); writer.close(); channel.close(); channel = server.accept(); channel.close(); channel = server.accept(); writer = new OutputStreamWriter(channel.getOutputStream()); writer.write("check\n"); writer.flush(); runner.waitForNumElements(2); runner.cancel(); runner.waitUntilDone(); } finally { if (channel != null) { IOUtils.closeQuietly(channel); } IOUtils.closeQuietly(server); } }
@Test public void testSocketSourceOutputAcrossRetries() throws Exception { ServerSocket server = new ServerSocket(0); Socket channel = null; try { SocketTextStreamFunction source = new SocketTextStreamFunction(LOCALHOST, server.getLocalPort(), '\n', 10, 100); SocketSourceThread runner = new SocketSourceThread(source, "test1", "check1", "check2"); runner.start(); channel = server.accept(); channel.close(); channel = server.accept(); OutputStreamWriter writer = new OutputStreamWriter(channel.getOutputStream()); writer.write("te"); writer.close(); channel.close(); channel = server.accept(); channel.close(); channel = server.accept(); writer = new OutputStreamWriter(channel.getOutputStream()); writer.write("st1\n"); writer.write("check1\n"); writer.write("check2\n"); writer.flush(); runner.waitForNumElements(2); runner.cancel(); runner.waitUntilDone(); } finally { if (channel != null) { IOUtils.closeQuietly(channel); } IOUtils.closeQuietly(server); } }