Netty如何自定义编码解码器

com.netty.codec.custom.InboundAndOutboundHandler#decode 

@Override
protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) {
    // Long => 8 byte
    if (byteBuf.readableBytes() >= 8) {
        list.add(byteBuf.readLong());
    } else log.warn("字节数异常 => {}", byteBuf.readableBytes());
}

@Override
protected void encode(ChannelHandlerContext channelHandlerContext, Long msg, ByteBuf byteBuf) {
    byteBuf.writeLong(msg);
    log.info("发送消息成功");
}

io.netty.handler.codec.ReplayingDecoder

public abstract class ReplayingDecoder<S> extends ByteToMessageDecoder {
    ...
}

@Override
protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
    replayable.setCumulation(in);
    try {
        while (in.isReadable()) {
            int oldReaderIndex = checkpoint = in.readerIndex();
            int outSize = out.size();

            if (outSize > 0) {
                fireChannelRead(ctx, out, outSize);
                out.clear();

                // Check if this handler was removed before continuing with decoding.
                // If it was removed, it is not safe to continue to operate on the buffer.
                //
                // See:
                // - https://github.com/netty/netty/issues/4635
                if (ctx.isRemoved()) {
                    break;
                }
                outSize = 0;
            }

            S oldState = state;
            int oldInputLength = in.readableBytes();
            try {
                decodeRemovalReentryProtection(ctx, replayable, out);

                // Check if this handler was removed before continuing the loop.
                // If it was removed, it is not safe to continue to operate on the buffer.
                //
                // See https://github.com/netty/netty/issues/1664
                if (ctx.isRemoved()) {
                    break;
                }

                if (outSize == out.size()) {
                    if (oldInputLength == in.readableBytes() && oldState == state) {
                        throw new DecoderException(
                                StringUtil.simpleClassName(getClass()) + ".decode() must consume the inbound " +
                                "data or change its state if it did not decode anything.");
                    } else {
                        // Previous data has been discarded or caused state transition.
                        // Probably it is reading on.
                        continue;
                    }
                }
            } catch (Signal replay) {
                replay.expect(REPLAY);

                // Check if this handler was removed before continuing the loop.
                // If it was removed, it is not safe to continue to operate on the buffer.
                //
                // See https://github.com/netty/netty/issues/1664
                if (ctx.isRemoved()) {
                    break;
                }

                // Return to the checkpoint (or oldPosition) and retry.
                int checkpoint = this.checkpoint;
                if (checkpoint >= 0) {
                    in.readerIndex(checkpoint);
                } else {
                    // Called by cleanup() - no need to maintain the readerIndex
                    // anymore because the buffer has been released already.
                }
                break;
            }

            if (oldReaderIndex == in.readerIndex() && oldState == state) {
                throw new DecoderException(
                       StringUtil.simpleClassName(getClass()) + ".decode() method must consume the inbound data " +
                       "or change its state if it decoded something.");
            }
            if (isSingleDecode()) {
                break;
            }
        }
    } catch (DecoderException e) {
        throw e;
    } catch (Exception cause) {
        throw new DecoderException(cause);
    }
}

int oldReaderIndex = checkpoint = in.readerIndex();