FixedChannelPoolTest Class — netty Architecture

Source Code

transport/src/test/java/io/netty/channel/pool/FixedChannelPoolTest.java lines 55–466

public class FixedChannelPoolTest {
    private static EventLoopGroup group;

    @BeforeAll
    public static void createEventLoop() {
        group = new MultiThreadIoEventLoopGroup(LocalIoHandler.newFactory());
    }

    @AfterAll
    public static void destroyEventLoop() {
        if (group != null) {
            group.shutdownGracefully();
        }
    }

    @Test
    public void testAcquire() throws Exception {
        Tuple t = bootstrap();

        // Start server
        Channel sc = t.sb.bind(t.address).syncUninterruptibly().channel();
        CountingChannelPoolHandler handler = new CountingChannelPoolHandler();

        ChannelPool pool = new FixedChannelPool(t.cb, handler, 1, Integer.MAX_VALUE);

        Channel channel = pool.acquire().syncUninterruptibly().getNow();
        Future<Channel> future = pool.acquire();
        assertFalse(future.isDone());

        pool.release(channel).syncUninterruptibly();
        assertTrue(future.await(1, TimeUnit.SECONDS));

        Channel channel2 = future.getNow();
        assertSame(channel, channel2);
        assertEquals(1, handler.channelCount());

        assertEquals(2, handler.acquiredCount());
        assertEquals(1, handler.releasedCount());

        sc.close().syncUninterruptibly();
        channel2.close().syncUninterruptibly();
        pool.close();
    }

    @Test
    public void testAcquireTimeout() throws Exception {
        testAcquireTimeout(500);
    }

    @Test
    public void testAcquireWithZeroTimeout() throws Exception {
        testAcquireTimeout(0);
    }

    private void testAcquireTimeout(long timeoutMillis) throws Exception {
        Tuple t = bootstrap();

        // Start server
        Channel sc = t.sb.bind(t.address).syncUninterruptibly().channel();
        ChannelPoolHandler handler = new TestChannelPoolHandler();
        ChannelPool pool = new FixedChannelPool(t.cb, handler, ChannelHealthChecker.ACTIVE,
                AcquireTimeoutAction.FAIL, timeoutMillis, 1, Integer.MAX_VALUE);

        Channel channel = pool.acquire().syncUninterruptibly().getNow();
        final Future<Channel> future = pool.acquire();
        assertThrows(TimeoutException.class, new Executable() {
            @Override
            public void execute() throws Throwable {
                future.syncUninterruptibly();
            }
        });
        sc.close().syncUninterruptibly();
        channel.close().syncUninterruptibly();
        pool.close();
    }

    @Test
    public void testAcquireNewConnection() throws Exception {
        Tuple t = bootstrap();

        // Start server