Page History

package com.example.cafe24.api;

import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpServer;

import java.io.IOException;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.URLDecoder;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Cafe24 API를 시뮬레이션하는 더미 서버.
 *
 * <p>Leaky Bucket 알고리즘으로 요청 수를 제한합니다.
 * <ul>
 *   <li>버킷 용량(capacity)을 초과하면 429 Too Many Requests 반환</li>
 *   <li>버킷은 매초 leakRate만큼 자동 감소</li>
 *   <li>응답 헤더: X-Api-Call-Limit, X-Cafe24-Call-Usage, X-Cafe24-Call-Remain</li>
 * </ul>
 *
 * <p>응답 규칙: "hello" 요청 → "world" 반환, 나머지 단어는 그대로 에코.
 */
public class DummyCafe24Server {

    private final HttpServer server;
    private final int bucketCapacity;
    private final int leakRatePerSecond;
    private final AtomicInteger currentBucketLevel = new AtomicInteger(0);
    private final ScheduledExecutorService leakScheduler;

    public DummyCafe24Server(int port, int bucketCapacity, int leakRatePerSecond) throws IOException {
        this.bucketCapacity = bucketCapacity;
        this.leakRatePerSecond = leakRatePerSecond;

        // Leaky Bucket 드레인: 매초 leakRate만큼 버킷 수위 감소
        this.leakScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
            Thread t = new Thread(r, "leaky-bucket-drain");
            t.setDaemon(true);
            return t;
        });
        this.leakScheduler.scheduleAtFixedRate(
                () -> currentBucketLevel.updateAndGet(c -> Math.max(0, c - leakRatePerSecond)),
                1, 1, TimeUnit.SECONDS
        );

        this.server = HttpServer.create(new InetSocketAddress(port), 0);
        this.server.createContext("/api/echo", this::handleEcho);
        this.server.setExecutor(Executors.newFixedThreadPool(4));
    }

    public void start() {
        server.start();
    }

    public void stop() {
        server.stop(0);
        leakScheduler.shutdown();
    }

    /** 테스트 간 버킷 상태 초기화 */
    public void reset() {
        currentBucketLevel.set(0);
    }

    public int getCurrentBucketLevel() {
        return currentBucketLevel.get();
    }

    public int getBucketCapacity() {
        return bucketCapacity;
    }

    private void handleEcho(HttpExchange exchange) throws IOException {
        if (!"GET".equals(exchange.getRequestMethod())) {
            exchange.sendResponseHeaders(405, -1);
            return;
        }

        int level = currentBucketLevel.incrementAndGet();

        if (level > bucketCapacity) {
            // 버킷 초과 - 429 반환 (실제 소비하지 않으므로 감소)
            currentBucketLevel.decrementAndGet();

            double usagePercent = ((double) level / bucketCapacity) * 100;
            int remainSeconds = (int) Math.ceil((double) (level - bucketCapacity) / leakRatePerSecond) + 1;

            exchange.getResponseHeaders().add("X-Api-Call-Limit", level + "/" + bucketCapacity);
            exchange.getResponseHeaders().add("X-Cafe24-Call-Usage", String.format("%.1f", usagePercent));
            exchange.getResponseHeaders().add("X-Cafe24-Call-Remain", String.valueOf(remainSeconds));

            byte[] body = "Too Many Requests".getBytes(StandardCharsets.UTF_8);
            exchange.sendResponseHeaders(429, body.length);
            try (OutputStream os = exchange.getResponseBody()) {
                os.write(body);
            }
            return;
        }

        // 파라미터 파싱
        String word = parseWordParam(exchange.getRequestURI().getQuery());

        // "hello" → "world", 나머지는 그대로 에코
        String result = "hello".equals(word) ? "world" : word;

        // 응답 헤더 설정
        double usagePercent = ((double) level / bucketCapacity) * 100;
        exchange.getResponseHeaders().add("X-Api-Call-Limit", level + "/" + bucketCapacity);
        exchange.getResponseHeaders().add("X-Cafe24-Call-Usage", String.format("%.1f", usagePercent));
        exchange.getResponseHeaders().add("Content-Type", "text/plain; charset=utf-8");

        byte[] responseBytes = result.getBytes(StandardCharsets.UTF_8);
        exchange.sendResponseHeaders(200, responseBytes.length);
        try (OutputStream os = exchange.getResponseBody()) {
            os.write(responseBytes);
        }
    }

    private String parseWordParam(String query) {
        if (query == null) return "";
        for (String param : query.split("&")) {
            String[] pair = param.split("=", 2);
            if (pair.length == 2 && "word".equals(pair[0])) {
                return URLDecoder.decode(pair[1], StandardCharsets.UTF_8);
            }
        }
        return "";
    }
}

