SingleThreadEventExecutorTest Class — netty Architecture

Source Code

common/src/test/java/io/netty/util/concurrent/SingleThreadEventExecutorTest.java lines 45–662

public class SingleThreadEventExecutorTest {

    private static final class TestThread extends Thread {
        private final CountDownLatch startedLatch = new CountDownLatch(1);
        private final CountDownLatch runLatch = new CountDownLatch(1);

        TestThread(Runnable task) {
            super(task);
        }

        @Override
        public void start() {
            super.start();
            startedLatch.countDown();
        }

        @Override
        public void run() {
            runLatch.countDown();
            super.run();
        }

        void awaitStarted() throws InterruptedException {
            startedLatch.await();
        }

        void awaitRunnableExecution() throws InterruptedException {
            runLatch.await();
        }
    }

    private static final class TestThreadFactory implements ThreadFactory {
        final LinkedBlockingQueue<TestThread> threads = new LinkedBlockingQueue<>();
        @Override
        public Thread newThread(@NotNull Runnable r) {
            TestThread thread = new TestThread(r);
            threads.add(thread);
            return thread;
        }
    }

    private static final class SuspendingSingleThreadEventExecutor extends SingleThreadEventExecutor {

        SuspendingSingleThreadEventExecutor(ThreadFactory threadFactory) {
            super(null, threadFactory, false, true,
                    Integer.MAX_VALUE, RejectedExecutionHandlers.reject());
        }

        @Override
        protected void run() {
            while (!confirmShutdown() && !canSuspend()) {
                Runnable task = takeTask();
                if (task != null) {
                    task.run();
                }
            }
        }

        @Override
        protected void wakeup(boolean inEventLoop) {
            interruptThread();
        }
    }

    @Test
    void testSuspension() throws Exception {
        TestThreadFactory threadFactory = new TestThreadFactory();
        final SingleThreadEventExecutor executor = new SuspendingSingleThreadEventExecutor(threadFactory);
        LatchTask task1 = new LatchTask();
        executor.execute(task1);
        Thread currentThread = threadFactory.threads.take();
        assertTrue(executor.trySuspend());
        task1.await();

        // Let's wait till the current Thread did die....
        currentThread.join();

        // Should be suspended now, we should be able to also call trySuspend() again.
        assertTrue(executor.isSuspended());
        // There was no thread created as we did not try to execute something yet.
        assertTrue(threadFactory.threads.isEmpty());