고급 코덱 패턴

// ReplayingDecoder를 사용하면 readableBytes() 체크가 필요 없다
public class SimpleReplayingDecoder extends ReplayingDecoder<Void> {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        // 바이트가 부족하면 자동으로 Signal을 던져서 재시도한다
        int magicNumber = in.readInt();   // 4바이트 필요
        int dataLength = in.readInt();    // 4바이트 필요
        byte[] data = new byte[dataLength];
        in.readBytes(data);               // dataLength만큼 필요

        out.add(new MyMessage(magicNumber, data));
    }
}

// 상태를 enum으로 정의
public enum MyProtocolState {
    READ_HEADER,  // 헤더 읽기 단계
    READ_BODY     // 바디 읽기 단계
}

public class StateMachineDecoder extends ReplayingDecoder<MyProtocolState> {

    private int bodyLength;

    // 초기 상태를 READ_HEADER로 설정
    public StateMachineDecoder() {
        super(MyProtocolState.READ_HEADER);
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        switch (state()) {
            case READ_HEADER:
                // 헤더에서 매직 넘버(4바이트)와 바디 길이(4바이트)를 읽는다
                int magic = in.readInt();
                bodyLength = in.readInt();

                // 상태를 전환하고 checkpoint를 설정
                checkpoint(MyProtocolState.READ_BODY);
                break;

            case READ_BODY:
                // 헤더에서 읽은 길이만큼 바디를 읽는다
                byte[] body = new byte[bodyLength];
                in.readBytes(body);

                // 디코딩 완료, 다음 메시지를 위해 상태 초기화
                checkpoint(MyProtocolState.READ_HEADER);
                out.add(new MyProtocolMessage(body));
                break;
        }
    }
}

// checkpoint 없이 사용하면 매번 처음부터 다시 읽는다
// checkpoint를 사용하면 저장된 상태부터 이어서 읽는다

// 잘못된 예시 — checkpoint 없음
case READ_HEADER:
    int magic = in.readInt();
    bodyLength = in.readInt();
    state = READ_BODY;  // 상태만 바꾸고 checkpoint를 안 찍음
    break;

// 올바른 예시 — checkpoint 사용
case READ_HEADER:
    int magic = in.readInt();
    bodyLength = in.readInt();
    checkpoint(MyProtocolState.READ_BODY);  // 상태 전환 + 위치 저장
    break;

바이트 스트림
    ↓
[프레임 디코더]        — TCP 스트림 → 프레임 단위 분리
    ↓
[프로토콜 디코더]      — 프레임 → 비즈니스 객체 변환
    ↓
[비즈니스 핸들러]      — 비즈니스 로직 처리
    ↓
[프로토콜 인코더]      — 비즈니스 객체 → 프레임 변환
    ↓
바이트 스트림

public class MyServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) {
        ChannelPipeline pipeline = ch.pipeline();

        // 1단계: 프레임 디코더 — 길이 기반으로 프레임 분리
        pipeline.addLast("frameDecoder",
            new LengthFieldBasedFrameDecoder(
                65535,  // 최대 프레임 크기
                0,      // 길이 필드 오프셋
                4,      // 길이 필드 크기 (4바이트 int)
                0,      // 길이 조정값
                4       // 길이 필드를 결과에서 제거할 바이트 수
            ));

        // 2단계: 프로토콜 디코더 — 프레임을 비즈니스 객체로 변환
        pipeline.addLast("protocolDecoder", new MyProtocolDecoder());

        // 3단계: 비즈니스 핸들러
        pipeline.addLast("businessHandler", new MyBusinessHandler());

        // 4단계: 프로토콜 인코더 — 응답 객체를 바이트로 변환
        pipeline.addLast("protocolEncoder", new MyProtocolEncoder());

        // 5단계: 프레임 인코더 — 길이 헤더 추가
        pipeline.addLast("frameEncoder",
            new LengthFieldPrepender(4));
    }
}

public class HttpServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) {
        ChannelPipeline pipeline = ch.pipeline();

        // HTTP 요청/응답 디코딩
        pipeline.addLast("httpDecoder", new HttpServerCodec());

        // 청크를 하나의 FullHttpRequest로 합침 (최대 1MB)
        pipeline.addLast("aggregator",
            new HttpObjectAggregator(1024 * 1024));

        // 비즈니스 핸들러 — FullHttpRequest를 받아서 처리
        pipeline.addLast("handler", new MyHttpHandler());
    }
}

public class MyHttpHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
        // 청크 걱정 없이 완전한 요청을 받는다
        String uri = request.uri();
        ByteBuf content = request.content();

        // 응답 생성
        FullHttpResponse response = new DefaultFullHttpResponse(
            HttpVersion.HTTP_1_1,
            HttpResponseStatus.OK,
            Unpooled.copiedBuffer("OK", CharsetUtil.UTF_8)
        );
        response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain");
        response.headers().set(HttpHeaderNames.CONTENT_LENGTH,
            response.content().readableBytes());

        ctx.writeAndFlush(response);
    }
}