package com.example.cafe24.actor;

import akka.actor.AbstractActor;
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import akka.stream.OverflowStrategy;
import akka.stream.SystemMaterializer;
import akka.stream.ThrottleMode;
import akka.stream.javadsl.Sink;
import akka.stream.javadsl.Source;
import com.example.cafe24.message.Messages.ApiRequest;
import com.example.cafe24.message.Messages.ApiResponse;
import com.example.cafe24.message.Messages.StreamEnvelope;
import com.example.cafe24.message.Messages.StreamResult;
import scala.concurrent.duration.FiniteDuration;

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;

/**
 * Cafe24 API를 백프레셔를 적용하여 안전하게 호출하는 액터.
 *
 * <p>내부에 Akka Streams 파이프라인(유압장치)을 구성하여:
 * <ul>
 *   <li>Source.actorRef: 요청 유입구 (밸브)</li>
 *   <li>throttle: 유량 조절기 (초당 maxRequestsPerSecond 제한)</li>
 *   <li>mapAsync: 비동기 HTTP 호출 펌프 (블로킹 없음)</li>
 *   <li>Sink.foreach: 응답 배출구 (원래 요청자에게 전달)</li>
 * </ul>
 *
 * <p>사용자는 {@link ApiRequest}를 보내고 {@link ApiResponse}를 받기만 하면 됩니다.
 * 호출 제약(백프레셔, TPS, 429 재시도)은 이 액터가 모두 처리합니다.
 *
 * <p><b>재사용 방법</b>: apiBaseUrl과 maxRequestsPerSecond를 설정하여
 * 다양한 Cafe24 API 엔드포인트에 대해 독립적인 안전 호출 장치를 생성할 수 있습니다.
 */
public class SafeApiCallerActor extends AbstractActor {

