testWriteTaskRejected() — netty Function Reference

Architecture documentation for the testWriteTaskRejected() function in ChannelOutboundBufferTest.java from the netty codebase.

Function java Buffer Telemetry calls 1

Entity Profile

Buffer→ Telemetry→ testWriteTaskRejected() — netty Function Reference

Dependency Diagram

graph TD
  e8421144_91d1_9ad4_6af2_9edbae1ce8ae["testWriteTaskRejected()"]
  06eb8f18_da0c_40bf_fb14_47e2c94a6923["ChannelOutboundBufferTest"]
  e8421144_91d1_9ad4_6af2_9edbae1ce8ae -->|defined in| 06eb8f18_da0c_40bf_fb14_47e2c94a6923
  00d35efd_91f9_9891_6cdb_8a43498309e6["safeClose()"]
  e8421144_91d1_9ad4_6af2_9edbae1ce8ae -->|calls| 00d35efd_91f9_9891_6cdb_8a43498309e6
  style e8421144_91d1_9ad4_6af2_9edbae1ce8ae fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

transport/src/test/java/io/netty/channel/ChannelOutboundBufferTest.java lines 430–526

    @Test
    @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
    public void testWriteTaskRejected() throws Exception {
        final SingleThreadEventExecutor executor = new SingleThreadEventExecutor(
                null, new DefaultThreadFactory("executorPool"),
                true, 1, RejectedExecutionHandlers.reject()) {
            @Override
            protected void run() {
                do {
                    Runnable task = takeTask();
                    if (task != null) {
                        task.run();
                        updateLastExecutionTime();
                    }
                } while (!confirmShutdown());
            }

            @Override
            protected Queue<Runnable> newTaskQueue(int maxPendingTasks) {
                return super.newTaskQueue(1);
            }
        };
        final CountDownLatch handlerAddedLatch = new CountDownLatch(1);
        final CountDownLatch handlerRemovedLatch = new CountDownLatch(1);
        EmbeddedChannel ch = new EmbeddedChannel();
        ch.pipeline().addLast(executor, "handler", new ChannelOutboundHandlerAdapter() {
            @Override
            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
                promise.setFailure(new AssertionError("Should not be called"));
            }

            @Override
            public void handlerAdded(ChannelHandlerContext ctx) {
                handlerAddedLatch.countDown();
            }

            @Override
            public void handlerRemoved(ChannelHandlerContext ctx) {
                handlerRemovedLatch.countDown();
            }
        });

        // Lets wait until we are sure the handler was added.
        handlerAddedLatch.await();

        final CountDownLatch executeLatch = new CountDownLatch(1);
        final CountDownLatch runLatch = new CountDownLatch(1);
        executor.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    runLatch.countDown();
                    executeLatch.await();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        });

        runLatch.await();

        executor.execute(new Runnable() {
            @Override
            public void run() {
                // Will not be executed but ensure the pending count is 1.
            }
        });

        assertEquals(1, executor.pendingTasks());
        assertEquals(0, ch.unsafe().outboundBuffer().totalPendingWriteBytes());

        ByteBuf buffer = buffer(128).writeZero(128);
        ChannelFuture future = ch.write(buffer);
        ch.runPendingTasks();

        assertTrue(future.cause() instanceof RejectedExecutionException);
        assertEquals(0, buffer.refCnt());

        // In case of rejected task we should not have anything pending.
        assertEquals(0, ch.unsafe().outboundBuffer().totalPendingWriteBytes());
        executeLatch.countDown();