1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.testsuite.transport.socket;
17
18 import io.netty.bootstrap.Bootstrap;
19 import io.netty.bootstrap.ServerBootstrap;
20 import io.netty.buffer.ByteBuf;
21 import io.netty.buffer.ByteBufAllocator;
22 import io.netty.buffer.ByteBufUtil;
23 import io.netty.buffer.Unpooled;
24 import io.netty.channel.Channel;
25 import io.netty.channel.ChannelConfig;
26 import io.netty.channel.ChannelFuture;
27 import io.netty.channel.ChannelFutureListener;
28 import io.netty.channel.ChannelHandlerContext;
29 import io.netty.channel.ChannelInboundHandlerAdapter;
30 import io.netty.channel.ChannelInitializer;
31 import io.netty.channel.ChannelOption;
32 import io.netty.channel.RecvByteBufAllocator;
33 import io.netty.channel.SimpleChannelInboundHandler;
34 import io.netty.channel.nio.NioEventLoopGroup;
35 import io.netty.channel.oio.OioEventLoopGroup;
36 import io.netty.channel.socket.ChannelInputShutdownEvent;
37 import io.netty.channel.socket.ChannelInputShutdownReadComplete;
38 import io.netty.channel.socket.ChannelOutputShutdownEvent;
39 import io.netty.channel.socket.DuplexChannel;
40 import io.netty.channel.socket.SocketChannel;
41 import io.netty.util.ReferenceCountUtil;
42 import io.netty.util.UncheckedBooleanSupplier;
43 import io.netty.util.internal.PlatformDependent;
44 import org.junit.jupiter.api.Test;
45 import org.junit.jupiter.api.TestInfo;
46 import org.junit.jupiter.api.Timeout;
47
48 import java.util.concurrent.CountDownLatch;
49 import java.util.concurrent.TimeUnit;
50 import java.util.concurrent.atomic.AtomicInteger;
51 import java.util.concurrent.atomic.AtomicReference;
52
53 import static java.util.concurrent.TimeUnit.MILLISECONDS;
54 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
55 import static org.junit.jupiter.api.Assertions.assertEquals;
56 import static org.junit.jupiter.api.Assertions.assertNull;
57 import static org.junit.jupiter.api.Assertions.assertTrue;
58 import static org.junit.jupiter.api.Assumptions.assumeFalse;
59
60 @Timeout(value = 20000, unit = MILLISECONDS)
61 public class SocketHalfClosedTest extends AbstractSocketTest {
62
63 protected int maxReadCompleteWithNoDataAfterInputShutdown() {
64 return 2;
65 }
66
67 @Test
68 public void testAllDataReadEventTriggeredAfterHalfClosure(TestInfo testInfo) throws Throwable {
69 run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
70 @Override
71 public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
72 if (bootstrap.config().group() instanceof OioEventLoopGroup) {
73 logger.debug("Ignoring test for incompatible OIO event system");
74 return;
75 } else if (bootstrap.config().group() instanceof NioEventLoopGroup) {
76 logger.debug("Ignoring test for incompatible NioHandler");
77 return;
78 }
79 allDataReadEventTriggeredAfterHalfClosure(serverBootstrap, bootstrap);
80 }
81 });
82 }
83
84 private void allDataReadEventTriggeredAfterHalfClosure(ServerBootstrap sb, Bootstrap cb) throws Throwable {
85 final int totalServerBytesWritten = 1;
86 final CountDownLatch clientReadAllDataLatch = new CountDownLatch(1);
87 final CountDownLatch clientHalfClosedLatch = new CountDownLatch(1);
88 final CountDownLatch clientHalfClosedAllBytesRead = new CountDownLatch(1);
89 final AtomicInteger clientReadCompletes = new AtomicInteger();
90 final AtomicInteger clientZeroDataReadCompletes = new AtomicInteger();
91 Channel serverChannel = null;
92 Channel clientChannel = null;
93 final AtomicReference<Channel> serverChildChannel = new AtomicReference<Channel>();
94 try {
95 cb.option(ChannelOption.ALLOW_HALF_CLOSURE, true)
96 .option(ChannelOption.AUTO_CLOSE, false)
97 .option(ChannelOption.AUTO_READ, false);
98
99 sb.option(ChannelOption.ALLOW_HALF_CLOSURE, true)
100 .option(ChannelOption.AUTO_CLOSE, false)
101 .childOption(ChannelOption.TCP_NODELAY, true);
102
103 sb.childHandler(new ChannelInitializer<Channel>() {
104 @Override
105 protected void initChannel(Channel ch) throws Exception {
106 serverChildChannel.set(ch);
107 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
108 @Override
109 public void channelActive(ChannelHandlerContext ctx) throws Exception {
110 ByteBuf buf = ctx.alloc().buffer(totalServerBytesWritten);
111 buf.writerIndex(buf.capacity());
112 ctx.writeAndFlush(buf);
113 }
114
115 @Override
116 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
117 ctx.close();
118 }
119 });
120 }
121 });
122
123
124 cb.handler(new ChannelInitializer<Channel>() {
125 @Override
126 protected void initChannel(final Channel ch) {
127 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
128 private int bytesRead;
129 private int bytesSinceReadComplete;
130
131 @Override
132 public void channelRead(ChannelHandlerContext ctx, Object msg) {
133 ByteBuf buf = (ByteBuf) msg;
134 bytesRead += buf.readableBytes();
135 bytesSinceReadComplete += buf.readableBytes();
136 buf.release();
137 }
138
139 @Override
140 public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
141 if (evt == ChannelInputShutdownEvent.INSTANCE) {
142 clientHalfClosedLatch.countDown();
143 } else if (evt == ChannelInputShutdownReadComplete.INSTANCE) {
144 clientHalfClosedAllBytesRead.countDown();
145 ctx.close();
146 }
147 }
148
149 @Override
150 public void channelReadComplete(ChannelHandlerContext ctx) {
151 if (bytesSinceReadComplete == 0) {
152 clientZeroDataReadCompletes.incrementAndGet();
153 } else {
154 bytesSinceReadComplete = 0;
155 }
156 clientReadCompletes.incrementAndGet();
157 if (bytesRead == totalServerBytesWritten) {
158
159
160 ch.eventLoop().execute(new Runnable() {
161 @Override
162 public void run() {
163 clientReadAllDataLatch.countDown();
164 }
165 });
166 } else {
167 ctx.read();
168 }
169 }
170
171 @Override
172 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
173 ctx.fireExceptionCaught(cause);
174 ctx.close();
175 }
176 });
177 ch.read();
178 }
179 });
180
181 serverChannel = sb.bind().sync().channel();
182 clientChannel = cb.connect(serverChannel.localAddress()).sync().channel();
183 clientChannel.read();
184
185 clientReadAllDataLatch.await();
186
187
188 ((DuplexChannel) serverChildChannel.get()).shutdownOutput();
189
190 clientHalfClosedLatch.await();
191 clientHalfClosedAllBytesRead.await();
192 } finally {
193 if (clientChannel != null) {
194 clientChannel.close().sync();
195 }
196 if (serverChannel != null) {
197 serverChannel.close().sync();
198 }
199 }
200 }
201
202 @Test
203 public void testHalfClosureReceiveDataOnFinalWait2StateWhenSoLingerSet(TestInfo testInfo) throws Throwable {
204 run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
205 @Override
206 public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
207 testHalfClosureReceiveDataOnFinalWait2StateWhenSoLingerSet(serverBootstrap, bootstrap);
208 }
209 });
210 }
211
212 private void testHalfClosureReceiveDataOnFinalWait2StateWhenSoLingerSet(ServerBootstrap sb, Bootstrap cb)
213 throws Throwable {
214 Channel serverChannel = null;
215 Channel clientChannel = null;
216
217 final CountDownLatch waitHalfClosureDone = new CountDownLatch(1);
218 try {
219 sb.childOption(ChannelOption.SO_LINGER, 1)
220 .childHandler(new ChannelInitializer<Channel>() {
221
222 @Override
223 protected void initChannel(Channel ch) throws Exception {
224 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
225
226 @Override
227 public void channelActive(final ChannelHandlerContext ctx) {
228 SocketChannel channel = (SocketChannel) ctx.channel();
229 channel.shutdownOutput();
230 }
231
232 @Override
233 public void channelRead(ChannelHandlerContext ctx, Object msg) {
234 ReferenceCountUtil.release(msg);
235 waitHalfClosureDone.countDown();
236 }
237 });
238 }
239 });
240
241 cb.option(ChannelOption.ALLOW_HALF_CLOSURE, true)
242 .handler(new ChannelInitializer<Channel>() {
243 @Override
244 protected void initChannel(Channel ch) throws Exception {
245 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
246
247 @Override
248 public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
249 if (ChannelInputShutdownEvent.INSTANCE == evt) {
250 ctx.writeAndFlush(ctx.alloc().buffer().writeZero(16));
251 }
252
253 if (ChannelInputShutdownReadComplete.INSTANCE == evt) {
254 ctx.close();
255 }
256 }
257 });
258 }
259 });
260
261 serverChannel = sb.bind().sync().channel();
262 clientChannel = cb.connect(serverChannel.localAddress()).sync().channel();
263 waitHalfClosureDone.await();
264 } finally {
265 if (clientChannel != null) {
266 clientChannel.close().sync();
267 }
268
269 if (serverChannel != null) {
270 serverChannel.close().sync();
271 }
272 }
273 }
274
275 @Test
276 @Timeout(value = 10000, unit = MILLISECONDS)
277 public void testHalfClosureOnlyOneEventWhenAutoRead(TestInfo testInfo) throws Throwable {
278 run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
279 @Override
280 public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
281 testHalfClosureOnlyOneEventWhenAutoRead(serverBootstrap, bootstrap);
282 }
283 });
284 }
285
286 public void testHalfClosureOnlyOneEventWhenAutoRead(ServerBootstrap sb, Bootstrap cb) throws Throwable {
287 Channel serverChannel = null;
288 try {
289 cb.option(ChannelOption.ALLOW_HALF_CLOSURE, true)
290 .option(ChannelOption.AUTO_READ, true);
291 sb.childHandler(new ChannelInitializer<Channel>() {
292 @Override
293 protected void initChannel(Channel ch) {
294 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
295 @Override
296 public void channelActive(ChannelHandlerContext ctx) {
297 ((DuplexChannel) ctx).shutdownOutput();
298 }
299
300 @Override
301 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
302 ctx.close();
303 }
304 });
305 }
306 });
307
308 final AtomicInteger shutdownEventReceivedCounter = new AtomicInteger();
309 final AtomicInteger shutdownReadCompleteEventReceivedCounter = new AtomicInteger();
310
311 cb.handler(new ChannelInitializer<Channel>() {
312 @Override
313 protected void initChannel(Channel ch) {
314 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
315
316 @Override
317 public void userEventTriggered(final ChannelHandlerContext ctx, Object evt) {
318 if (evt == ChannelInputShutdownEvent.INSTANCE) {
319 shutdownEventReceivedCounter.incrementAndGet();
320 } else if (evt == ChannelInputShutdownReadComplete.INSTANCE) {
321 shutdownReadCompleteEventReceivedCounter.incrementAndGet();
322 ctx.executor().schedule(new Runnable() {
323 @Override
324 public void run() {
325 ctx.close();
326 }
327 }, 100, MILLISECONDS);
328 }
329 }
330
331 @Override
332 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
333 ctx.close();
334 }
335 });
336 }
337 });
338
339 serverChannel = sb.bind().sync().channel();
340 Channel clientChannel = cb.connect(serverChannel.localAddress()).sync().channel();
341 clientChannel.closeFuture().await();
342 assertEquals(1, shutdownEventReceivedCounter.get());
343 assertEquals(1, shutdownReadCompleteEventReceivedCounter.get());
344 } finally {
345 if (serverChannel != null) {
346 serverChannel.close().sync();
347 }
348 }
349 }
350
351 @Test
352 public void testAllDataReadAfterHalfClosure(TestInfo testInfo) throws Throwable {
353 run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
354 @Override
355 public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
356 testAllDataReadAfterHalfClosure(serverBootstrap, bootstrap);
357 }
358 });
359 }
360
361 public void testAllDataReadAfterHalfClosure(ServerBootstrap sb, Bootstrap cb) throws Throwable {
362 testAllDataReadAfterHalfClosure(true, sb, cb);
363 testAllDataReadAfterHalfClosure(false, sb, cb);
364 }
365
366 private void testAllDataReadAfterHalfClosure(final boolean autoRead,
367 ServerBootstrap sb, Bootstrap cb) throws Throwable {
368 final int totalServerBytesWritten = 1024 * 16;
369 final int numReadsPerReadLoop = 2;
370 final CountDownLatch serverInitializedLatch = new CountDownLatch(1);
371 final CountDownLatch clientReadAllDataLatch = new CountDownLatch(1);
372 final CountDownLatch clientHalfClosedLatch = new CountDownLatch(1);
373 final AtomicInteger clientReadCompletes = new AtomicInteger();
374 final AtomicInteger clientZeroDataReadCompletes = new AtomicInteger();
375 Channel serverChannel = null;
376 Channel clientChannel = null;
377 try {
378 cb.option(ChannelOption.ALLOW_HALF_CLOSURE, true)
379 .option(ChannelOption.AUTO_READ, autoRead)
380 .option(ChannelOption.RCVBUF_ALLOCATOR, new TestNumReadsRecvByteBufAllocator(numReadsPerReadLoop));
381
382 sb.childHandler(new ChannelInitializer<Channel>() {
383 @Override
384 protected void initChannel(Channel ch) throws Exception {
385 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
386 @Override
387 public void channelActive(ChannelHandlerContext ctx) throws Exception {
388 ByteBuf buf = ctx.alloc().buffer(totalServerBytesWritten);
389 buf.writerIndex(buf.capacity());
390 ctx.writeAndFlush(buf).addListener(new ChannelFutureListener() {
391 @Override
392 public void operationComplete(ChannelFuture future) throws Exception {
393 ((DuplexChannel) future.channel()).shutdownOutput();
394 }
395 });
396 serverInitializedLatch.countDown();
397 }
398
399 @Override
400 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
401 ctx.close();
402 }
403 });
404 }
405 });
406
407 cb.handler(new ChannelInitializer<Channel>() {
408 @Override
409 protected void initChannel(Channel ch) throws Exception {
410 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
411 private int bytesRead;
412 private int bytesSinceReadComplete;
413
414 @Override
415 public void channelRead(ChannelHandlerContext ctx, Object msg) {
416 ByteBuf buf = (ByteBuf) msg;
417 bytesRead += buf.readableBytes();
418 bytesSinceReadComplete += buf.readableBytes();
419 buf.release();
420 }
421
422 @Override
423 public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
424 if (evt == ChannelInputShutdownEvent.INSTANCE) {
425 clientHalfClosedLatch.countDown();
426 } else if (evt == ChannelInputShutdownReadComplete.INSTANCE) {
427 ctx.close();
428 }
429 }
430
431 @Override
432 public void channelReadComplete(ChannelHandlerContext ctx) {
433 if (bytesSinceReadComplete == 0) {
434 clientZeroDataReadCompletes.incrementAndGet();
435 } else {
436 bytesSinceReadComplete = 0;
437 }
438 clientReadCompletes.incrementAndGet();
439 if (bytesRead == totalServerBytesWritten) {
440 clientReadAllDataLatch.countDown();
441 }
442 if (!autoRead) {
443 ctx.read();
444 }
445 }
446
447 @Override
448 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
449 ctx.close();
450 }
451 });
452 }
453 });
454
455 serverChannel = sb.bind().sync().channel();
456 clientChannel = cb.connect(serverChannel.localAddress()).sync().channel();
457 clientChannel.read();
458
459 serverInitializedLatch.await();
460 clientReadAllDataLatch.await();
461 clientHalfClosedLatch.await();
462
463
464
465 assertTrue(totalServerBytesWritten > clientReadCompletes.get(),
466 "too many read complete events: " + clientReadCompletes.get());
467 assertTrue(clientZeroDataReadCompletes.get() <= maxReadCompleteWithNoDataAfterInputShutdown(),
468 "too many readComplete with no data: " + clientZeroDataReadCompletes.get() + " readComplete: " +
469 clientReadCompletes.get());
470 } finally {
471 if (clientChannel != null) {
472 clientChannel.close().sync();
473 }
474 if (serverChannel != null) {
475 serverChannel.close().sync();
476 }
477 }
478 }
479
480 @Test
481 public void testAutoCloseFalseDoesShutdownOutput(TestInfo testInfo) throws Throwable {
482
483 assumeFalse(PlatformDependent.isWindows());
484 run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
485 @Override
486 public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
487 testAutoCloseFalseDoesShutdownOutput(serverBootstrap, bootstrap);
488 }
489 });
490 }
491
492 public void testAutoCloseFalseDoesShutdownOutput(ServerBootstrap sb, Bootstrap cb) throws Throwable {
493 testAutoCloseFalseDoesShutdownOutput(false, false, sb, cb);
494 testAutoCloseFalseDoesShutdownOutput(false, true, sb, cb);
495 testAutoCloseFalseDoesShutdownOutput(true, false, sb, cb);
496 testAutoCloseFalseDoesShutdownOutput(true, true, sb, cb);
497 }
498
499 private static void testAutoCloseFalseDoesShutdownOutput(boolean allowHalfClosed,
500 final boolean clientIsLeader,
501 ServerBootstrap sb,
502 Bootstrap cb) throws InterruptedException {
503 final int expectedBytes = 100;
504 final CountDownLatch serverReadExpectedLatch = new CountDownLatch(1);
505 final CountDownLatch doneLatch = new CountDownLatch(2);
506 final AtomicReference<Throwable> causeRef = new AtomicReference<Throwable>();
507 Channel serverChannel = null;
508 Channel clientChannel = null;
509 try {
510 cb.option(ChannelOption.ALLOW_HALF_CLOSURE, allowHalfClosed)
511 .option(ChannelOption.AUTO_CLOSE, false)
512 .option(ChannelOption.SO_LINGER, 0);
513 sb.childOption(ChannelOption.ALLOW_HALF_CLOSURE, allowHalfClosed)
514 .childOption(ChannelOption.AUTO_CLOSE, false)
515 .childOption(ChannelOption.SO_LINGER, 0);
516
517 final AutoCloseFalseLeader leaderHandler = new AutoCloseFalseLeader(expectedBytes,
518 serverReadExpectedLatch, doneLatch, causeRef);
519 final AutoCloseFalseFollower followerHandler = new AutoCloseFalseFollower(expectedBytes,
520 serverReadExpectedLatch, doneLatch, causeRef);
521 sb.childHandler(new ChannelInitializer<Channel>() {
522 @Override
523 protected void initChannel(Channel ch) throws Exception {
524 ch.pipeline().addLast(clientIsLeader ? followerHandler :leaderHandler);
525 }
526 });
527
528 cb.handler(new ChannelInitializer<Channel>() {
529 @Override
530 protected void initChannel(Channel ch) throws Exception {
531 ch.pipeline().addLast(clientIsLeader ? leaderHandler : followerHandler);
532 }
533 });
534
535 serverChannel = sb.bind().sync().channel();
536 clientChannel = cb.connect(serverChannel.localAddress()).sync().channel();
537
538 doneLatch.await();
539 assertNull(causeRef.get());
540 assertTrue(leaderHandler.seenOutputShutdown);
541 } finally {
542 if (clientChannel != null) {
543 clientChannel.close().sync();
544 }
545 if (serverChannel != null) {
546 serverChannel.close().sync();
547 }
548 }
549 }
550
551 private static final class AutoCloseFalseFollower extends SimpleChannelInboundHandler<ByteBuf> {
552 private final int expectedBytes;
553 private final CountDownLatch followerCloseLatch;
554 private final CountDownLatch doneLatch;
555 private final AtomicReference<Throwable> causeRef;
556 private int bytesRead;
557
558 AutoCloseFalseFollower(int expectedBytes, CountDownLatch followerCloseLatch, CountDownLatch doneLatch,
559 AtomicReference<Throwable> causeRef) {
560 this.expectedBytes = expectedBytes;
561 this.followerCloseLatch = followerCloseLatch;
562 this.doneLatch = doneLatch;
563 this.causeRef = causeRef;
564 }
565
566 @Override
567 public void channelInactive(ChannelHandlerContext ctx) {
568 checkPrematureClose();
569 }
570
571 @Override
572 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
573 ctx.close();
574 checkPrematureClose();
575 }
576
577 @Override
578 protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
579 bytesRead += msg.readableBytes();
580 if (bytesRead >= expectedBytes) {
581
582 ByteBuf buf = ctx.alloc().buffer(expectedBytes);
583 buf.writerIndex(buf.writerIndex() + expectedBytes);
584 ctx.writeAndFlush(buf).addListener(new ChannelFutureListener() {
585 @Override
586 public void operationComplete(ChannelFuture future) throws Exception {
587 future.channel().close().addListener(new ChannelFutureListener() {
588 @Override
589 public void operationComplete(final ChannelFuture future) throws Exception {
590
591
592
593
594
595 future.channel().eventLoop().schedule(new Runnable() {
596 @Override
597 public void run() {
598 followerCloseLatch.countDown();
599 }
600 }, 200, TimeUnit.MILLISECONDS);
601 }
602 });
603 }
604 });
605 }
606 }
607
608 private void checkPrematureClose() {
609 if (bytesRead < expectedBytes) {
610 causeRef.set(new IllegalStateException("follower premature close"));
611 doneLatch.countDown();
612 }
613 }
614 }
615
616 private static final class AutoCloseFalseLeader extends SimpleChannelInboundHandler<ByteBuf> {
617 private final int expectedBytes;
618 private final CountDownLatch followerCloseLatch;
619 private final CountDownLatch doneLatch;
620 private final AtomicReference<Throwable> causeRef;
621 private int bytesRead;
622 boolean seenOutputShutdown;
623
624 AutoCloseFalseLeader(int expectedBytes, CountDownLatch followerCloseLatch, CountDownLatch doneLatch,
625 AtomicReference<Throwable> causeRef) {
626 this.expectedBytes = expectedBytes;
627 this.followerCloseLatch = followerCloseLatch;
628 this.doneLatch = doneLatch;
629 this.causeRef = causeRef;
630 }
631
632 @Override
633 public void channelActive(ChannelHandlerContext ctx) throws Exception {
634 ByteBuf buf = ctx.alloc().buffer(expectedBytes);
635 buf.writerIndex(buf.writerIndex() + expectedBytes);
636 ctx.writeAndFlush(buf.retainedDuplicate());
637
638
639
640 followerCloseLatch.await();
641
642
643 ctx.writeAndFlush(buf).addListener(new ChannelFutureListener() {
644 @Override
645 public void operationComplete(ChannelFuture future) throws Exception {
646 if (future.cause() == null) {
647 causeRef.set(new IllegalStateException("second write should have failed!"));
648 doneLatch.countDown();
649 }
650 }
651 });
652 }
653
654 @Override
655 protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
656 bytesRead += msg.readableBytes();
657 if (bytesRead >= expectedBytes) {
658 doneLatch.countDown();
659 }
660 }
661
662 @Override
663 public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
664 if (evt instanceof ChannelOutputShutdownEvent) {
665 seenOutputShutdown = true;
666 doneLatch.countDown();
667 }
668 }
669
670 @Override
671 public void channelInactive(ChannelHandlerContext ctx) {
672 checkPrematureClose();
673 }
674
675 @Override
676 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
677 ctx.close();
678 checkPrematureClose();
679 }
680
681 private void checkPrematureClose() {
682 if (bytesRead < expectedBytes || !seenOutputShutdown) {
683 causeRef.set(new IllegalStateException("leader premature close"));
684 doneLatch.countDown();
685 }
686 }
687 }
688
689 @Test
690 public void testAllDataReadClosure(TestInfo testInfo) throws Throwable {
691 run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
692 @Override
693 public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
694 testAllDataReadClosure(serverBootstrap, bootstrap);
695 }
696 });
697 }
698
699 public void testAllDataReadClosure(ServerBootstrap sb, Bootstrap cb) throws Throwable {
700 testAllDataReadClosure(true, false, sb, cb);
701 testAllDataReadClosure(true, true, sb, cb);
702 testAllDataReadClosure(false, false, sb, cb);
703 testAllDataReadClosure(false, true, sb, cb);
704 }
705
706 private static void testAllDataReadClosure(final boolean autoRead, final boolean allowHalfClosed,
707 ServerBootstrap sb, Bootstrap cb) throws Throwable {
708 final int totalServerBytesWritten = 1024 * 16;
709 final int numReadsPerReadLoop = 2;
710 final CountDownLatch serverInitializedLatch = new CountDownLatch(1);
711 final CountDownLatch clientReadAllDataLatch = new CountDownLatch(1);
712 final CountDownLatch clientHalfClosedLatch = new CountDownLatch(1);
713 final AtomicInteger clientReadCompletes = new AtomicInteger();
714 Channel serverChannel = null;
715 Channel clientChannel = null;
716 try {
717 cb.option(ChannelOption.ALLOW_HALF_CLOSURE, allowHalfClosed)
718 .option(ChannelOption.AUTO_READ, autoRead)
719 .option(ChannelOption.RCVBUF_ALLOCATOR, new TestNumReadsRecvByteBufAllocator(numReadsPerReadLoop));
720
721 sb.childHandler(new ChannelInitializer<Channel>() {
722 @Override
723 protected void initChannel(Channel ch) throws Exception {
724 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
725 @Override
726 public void channelActive(ChannelHandlerContext ctx) throws Exception {
727 ByteBuf buf = ctx.alloc().buffer(totalServerBytesWritten);
728 buf.writerIndex(buf.capacity());
729 ctx.writeAndFlush(buf).addListener(ChannelFutureListener.CLOSE);
730 serverInitializedLatch.countDown();
731 }
732
733 @Override
734 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
735 ctx.close();
736 }
737 });
738 }
739 });
740
741 cb.handler(new ChannelInitializer<Channel>() {
742 @Override
743 protected void initChannel(Channel ch) throws Exception {
744 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
745 private int bytesRead;
746
747 @Override
748 public void channelRead(ChannelHandlerContext ctx, Object msg) {
749 ByteBuf buf = (ByteBuf) msg;
750 bytesRead += buf.readableBytes();
751 buf.release();
752 }
753
754 @Override
755 public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
756 if (evt == ChannelInputShutdownEvent.INSTANCE && allowHalfClosed) {
757 clientHalfClosedLatch.countDown();
758 } else if (evt == ChannelInputShutdownReadComplete.INSTANCE) {
759 ctx.close();
760 }
761 }
762
763 @Override
764 public void channelInactive(ChannelHandlerContext ctx) {
765 if (!allowHalfClosed) {
766 clientHalfClosedLatch.countDown();
767 }
768 }
769
770 @Override
771 public void channelReadComplete(ChannelHandlerContext ctx) {
772 clientReadCompletes.incrementAndGet();
773 if (bytesRead == totalServerBytesWritten) {
774 clientReadAllDataLatch.countDown();
775 }
776 if (!autoRead) {
777 ctx.read();
778 }
779 }
780
781 @Override
782 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
783 ctx.close();
784 }
785 });
786 }
787 });
788
789 serverChannel = sb.bind().sync().channel();
790 clientChannel = cb.connect(serverChannel.localAddress()).sync().channel();
791 clientChannel.read();
792
793 serverInitializedLatch.await();
794 clientReadAllDataLatch.await();
795 clientHalfClosedLatch.await();
796 assertTrue(totalServerBytesWritten / numReadsPerReadLoop + 10 > clientReadCompletes.get(),
797 "too many read complete events: " + clientReadCompletes.get());
798 } finally {
799 if (clientChannel != null) {
800 clientChannel.close().sync();
801 }
802 if (serverChannel != null) {
803 serverChannel.close().sync();
804 }
805 }
806 }
807
808 @Test
809 public void testShutdownInputAllowsOutboundWrites(TestInfo testInfo) throws Throwable {
810 run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
811 @Override
812 public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
813 testShutdownInputAllowsOutboundWrites(true, serverBootstrap, bootstrap);
814 testShutdownInputAllowsOutboundWrites(false, serverBootstrap, bootstrap);
815 }
816 });
817 }
818
819 private static void testShutdownInputAllowsOutboundWrites(final boolean clientInitiates, final ServerBootstrap sb,
820 final Bootstrap cb)
821 throws InterruptedException {
822 final byte[] expectedBytes = new byte[100];
823 PlatformDependent.threadLocalRandom().nextBytes(expectedBytes);
824 final CountDownLatch readDoneLatch = new CountDownLatch(1);
825
826 Channel serverChannel = null;
827 Channel clientChannel = null;
828 try {
829 final ShutdownInputWriter shutdownInputWriter = new ShutdownInputWriter(expectedBytes);
830 final ShutdownInputReader shutdownInputReader =
831 new ShutdownInputReader(expectedBytes.length, readDoneLatch);
832 sb.childOption(ChannelOption.ALLOW_HALF_CLOSURE, true);
833 cb.option(ChannelOption.ALLOW_HALF_CLOSURE, true);
834 sb.childHandler(new ChannelInitializer<Channel>() {
835 @Override
836 protected void initChannel(Channel ch) throws Exception {
837 ch.pipeline().addLast(clientInitiates? shutdownInputReader : shutdownInputWriter);
838 }
839 });
840 cb.handler(new ChannelInitializer<Channel>() {
841 @Override
842 protected void initChannel(Channel ch) throws Exception {
843 ch.pipeline().addLast(clientInitiates? shutdownInputWriter : shutdownInputReader);
844 }
845 });
846 serverChannel = sb.bind().sync().channel();
847 clientChannel = cb.connect(serverChannel.localAddress()).sync().channel();
848 assertTrue(readDoneLatch.await(5, TimeUnit.SECONDS),
849 "Timed out waiting for inbound data after input shutdown");
850 assertArrayEquals(expectedBytes, shutdownInputReader.receivedBytes());
851 } finally {
852 if (serverChannel != null) {
853 serverChannel.close().sync();
854 }
855 if (clientChannel != null) {
856 clientChannel.close().sync();
857 }
858 }
859 }
860
861 private static final class ShutdownInputReader extends SimpleChannelInboundHandler<ByteBuf> {
862 private final CountDownLatch readDoneLatch;
863 private final int expectedBytesCount;
864 private final ByteBuf receivedMsgs;
865 private byte[] actualBytes;
866
867 ShutdownInputReader(final int expectedBytesCount,
868 final CountDownLatch readDoneLatch) {
869 this.readDoneLatch = readDoneLatch;
870 this.expectedBytesCount = expectedBytesCount;
871 this.receivedMsgs = Unpooled.buffer(expectedBytesCount);
872 }
873
874 @Override
875 protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
876 receivedMsgs.writeBytes(msg);
877 if (receivedMsgs.readableBytes() == expectedBytesCount) {
878 actualBytes = ByteBufUtil.getBytes(receivedMsgs);
879 readDoneLatch.countDown();
880 }
881 }
882
883 @Override
884 public void handlerRemoved(ChannelHandlerContext ctx) {
885 receivedMsgs.release();
886 }
887
888 byte[] receivedBytes() {
889 return actualBytes;
890 }
891 }
892
893 private static final class ShutdownInputWriter extends ChannelInboundHandlerAdapter {
894 private final byte[] expectedBytes;
895
896 ShutdownInputWriter(final byte[] expectedBytes) {
897 this.expectedBytes = expectedBytes;
898 }
899
900 @Override
901 public void channelActive(final ChannelHandlerContext ctx) throws Exception {
902 ((DuplexChannel) ctx.channel()).shutdownInput()
903 .addListener(new ChannelFutureListener() {
904 @Override
905 public void operationComplete(ChannelFuture future) {
906 ctx.writeAndFlush(Unpooled.wrappedBuffer(expectedBytes));
907 }
908 });
909 }
910 }
911
912 @Test
913 public void testShutdownOutputAllowsInboundReads(TestInfo testInfo) throws Throwable {
914 run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
915 @Override
916 public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
917 testShutdownOutputAllowsInboundReads(true, serverBootstrap, bootstrap);
918 testShutdownOutputAllowsInboundReads(false, serverBootstrap, bootstrap);
919 }
920 });
921 }
922
923 private static void testShutdownOutputAllowsInboundReads(final boolean clientInitiates, final ServerBootstrap sb,
924 final Bootstrap cb) throws InterruptedException {
925 final byte[] expectedBytes = new byte[100];
926 PlatformDependent.threadLocalRandom().nextBytes(expectedBytes);
927 final CountDownLatch readDoneLatch = new CountDownLatch(1);
928
929 Channel serverChannel = null;
930 Channel clientChannel = null;
931 try {
932 final ShutdownOutputWriter shutdownOutputWriter = new ShutdownOutputWriter(expectedBytes);
933 final ShutdownOutputReader
934 shutdownOutputReader =
935 new ShutdownOutputReader(expectedBytes.length, readDoneLatch);
936 sb.childOption(ChannelOption.ALLOW_HALF_CLOSURE, true);
937 cb.option(ChannelOption.ALLOW_HALF_CLOSURE, true);
938 sb.childHandler(new ChannelInitializer<Channel>() {
939 @Override
940 protected void initChannel(Channel ch) throws Exception {
941 ch.pipeline().addLast(clientInitiates? shutdownOutputWriter : shutdownOutputReader);
942 }
943 });
944 cb.handler(new ChannelInitializer<Channel>() {
945
946 @Override
947 protected void initChannel(Channel ch) throws Exception {
948 ch.pipeline().addLast(clientInitiates? shutdownOutputReader : shutdownOutputWriter);
949 }
950 });
951
952 serverChannel = sb.bind().sync().channel();
953 clientChannel = cb.connect(serverChannel.localAddress()).sync().channel();
954 assertTrue(readDoneLatch.await(5, TimeUnit.SECONDS),
955 "Timed out waiting for inbound data after output shutdown");
956 assertArrayEquals(expectedBytes, shutdownOutputReader.receivedBytes());
957 } finally {
958 if (serverChannel != null) {
959 serverChannel.close().sync();
960 }
961 if (clientChannel != null) {
962 clientChannel.close().sync();
963 }
964 }
965 }
966
967 private static final class ShutdownOutputWriter extends ChannelInboundHandlerAdapter {
968 private final byte[] expectedBytes;
969
970 ShutdownOutputWriter(final byte[] expectedBytes) {
971 this.expectedBytes = expectedBytes;
972 }
973
974 @Override
975 public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
976 if (evt == ChannelInputShutdownEvent.INSTANCE) {
977 ctx.writeAndFlush(Unpooled.wrappedBuffer(expectedBytes));
978 }
979 ctx.fireUserEventTriggered(evt);
980 }
981 }
982
983 private static final class ShutdownOutputReader extends SimpleChannelInboundHandler<ByteBuf> {
984 private final CountDownLatch readDoneLatch;
985 private final int expectedBytesCount;
986 private final ByteBuf receivedMsgs;
987 private byte[] actualbytes;
988
989 ShutdownOutputReader(final int expectedBytesCount,
990 final CountDownLatch readDoneLatch) {
991 this.readDoneLatch = readDoneLatch;
992 this.expectedBytesCount = expectedBytesCount;
993 this.receivedMsgs = Unpooled.buffer(expectedBytesCount);
994 }
995
996 @Override
997 public void channelActive(ChannelHandlerContext ctx) {
998 ((DuplexChannel) ctx.channel()).shutdownOutput();
999 }
1000
1001 @Override
1002 protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
1003 receivedMsgs.writeBytes(msg);
1004 if (receivedMsgs.readableBytes() == expectedBytesCount) {
1005 this.actualbytes = ByteBufUtil.getBytes(receivedMsgs);
1006 readDoneLatch.countDown();
1007 }
1008 }
1009
1010 @Override
1011 public void handlerRemoved(ChannelHandlerContext ctx) {
1012 receivedMsgs.release();
1013 }
1014
1015 byte[] receivedBytes() {
1016 return actualbytes;
1017 }
1018 }
1019
1020
1021
1022
1023 private static final class TestNumReadsRecvByteBufAllocator implements RecvByteBufAllocator {
1024 private final int numReads;
1025 TestNumReadsRecvByteBufAllocator(int numReads) {
1026 this.numReads = numReads;
1027 }
1028
1029 @Override
1030 public ExtendedHandle newHandle() {
1031 return new ExtendedHandle() {
1032 private int attemptedBytesRead;
1033 private int lastBytesRead;
1034 private int numMessagesRead;
1035 @Override
1036 public ByteBuf allocate(ByteBufAllocator alloc) {
1037 return alloc.ioBuffer(guess(), guess());
1038 }
1039
1040 @Override
1041 public int guess() {
1042 return 1;
1043 }
1044
1045 @Override
1046 public void reset(ChannelConfig config) {
1047 numMessagesRead = 0;
1048 }
1049
1050 @Override
1051 public void incMessagesRead(int numMessages) {
1052 numMessagesRead += numMessages;
1053 }
1054
1055 @Override
1056 public void lastBytesRead(int bytes) {
1057 lastBytesRead = bytes;
1058 }
1059
1060 @Override
1061 public int lastBytesRead() {
1062 return lastBytesRead;
1063 }
1064
1065 @Override
1066 public void attemptedBytesRead(int bytes) {
1067 attemptedBytesRead = bytes;
1068 }
1069
1070 @Override
1071 public int attemptedBytesRead() {
1072 return attemptedBytesRead;
1073 }
1074
1075 @Override
1076 public boolean continueReading() {
1077 return numMessagesRead < numReads;
1078 }
1079
1080 @Override
1081 public boolean continueReading(UncheckedBooleanSupplier maybeMoreDataSupplier) {
1082 return continueReading() && maybeMoreDataSupplier.get();
1083 }
1084
1085 @Override
1086 public void readComplete() {
1087
1088 }
1089 };
1090 }
1091 }
1092 }