// Integer ↔ String 양방향 변환 코덱
public class IntegerStringCodec
        extends MessageToMessageCodec<Integer, String> {

    @Override
    protected void encode(ChannelHandlerContext ctx,
                          String msg, List<Object> out) {
        // 아웃바운드: String → Integer
        out.add(Integer.valueOf(msg));
    }

    @Override
    protected void decode(ChannelHandlerContext ctx,
                          Integer msg, List<Object> out) {
        // 인바운드: Integer → String
        out.add(msg.toString());
    }
}

// 프로토콜 v1 ↔ v2 변환 코덱
public class ProtocolVersionCodec
        extends MessageToMessageCodec<ProtocolV1Message, ProtocolV2Message> {

    @Override
    protected void encode(ChannelHandlerContext ctx,
                          ProtocolV2Message msg, List<Object> out) {
        // 내부(v2) → 외부(v1)로 변환하여 전송
        ProtocolV1Message v1 = convertToV1(msg);
        out.add(v1);
    }

    @Override
    protected void decode(ChannelHandlerContext ctx,
                          ProtocolV1Message msg, List<Object> out) {
        // 외부(v1) → 내부(v2)로 변환하여 처리
        ProtocolV2Message v2 = convertToV2(msg);
        out.add(v2);
    }

    private ProtocolV1Message convertToV1(ProtocolV2Message v2) {
        // v2 → v1 변환 로직
        return new ProtocolV1Message(v2.getPayload());
    }

    private ProtocolV2Message convertToV2(ProtocolV1Message v1) {
        // v1 → v2 변환 로직
        return new ProtocolV2Message(v1.getData(), DEFAULT_VERSION);
    }
}

// 프로토콜별 코덱 체인을 팩토리로 분리
public final class MyProtocolCodecFactory {

    private MyProtocolCodecFactory() {
        // 인스턴스 생성 방지
    }

    // 서버 측 코덱 체인 구성
    public static void addServerCodecs(ChannelPipeline pipeline) {
        // 프레임 디코딩
        pipeline.addLast("frameDecoder",
            new LengthFieldBasedFrameDecoder(65535, 0, 4, 0, 4));
        pipeline.addLast("frameEncoder",
            new LengthFieldPrepender(4));

        // 프로토콜 디코딩/인코딩
        pipeline.addLast("protocolDecoder", new MyProtocolDecoder());
        pipeline.addLast("protocolEncoder", new MyProtocolEncoder());

        // 공통 핸들러
        pipeline.addLast("loggingHandler", new LoggingHandler(LogLevel.DEBUG));
    }

    // 클라이언트 측 코덱 체인 구성
    public static void addClientCodecs(ChannelPipeline pipeline) {
        pipeline.addLast("frameDecoder",
            new LengthFieldBasedFrameDecoder(65535, 0, 4, 0, 4));
        pipeline.addLast("frameEncoder",
            new LengthFieldPrepender(4));

        pipeline.addLast("protocolDecoder", new MyProtocolDecoder());
        pipeline.addLast("protocolEncoder", new MyProtocolEncoder());
    }
}

// 팩토리를 사용하면 ChannelInitializer가 깔끔해진다
public class MyServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) {
        ChannelPipeline pipeline = ch.pipeline();

        // 코덱 체인을 한 줄로 구성
        MyProtocolCodecFactory.addServerCodecs(pipeline);

        // 비즈니스 핸들러만 추가
        pipeline.addLast("businessHandler", new MyBusinessHandler());
    }
}

// 나쁜 예시 — decode()가 호출될 때마다 새 바이트 배열을 생성
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
    int length = in.readInt();
    byte[] data = new byte[length];  // 매번 new 배열 생성 → GC 부담
    in.readBytes(data);
    out.add(new MyMessage(data));
}

// 개선 예시 — ByteBuf를 직접 전달하여 복사 줄이기
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
    int length = in.readInt();
    ByteBuf payload = in.readRetainedSlice(length);  // 복사 없이 슬라이스
    out.add(new MyMessage(payload));  // 참조 카운팅 주의!
}

public class EfficientEncoder extends MessageToByteEncoder<MyMessage> {
    @Override
    protected void encode(ChannelHandlerContext ctx,
                          MyMessage msg, ByteBuf out) {
        // out은 네티가 미리 할당한 ByteBuf — 이걸 그대로 쓰면 된다
        out.writeInt(msg.getType());
        out.writeInt(msg.getPayload().length);
        out.writeBytes(msg.getPayload());
        // 별도의 ByteBuf를 할당하지 않는 것이 핵심
    }
}

// 상태가 없는 인코더 — 인스턴스 하나를 공유 가능
@ChannelHandler.Sharable
public class MyProtocolEncoder extends MessageToByteEncoder<MyMessage> {
    // 인스턴스 필드 없음 → 스레드 안전

    @Override
    protected void encode(ChannelHandlerContext ctx,
                          MyMessage msg, ByteBuf out) {
        out.writeInt(msg.getType());
        out.writeBytes(msg.getPayload());
    }
}

// 사용 시 인스턴스를 한 번만 생성
public class MyServerInitializer extends ChannelInitializer<SocketChannel> {
    // 인코더 인스턴스를 공유
    private final MyProtocolEncoder sharedEncoder = new MyProtocolEncoder();

    @Override
    protected void initChannel(SocketChannel ch) {
        ch.pipeline().addLast("encoder", sharedEncoder);  // 동일 인스턴스 재사용
    }
}