@Override
    public void read(ChannelHandlerContext ctx) throws Exception {
        if (dequeue(ctx, 1) == 0) {
            // It seems no messages were consumed. We need to read() some
            // messages from upstream and once one arrives it need to be
            // relayed to downstream to keep the flow going.
            shouldConsume = true;
            ctx.read();
        } else if (config.isAutoRead()) {
            ctx.read();
        }
    }
@Test
    public void testSwallowedReadComplete() throws Exception {
        final long delayMillis = 100;
        final Queue<IdleStateEvent> userEvents = new LinkedBlockingQueue<IdleStateEvent>();
        final EmbeddedChannel channel = new EmbeddedChannel(false, false,
            new FlowControlHandler(),
            new IdleStateHandler(delayMillis, 0, 0, MILLISECONDS),
            new ChannelInboundHandlerAdapter() {
                @Override
                public void channelActive(ChannelHandlerContext ctx) {
                    ctx.fireChannelActive();
                    ctx.read();
                }

                @Override
                public void channelRead(ChannelHandlerContext ctx, Object msg) {
                    ctx.fireChannelRead(msg);
                    ctx.read();
                }

                @Override
                public void channelReadComplete(ChannelHandlerContext ctx) {
                    ctx.fireChannelReadComplete();
                    ctx.read();
                }

                @Override
                public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
                    if (evt instanceof IdleStateEvent) {
                        userEvents.add((IdleStateEvent) evt);
                    }
                    ctx.fireUserEventTriggered(evt);
                }
            }
        );

        channel.config().setAutoRead(false);
        assertFalse(channel.config().isAutoRead());

        channel.register();

        // Reset read timeout by some message
        assertTrue(channel.writeInbound(Unpooled.EMPTY_BUFFER));
        channel.flushInbound();
        assertEquals(Unpooled.EMPTY_BUFFER, channel.readInbound());

        // Emulate 'no more messages in NIO channel' on the next read attempt.
        channel.flushInbound();
        assertNull(channel.readInbound());

        Thread.sleep(delayMillis + 20L);
        channel.runPendingTasks();
        assertEquals(IdleStateEvent.FIRST_READER_IDLE_STATE_EVENT, userEvents.poll());
        assertFalse(channel.finish());
    }