testConcurrentAddRemove() — netty Function Reference

Source Code

transport/src/test/java/io/netty/channel/local/LocalTransportThreadModelTest3.java lines 127–243

    private static void testConcurrentAddRemove(boolean inbound) throws Exception {
        EventLoopGroup l = new MultiThreadIoEventLoopGroup(
                4, new DefaultThreadFactory("l"), LocalIoHandler.newFactory());
        EventExecutorGroup e1 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e1"));
        EventExecutorGroup e2 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e2"));
        EventExecutorGroup e3 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e3"));
        EventExecutorGroup e4 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e4"));
        EventExecutorGroup e5 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e5"));

        final EventExecutorGroup[] groups = {e1, e2, e3, e4, e5};
        try {
            Deque<EventType> events = new ConcurrentLinkedDeque<EventType>();
            final EventForwarder h1 = new EventForwarder();
            final EventForwarder h2 = new EventForwarder();
            final EventForwarder h3 = new EventForwarder();
            final EventForwarder h4 = new EventForwarder();
            final EventForwarder h5 = new EventForwarder();
            final EventRecorder h6 = new EventRecorder(events, inbound);

            final Channel ch = new LocalChannel();
            if (!inbound) {
                ch.config().setAutoRead(false);
            }
            ch.pipeline().addLast(e1, h1)
                    .addLast(e1, h2)
                    .addLast(e1, h3)
                    .addLast(e1, h4)
                    .addLast(e1, h5)
                    .addLast(e1, "recorder", h6);

            l.register(ch).sync().channel().connect(localAddr).sync();

            final LinkedList<EventType> expectedEvents = events(inbound, 8192);

            Throwable cause = new Throwable();

            Thread pipelineModifier = new Thread(new Runnable() {
                @Override
                public void run() {
                    Random random = new Random();

                    while (true) {
                        try {
                            Thread.sleep(100);
                        } catch (InterruptedException e) {
                            return;
                        }
                        if (!ch.isRegistered()) {
                            continue;
                        }
                        //EventForwardHandler forwardHandler = forwarders[random.nextInt(forwarders.length)];
                        ChannelHandler handler = ch.pipeline().removeFirst();
                        ch.pipeline().addBefore(groups[random.nextInt(groups.length)], "recorder",
                                UUID.randomUUID().toString(), handler);
                    }
                }
            });
            pipelineModifier.setDaemon(true);
            pipelineModifier.start();
            for (EventType event: expectedEvents) {
                switch (event) {
                    case EXCEPTION_CAUGHT:
                        ch.pipeline().fireExceptionCaught(cause);
                        break;
                    case MESSAGE_RECEIVED:
                        ch.pipeline().fireChannelRead("");
                        break;
                    case MESSAGE_RECEIVED_LAST:
                        ch.pipeline().fireChannelReadComplete();
                        break;
                    case USER_EVENT:
                        ch.pipeline().fireUserEventTriggered("");
                        break;
                    case WRITE:
                        ch.pipeline().write("");
                        break;
                    case READ:
                        ch.pipeline().read();
                        break;
                }
            }