How to Build a Mini‑Netty Pipeline that Decouples Decoding from Business Logic

In the previous article we implemented free read/write, non‑blocking write, and cumulative read capabilities, but the decoder and business logic were tightly coupled, making extension difficult.

The current implementation initializes PacketServerBootstrap with a Codec and a PacketHandler, directly passing decoded messages to handler.channelRead, which hard‑codes the decoder and handler together.

public PacketServerBootstrap childHandler(MessageCodec codec, PacketHandler handler) {
    this.codec = codec;
    this.handler = handler;
    return this;
}

If we want to add extra capabilities such as logging, message filtering, or authentication after decoding, we would have to modify the bootstrap code each time, leading to cumbersome hard‑coded logic.

Inspired by Spring MVC filters, Netty uses a Chain of Responsibility where decoding, encoding, and business processing are split into a ChannelPipeline. This allows developers to plug in handlers for logging, authentication, compression, heartbeat, etc., without touching the core decoder.

1. Responsibility Chain Model Comparison

The chain can be implemented as broadcast (loop through all handlers) or as a linked‑list style pass‑along. Broadcast calls every handler for each event, which can cause handlers to receive inappropriate data (e.g., both MessageCodec and ServerHandler receiving a ByteBuffer).

for (ChannelHandler h : handlers) {
    h.channelRead(ctx, msg);
}

Linked‑list passing lets each handler focus on its own step and invoke ctx.fireChannelRead(x) to forward the event to the next handler, effectively forming a double‑linked list.

2. Designing the Responsibility Chain

1) Handler

Handlers are divided into inbound (processing from channel to business) and outbound (processing from business to channel). Inbound handlers handle connection establishment, data reading, connection closure, and exception catching. Outbound handlers handle data writing and message encoding.

public interface MiniChannelOutboundHandler extends MiniChannelHandler {
    // write method
}

public interface MiniChannelInboundHandler extends MiniChannelHandler {
    // read method
}

2) Handler Context (MiniHandlerContext)

The context holds the handler instance, its name, the associated channel, the pipeline reference, its index in the handler array, and other metadata. It enables the pipeline to locate the next or previous handler efficiently.

public class MiniHandlerContext {
    private String name;
    private final MiniChannelHandler handler;
    private final MiniChannel channel;
    private final MiniChannelPipeline pipeline;
    private int index;
}

Methods such as fireChannelRead and fireChannelWrite use the stored index to forward events without external loops.

3) Pipeline Object

The pipeline maintains an array of MiniHandlerContext objects. It provides methods to add handlers, retrieve the next handler based on index, and trigger event propagation.

MessageCodec.channelRead(ctx, ByteBuffer)
   ⬇ decode
ctx.fireChannelRead("hello")
   ⬇ ServerHandler1.channelRead(...)
   ⬇ ServerHandler2.channelRead(...)
MessageCodec.channelWrite(ctx, ByteBuffer)
   ⬇ encode

Read operations start from the channel, fill a ByteBuffer, and pass it to the pipeline head. Write operations collect the final ByteBuffer at the pipeline tail, enqueue it, and register OP_WRITE on the selector.

public static void doRead(MiniChannel ch) {
    ByteBuffer buffer = ByteBuffer.allocate(7);
    try {
        int len;
        while ((len = ch.socketChannel().read(buffer)) > 0) {
            buffer.flip();
            ByteBuffer accumulate = ch.accumulate(buffer);
            ch.receive(accumulate);
            buffer.clear();
        }
        if (len == -1) {
            ch.inactive();
        }
    } catch (Exception e) {
        ch.exception(e);
    }
}

protected void fireChannelWrite(Object msg, int index) {
    if (index < 0) {
        try {
            channel.outQueue().add((ByteBuffer) msg);
            SelectionKey key = channel.socketChannel().keyFor(channel.selector());
            key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
        } catch (Exception e) {
            fireExceptionCaught(e, index);
        }
        return;
    }
    MiniHandlerContext context = contexts.get(index);
    MiniChannelHandler handler = context.handler();
    if (handler instanceof MiniChannelOutboundHandler) {
        ((MiniChannelOutboundHandler) handler).channelWrite(context, msg);
    } else {
        fireChannelWrite(msg, index - 1);
    }
}

3. Pipeline Entry and Exit

When the selector detects OP_READ, bytes are read from the SocketChannel into a ByteBuffer and handed to the pipeline head for processing. The read flow is now mediated by the pipeline, allowing layered decoding, authentication, logging, etc.

For writes, the final ByteBuffer is placed into the channel’s out‑queue, and OP_WRITE is registered. The actual write occurs only when the selector reports the channel is writable.

4. Initializing the Pipeline

A MiniChannelInitializer subclass configures the pipeline in the server’s main method, adding encoder, decoder, logger, and business handlers in order.

public class MiniNettyServer {
    public static void main(String[] args) {
        MiniEventLoopGroup workGroup = new MiniEventLoopGroup(4);
        try {
            MiniServerBootstrap bootstrap = new MiniServerBootstrap();
            bootstrap.group(workGroup)
                .childHandler(new MiniChannelInitializer() {
                    @Override
                    public void initChannel(MiniChannel ch) {
                        ch.pipeline().addLast("Encoder", new LengthFieldMessageEncoder());
                        ch.pipeline().addLast("Decoder", new LengthFieldMessageDecoder());
                        ch.pipeline().addLast("LoggerHandler", new LoggerHandler());
                        ch.pipeline().addLast("ServerCheckHandler", new ServerCheckHandler());
                        ch.pipeline().addLast("ServerEchoHandler", new ServerEchoHandler());
                    }
                })
                .bind(10030)
                .sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            workGroup.shutdownGracefully();
        }
    }
}

With this mini‑Netty framework, developers no longer need to worry about thread saturation, socket sticky packets, blocking reads/writes, or complex business dispatching. The next article will cover heartbeat mechanisms to keep long‑lived connections stable.