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