testMultipleBindDatagramChannel() — netty Function Reference

Source Code

transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollReuseAddrTest.java lines 142–194

    @Test
    @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
    @Disabled // TODO: Unignore after making it pass on centos6-1 and debian7-1
    public void testMultipleBindDatagramChannel() throws Exception {
        ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.ADVANCED);
        assumeTrue(versionEqOrGt(3, 9, 0));
        Bootstrap bootstrap = createBootstrap();
        bootstrap.option(EpollChannelOption.SO_REUSEPORT, true);
        final AtomicBoolean received1 = new AtomicBoolean();
        bootstrap.handler(new DatagramSocketTestHandler(received1));
        ChannelFuture future = bootstrap.bind().syncUninterruptibly();
        final InetSocketAddress address1 = (InetSocketAddress) future.channel().localAddress();

        final AtomicBoolean received2 = new AtomicBoolean();
        bootstrap.handler(new DatagramSocketTestHandler(received2));
        ChannelFuture future2 = bootstrap.bind(address1).syncUninterruptibly();
        final InetSocketAddress address2 = (InetSocketAddress) future2.channel().localAddress();

        assertEquals(address1, address2);
        final byte[] bytes = "data".getBytes();

        // fire up 16 Threads and send DatagramPackets to make sure we stress it enough to see DatagramPackets received
        // on both sockets.
        int count = 16;
        final CountDownLatch latch = new CountDownLatch(count);
        Runnable r = new Runnable() {
            @Override
            public void run() {
                try {
                    DatagramSocket socket = new DatagramSocket();
                    while (!received1.get() || !received2.get()) {
                        socket.send(new DatagramPacket(
                                bytes, 0, bytes.length, address1.getAddress(), address1.getPort()));
                    }
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                latch.countDown();
            }
        };

        ExecutorService executor = Executors.newFixedThreadPool(count);
        for (int i = 0 ; i < count; i++) {
            executor.execute(r);
        }
        latch.await();
        executor.shutdown();
        future.channel().close().syncUninterruptibly();
        future2.channel().close().syncUninterruptibly();
        assertTrue(received1.get());
        assertTrue(received2.get());
    }