In Netty, packet decoders handle inbound byte streams using consistent principles regardless of their type—length-based, delimiter-based, or custom protocol decoders. The core challenge lies in reconstructing full application messages from partial TCP reads, handling both complete packets and fragmented ones.

Reassembling Complete Packets

When enough bytes arrive to form a valid message, they are extracted from the incoming stream and past as a discrete unit to subsequent handlers. This ensures business logic processes whole payloads.

Handling Fragmented Data

If an incoming chunk does not contain a full message, it is retained temporarily. Accumulated bytes are held until additional reads complete the message, at which point the reconstructed packet is forwarded.

Manual Packet Reconstruction Without Netty

Without a framework, developers manually pull data from the TCP buffer, checking after each read whether a full logical message can be formed:

1. If current data is insufficient, retain it and continue reading until a full message is available.
2. If combined with previously retained bytes a full message emerges, extract it for processing while preserving any leftover bytes for future assembly.

Netty's Base Decoder Class

All built-in decoders extend ByteToMessageDecoder. Each channel owns a pipeline containing these decoder instances. The base class maintains a cumulative buffer where incoming bytes are aggregated before decoding attempts.

Key attributes of ByteToMessageDecoder:

public abstract class ByteToMessageDecoder extends ChannelInboundHandlerAdapter {
    ByteBuf accumulationBuf;
    private Cumulator merger;
    private boolean decodeOnceMode;
    private boolean decodedEmpty;
    private boolean initialPass;
    private int maxReadAttempts;
    private int readCounter;
}

The channelRead method is invoked whenever bytes are received:

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (msg instanceof ByteBuf) {
        CodecOutputList resultList = CodecOutputList.newInstance();
        try {
            ByteBuf inputBuf = (ByteBuf) msg;
            initialPass = (accumulationBuf == null);
            if (initialPass) {
                accumulationBuf = inputBuf;
            } else {
                accumulationBuf = merger.merge(ctx.alloc(), accumulationBuf, inputBuf);
            }
            decodeLoop(ctx, accumulationBuf, resultList);
        } catch (DecoderException e) {
            throw e;
        } catch (Throwable t) {
            throw new DecoderException(t);
        } finally {
            if (accumulationBuf != null && !accumulationBuf.isReadable()) {
                readCounter = 0;
                accumulationBuf.release();
                accumulationBuf = null;
            } else if (++readCounter >= maxReadAttempts) {
                readCounter = 0;
                compactAccumulation();
            }
            int resultCount = resultList.size();
            decodedEmpty = !resultList.insertSinceRecycled();
            dispatchResults(ctx, resultList, resultCount);
            resultList.recycle();
        }
    } else {
        ctx.fireChannelRead(msg);
    }
}

Processing steps:

1. Obtain a reusable list from the pool.
2. On first invocation, assign incoming buffer directly; otherwise merge with existing accumulation buffer.
3. Invoke subclass decode logic repeatedly via decodeLoop, populating the result list with fully parsed messages.
4. Release accumulation buffer if fully consumed.
5. If unconsumed bytes remain after many read attempts, compact the buffer by discarding already-processed bytes.
6. Forward all decoded messages downstream and recycle the list.

Merging Incoming Bytes

If no prior data exists, assignment avoids copying. Otherwise, merging appends new bytes:

ByteBuf inputBuf = (ByteBuf) msg;
initialPass = (accumulationBuf == null);
if (initialPass) {
    accumulationBuf = inputBuf;
} else {
    accumulationBuf = merger.merge(ctx.alloc(), accumulationBuf, inputBuf);
}

Default merger implementation:

public static final Cumulator MERGE_CUMULATOR = (allocator, existing, incoming) -> {
    ByteBuf target;
    if (existing.writerIndex() <= existing.maxCapacity() - incoming.readableBytes() && existing.refCnt() <= 1) {
        target = existing;
    } else {
        target = growBuffer(allocator, existing, incoming.readableBytes());
    }
    target.writeBytes(incoming);
    incoming.release();
    return target;
};

Buffer growth helper:

static ByteBuf growBuffer(ByteBufAllocator allocator, ByteBuf oldBuf, int extraSize) {
    ByteBuf newBuf = allocator.buffer(oldBuf.readableBytes() + extraSize);
    newBuf.writeBytes(oldBuf);
    oldBuf.release();
    return newBuf;
}

Decoding Loop Mechanics

After appending data, decodeLoop extracts complete messages:

protected void decodeLoop(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
    while (in.isReadable()) {
        int prevSize = out.size();
        if (prevSize > 0) {
            emitBatch(ctx, out, prevSize);
            out.clear();
            if (ctx.isRemoved()) break;
            prevSize = 0;
        }
        int beforeLen = in.readableBytes();
        decode(ctx, in, out);
        if (ctx.isRemoved()) break;
        if (prevSize == out.size()) {
            if (beforeLen == in.readableBytes()) break;
            else continue;
        }
        if (decodeOnceMode) break;
    }
}

Emission helper:

static void emitBatch(ChannelHandlerContext ctx, List<Object> results, int count) {
    for (int i = 0; i < count; i++) {
        ctx.fireChannelRead(results.get(i));
    }
}

Key points:

• Continues looping while bytes remain.
• After each successful extraction, forwards batch downstream.
• Stops if context removed, no progress made, or single-decode mode enabled.
• Subclasses implement decode to parse exactly one message from the buffer, adding it to output list when complete.

Cleaning Up the Accumulation Buffer

To prevent memory leaks, processed bytes must be discarded:

finally {
    if (accumulationBuf != null && !accumulationBuf.isReadable()) {
        readCounter = 0;
        accumulationBuf.release();
        accumulationBuf = null;
    } else if (++readCounter >= maxReadAttempts) {
        readCounter = 0;
        compactAccumulation();
    }
    int total = out.size();
    decodedEmpty = !out.insertSinceRecycled();
    dispatchResults(ctx, out, total);
    out.recycle();
}

Compaction strategy:

private void compactAccumulation() {
    ensureAccessible();
    if (readerIdx == 0) return;
    if (readerIdx == writerIdx) {
        adjustMarkers(readerIdx);
        writerIdx = readerIdx = 0;
        return;
    }
    if (readerIdx >= capacity() >>> 1) {
        setBytes(0, this, readerIdx, writerIdx - readerIdx);
        writerIdx -= readerIdx;
        adjustMarkers(readerIdx);
        readerIdx = 0;
    }
}

This shifts unread bytes to the start of the buffer, resetting indices to reclaim space.

The design uses a template pattern: ByteToMessageDecoder manages accumulation and iteration, while subclasses define protocol-specific parsing in decode. This approach cleanly handles both concatenated packets and fragments, ensuring reliable delivery of whole messages to business handlers.