fix #35 Test case for half closed connection

reactor · Mar 17, 2020 · a40c949 · a40c949
1 parent fee345c
commit a40c949
Showing 1 changed file with 49 additions and 0 deletions.
diff --git a/src/test/java/reactor/netty/tcp/TcpServerTests.java b/src/test/java/reactor/netty/tcp/TcpServerTests.java
@@ -44,6 +44,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelOption;
 import io.netty.channel.group.ChannelGroup;
 import io.netty.channel.group.DefaultChannelGroup;
 import io.netty.handler.codec.LineBasedFrameDecoder;
@@ -865,6 +866,54 @@ else if (newState == ConnectionObserver.State.DISCONNECTING) {
 		server.disposeNow();
 	}
 
+	@Test
+	@SuppressWarnings("FutureReturnValueIgnored")
+	public void testHalfClosedConnection() throws Exception {
+		DisposableServer server =
+				TcpServer.create()
+				         .port(0)
+				         .option(ChannelOption.ALLOW_HALF_CLOSURE, true)
+				         .wiretap(true)
+				         .handle((in, out) -> in.receive()
+				                                .asString()
+				                                .doOnNext(s -> {
+				                                    if (s.endsWith("257\n")) {
+				                                        out.sendString(Mono.just("END")
+				                                                           .delayElement(Duration.ofMillis(100)))
+				                                           .then()
+				                                           .subscribe();
+				                                    }
+				                                })
+				                                .then())
+				         .bindNow();
+
+		Connection conn =
+				TcpClient.create()
+				         .addressSupplier(server::address)
+				         .wiretap(true)
+				         .connectNow();
+
+		CountDownLatch latch = new CountDownLatch(1);
+		conn.inbound()
+		    .receive()
+		    .asString()
+		    .subscribe(s -> {
+		        if ("END".equals(s)) {
+		            latch.countDown();
+		        }
+		    });
+
+		conn.outbound()
+		    .sendString(Flux.range(1, 257).map(count -> count + "\n"))
+		    .then()
+		    .subscribe(null, null, () -> ((io.netty.channel.socket.SocketChannel) conn.channel()).shutdownOutput()); // FutureReturnValueIgnored
+
+		assertTrue(latch.await(30, TimeUnit.SECONDS));
+
+		conn.disposeNow();
+		server.disposeNow();
+	}
+
 	private static class SimpleClient extends Thread {
 		private final int port;
 		private final CountDownLatch latch;