    private final LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this);
    private final String apiBaseUrl;
    private final int maxRequestsPerSecond;
    private final HttpClient httpClient;
    private ActorRef streamEntry;

    public SafeApiCallerActor(String apiBaseUrl, int maxRequestsPerSecond) {
        this.apiBaseUrl = apiBaseUrl;
        this.maxRequestsPerSecond = maxRequestsPerSecond;
        this.httpClient = HttpClient.newBuilder()
                .connectTimeout(Duration.ofSeconds(5))
                .build();
    }

    public static Props props(String apiBaseUrl, int maxRequestsPerSecond) {
        return Props.create(SafeApiCallerActor.class, apiBaseUrl, maxRequestsPerSecond);
    }

    @Override
    public void preStart() {
        var materializer = SystemMaterializer.get(getContext().getSystem()).materializer();

        // Akka Streams 백프레셔 파이프라인 구성
        // Source.actorRef(유입) → throttle(유량제한) → mapAsync(비동기호출) → Sink(응답전달)
        streamEntry = Source.<StreamEnvelope>actorRef(100, OverflowStrategy.dropNew())
                .throttle(
                        maxRequestsPerSecond,
                        FiniteDuration.create(1, TimeUnit.SECONDS),
                        maxRequestsPerSecond,
                        ThrottleMode.shaping()
                )
                .mapAsync(1, this::callApiWithAdaptiveBackpressure)
                .to(Sink.foreach(this::deliverResponse))
                .run(materializer);

        log.info("SafeApiCallerActor 초기화 완료 - baseUrl={}, maxRps={}", apiBaseUrl, maxRequestsPerSecond);
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(ApiRequest.class, request -> {
                    // 요청을 스트림 파이프라인에 투입 (논블로킹)
                    streamEntry.tell(new StreamEnvelope(request), getSelf());
                })
                .build();
    }

    /**
     * 적응형 백프레셔를 적용한 API 호출.
     * 응답 헤더의 버킷 사용률에 따라 추가 지연을 동적으로 적용합니다.
     */
    private CompletionStage<StreamResult> callApiWithAdaptiveBackpressure(StreamEnvelope envelope) {
        return executeHttpCall(envelope, 0)
                .exceptionally(ex -> {
                    log.error("API 호출 실패: {}", ex.getMessage());
                    return new StreamResult(
                            envelope.request(),
                            new ApiResponse(envelope.request().word(), "ERROR: " + ex.getMessage(), 500, 0, 0)
                    );
                });
    }

    private CompletionStage<StreamResult> executeHttpCall(StreamEnvelope envelope, int retryCount) {
        String word = envelope.request().word();
        HttpRequest request = HttpRequest.newBuilder()
                .uri(URI.create(apiBaseUrl + "/api/echo?word=" + word))
                .timeout(Duration.ofSeconds(5))
                .GET()
                .build();

        return httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofString())
                .thenCompose(response -> {
                    int statusCode = response.statusCode();

                    if (statusCode == 429 && retryCount < 3) {
                        // Rate limited - X-Cafe24-Call-Remain 헤더 기반 대기 후 재시도
                        String remainStr = response.headers()
                                .firstValue("X-Cafe24-Call-Remain").orElse("1");
                        int waitSeconds = Integer.parseInt(remainStr);
                        log.warning("429 응답 수신 - {}초 후 재시도 (retry={})", waitSeconds, retryCount + 1);

                        return delay(waitSeconds * 1000L)
                                .thenCompose(v -> executeHttpCall(envelope, retryCount + 1));
                    }

                    // 응답 헤더에서 버킷 상태 파싱
                    String callLimit = response.headers()
                            .firstValue("X-Api-Call-Limit").orElse("0/10");
                    String[] parts = callLimit.split("/");
                    int bucketUsed = Integer.parseInt(parts[0]);
                    int bucketMax = Integer.parseInt(parts[1]);
                    double usageRatio = (double) bucketUsed / bucketMax;

                    StreamResult result = new StreamResult(
                            envelope.request(),
                            new ApiResponse(word, response.body(), statusCode, bucketUsed, bucketMax)
                    );

                    // 적응형 백프레셔: 버킷 사용률 기반 추가 지연
                    // 유압장치의 압력 게이지를 읽고 밸브를 조절하는 것과 같은 원리
                    long adaptiveDelayMs = 0;
                    if (usageRatio > 0.8) {
                        adaptiveDelayMs = 500;
                        log.info("버킷 사용률 {}% - 500ms 추가 지연 적용", (int) (usageRatio * 100));
                    } else if (usageRatio > 0.5) {
                        adaptiveDelayMs = 200;
                    }

                    if (adaptiveDelayMs > 0) {
                        return delay(adaptiveDelayMs).thenApply(v -> result);
                    }
                    return CompletableFuture.completedFuture(result);
                });
    }

    /** 스트림 출구: 완료된 응답을 원래 요청자에게 전달 */
    private void deliverResponse(StreamResult result) {
        ActorRef replyTo = result.originalRequest().replyTo();
        replyTo.tell(result.response(), getSelf());
    }

    /** 논블로킹 지연 (Thread.sleep 대신 delayedExecutor 사용) */
    private CompletionStage<Void> delay(long millis) {
        return CompletableFuture.supplyAsync(
                () -> null,
                CompletableFuture.delayedExecutor(millis, TimeUnit.MILLISECONDS)
        );
    }
}