Netty Async Flow Control

import io.netty.channel.*;
import io.netty.util.concurrent.*;

import java.nio.charset.StandardCharsets;
import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public final class PromiseFutureListenerExampleHandler extends SimpleChannelInboundHandler<Object> {

    // 블로킹 작업(외부 API, DB 등)을 오프로드할 풀
    private final ExecutorService blockingPool = Executors.newFixedThreadPool(16);

    // write 백프레셔 시 잠깐 쌓아두는 큐(데모용)
    private final Queue<Object> pendingWrites = new ArrayDeque<>();

    // backpressure 제어 예시: 너무 많이 쌓이면 읽기 끄기
    private static final int MAX_INFLIGHT = 100;
    private int inflight = 0;

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        // 필요 시 autoRead 제어 가능
        ctx.channel().config().setAutoRead(true);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) {
        // inflight 제한(세션/동시성 제한 같은 개념)
        if (inflight >= MAX_INFLIGHT) {
            // 더 못 받겠으면 읽기 중단
            ctx.channel().config().setAutoRead(false);
            // msg는 버퍼/레퍼런스 관리 필요할 수 있음(여기서는 Object로 단순화)
            return;
        }

        inflight++;

        // 1) Promise 생성: 이 작업의 완료를 내가 setSuccess/setFailure로 완료시킨다.
        Promise<String> promise = new DefaultPromise<>(ctx.executor()); // ctx.executor() == EventLoop

        // 2) Promise 완료 시 동작(리스너)
        promise.addListener((Future<String> f) -> {
            // 이 리스너는 기본적으로 EventLoop에서 실행되도록 설계됨(DefaultPromise가 ctx.executor 기반)
            try {
                if (f.isSuccess()) {
                    String response = f.getNow();

                    // 3) ChannelFuture: write의 비동기 결과
                    ChannelFuture writeFuture;
                    if (ctx.channel().isWritable()) {
                        writeFuture = ctx.writeAndFlush(toByteBuf(ctx, response));
                    } else {
                        // 백프레셔면 큐에 넣고 나중에 flush
                        pendingWrites.add(toByteBuf(ctx, response));
                        return;
                    }

                    // 4) ChannelFuture에 GenericFutureListener 붙이기
                    writeFuture.addListener((ChannelFuture wf) -> {
                        if (wf.isSuccess()) {
                            // write 성공
                        } else {
                            // write 실패 - 원인 확인
                            Throwable cause = wf.cause();
                            wf.channel().close();
                        }
                    });

                } else {
                    // 비동기 작업 실패
                    Throwable cause = f.cause();
                    ctx.writeAndFlush(toByteBuf(ctx, "ERROR: " + cause.getMessage()))
                       .addListener(ChannelFutureListener.CLOSE);
                }
            } finally {
                inflight--;
                // inflight 줄었으니 읽기 재개
                if (!ctx.channel().config().isAutoRead() && inflight < MAX_INFLIGHT) {
                    ctx.channel().config().setAutoRead(true);
                    // autoRead를 켠 뒤 필요하면 read 트리거
                    ctx.read();
                }
            }
        });

        // 5) 실제 블로킹 작업(예: 외부 API 호출)을 별도 풀에서 실행
        blockingPool.execute(() -> {
            try {
                // (중요) 여기선 EventLoop 스레드가 아님. 절대 ctx.write 같은 Netty 작업 직접 호출하지 말기.
                String result = doBlockingCall(msg);

                // 6) Promise 완료는 EventLoop에서 수행되도록 넘기는 게 안전(스레드 안전/일관성)
                ctx.executor().execute(() -> promise.setSuccess(result));

            } catch (Throwable t) {
                ctx.executor().execute(() -> promise.setFailure(t));
            }
        });
    }

    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) {
        // 채널이 다시 writable이 되면 pendingWrites를 비운다.
        if (ctx.channel().isWritable()) {
            while (!pendingWrites.isEmpty()) {
                Object out = pendingWrites.poll();
                ChannelFuture f = ctx.write(out);
                f.addListener((ChannelFuture wf) -> {
                    if (!wf.isSuccess()) {
                        wf.channel().close();
                    }
                });
            }
            ctx.flush();
        }
        ctx.fireChannelWritabilityChanged();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        ctx.close();
    }

    private static Object toByteBuf(ChannelHandlerContext ctx, String s) {
        return ctx.alloc().buffer().writeBytes(s.getBytes(StandardCharsets.UTF_8));
    }

    private static String doBlockingCall(Object msg) {
        // 데모: 여기서 외부 API/DB 호출 등 블로킹을 한다고 가정
        // Thread.sleep(50);
        return "OK";
    }
}