DatagramConnectedWriteExceptionTest Class — netty Architecture

Source Code

testsuite/src/main/java/io/netty/testsuite/transport/socket/DatagramConnectedWriteExceptionTest.java lines 46–128

public class DatagramConnectedWriteExceptionTest extends AbstractClientSocketTest {

    @Override
    protected List<TestsuitePermutation.BootstrapFactory<Bootstrap>> newFactories() {
        return SocketTestPermutation.INSTANCE.datagramSocket();
    }

    @Test
    @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
    @DisabledOnOs(OS.WINDOWS)
    public void testWriteThrowsPortUnreachableException(TestInfo testInfo) throws Throwable {
        run(testInfo, (Runner<Bootstrap>) this::testWriteExceptionAfterServerStop);
    }

    protected void testWriteExceptionAfterServerStop(Bootstrap clientBootstrap) throws Throwable {
        CountDownLatch serverReceivedLatch = new CountDownLatch(1);
        Bootstrap serverBootstrap = clientBootstrap.clone()
                .option(ChannelOption.SO_BROADCAST, false)
                .handler(new SimpleChannelInboundHandler<DatagramPacket>() {

                    @Override
                    protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) {
                        serverReceivedLatch.countDown();
                    }
                });

        Channel serverChannel = serverBootstrap.bind(new InetSocketAddress(NetUtil.LOCALHOST, 0)).sync().channel();
        InetSocketAddress serverAddress = (InetSocketAddress) serverChannel.localAddress();

        clientBootstrap.option(ChannelOption.AUTO_READ, false)
                .handler(new SimpleChannelInboundHandler<DatagramPacket>() {

                    @Override
                    protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) {
                        // no-op
                    }
                });

        Channel clientChannel = clientBootstrap.connect(serverAddress).sync().channel();

        CountDownLatch clientFirstSendLatch = new CountDownLatch(1);
        try {
            ByteBuf firstMessage = Unpooled.wrappedBuffer("First message".getBytes(CharsetUtil.UTF_8));
            clientChannel.writeAndFlush(firstMessage)
                    .addListener(future -> {
                        if (future.isSuccess()) {
                            clientFirstSendLatch.countDown();
                        }
                    });

            assertTrue(serverReceivedLatch.await(5, TimeUnit.SECONDS), "Server should receive first message");
            assertTrue(clientFirstSendLatch.await(5, TimeUnit.SECONDS), "Client should send first message");

            serverChannel.close().sync();

            AtomicReference<Throwable> writeException = new AtomicReference<>();
            CountDownLatch writesCompleteLatch = new CountDownLatch(10);

            for (int i = 0; i < 10; i++) {
                ByteBuf message = Unpooled.wrappedBuffer(("Message " + i).getBytes(CharsetUtil.UTF_8));
                clientChannel.writeAndFlush(message)
                        .addListener(future -> {
                            if (!future.isSuccess()) {
                                writeException.compareAndSet(null, future.cause());
                            }
                            writesCompleteLatch.countDown();
                        });
                Thread.sleep(50);
            }

            assertTrue(writesCompleteLatch.await(5, TimeUnit.SECONDS), "All writes should complete");

            assertNotNull(writeException.get(), "Should have captured a write exception");

            assertInstanceOf(PortUnreachableException.class, writeException.get(), "Expected " +
                    "PortUnreachableException but got: " + writeException.get().getClass().getName());
        } finally {
            if (clientChannel != null) {
                clientChannel.close().sync();
            }
        }