느리고 큰 외부 API 응답, 스트리밍으로 처리하기

@Slf4j
@RequiredArgsConstructor
public abstract class AbstractApiClient {
    private final HttpClientExecutor httpClientExecutor;
 
    // 각 구현체가 반드시 제공해야 하는 것: 베이스 URL
    protected abstract String getBaseUrl();
 
    protected <R extends ExternalApiResponse, Q extends ExternalQueryParams> Mono<R> get(
            String uri, Class<R> responseType, Q queryParams) {
        final Map<String, Object> params = queryParams != null
                ? queryParams.toMap()
                : Collections.emptyMap();
 
        log.info("[API-GET] baseUrl={}, uri={}, params={}", getBaseUrl(), uri, params);
 
        return httpClientExecutor.get(getBaseUrl(), uri, responseType, params)
                .doOnSuccess(response -> log.info("[API-GET-SUCCESS] uri={}", uri))
                .doOnError(error -> log.error("[API-GET-ERROR] uri={}, error={}", uri, error.getMessage()))
                .onErrorMap(error -> new CoreException(
                        CommonErrorType.EXTERNAL_API_ERROR, error.getMessage(), error));
    }
 
    // POST 스트리밍 전용
    protected <T extends ExternalApiRequest> Flux<String> postStreaming(
            String uri, T requestBody) {
        log.info("[API-POST-STREAMING] baseUrl={}, uri={}", getBaseUrl(), uri);
 
        return httpClientExecutor.postStreaming(getBaseUrl(), uri, requestBody)
                .map(dataBuffer -> {
                    byte[] bytes = new byte[dataBuffer.readableByteCount()];
                    dataBuffer.read(bytes);
                    DataBufferUtils.release(dataBuffer);
                    return new String(bytes);
                })
                .doOnComplete(() -> log.info("[API-POST-STREAMING-COMPLETE] uri={}", uri))
                .doOnError(error -> log.error(
                        "[API-POST-STREAMING-ERROR] uri={}, error={}", uri, error.getMessage()));
    }
 
    protected <T extends ExternalApiRequest, R extends ExternalApiResponse> Mono<R> post(
            String uri, T requestBody, Class<R> responseType) {
        return httpClientExecutor.post(getBaseUrl(), uri, requestBody, responseType)
                .onErrorMap(error -> new CoreException(
                        CommonErrorType.EXTERNAL_API_ERROR, error.getMessage(), error));
    }
}

@Component
public class AiAnalysisApiClient extends AbstractApiClient {
 
    @Value("${ai.api.base-url}")
    private String baseUrl;
 
    public AiAnalysisApiClient(HttpClientExecutor httpClientExecutor) {
        super(httpClientExecutor);
    }
 
    @Override
    protected String getBaseUrl() {
        return baseUrl;
    }
 
    public Flux<String> streamAnalysis(AnalysisRequest request) {
        return postStreaming("/v1/analyze", request);
    }
}

return httpClientExecutor.postStreaming(getBaseUrl(), uri, requestBody)
    .map(dataBuffer -> {
        byte[] bytes = new byte[dataBuffer.readableByteCount()];
        dataBuffer.read(bytes);
        DataBufferUtils.release(dataBuffer);          // ← 이게 핵심
        return new String(bytes);
    })

.map(dataBuffer -> {
    // 1. DataBuffer에서 데이터를 byte[]로 복사
    byte[] bytes = new byte[dataBuffer.readableByteCount()];
    dataBuffer.read(bytes);
 
    // 2. 복사 후 즉시 release — 이 순서가 중요하다
    DataBufferUtils.release(dataBuffer);
 
    // 3. 복사한 byte[]로 String 생성
    return new String(bytes, StandardCharsets.UTF_8);
})

@Service
@RequiredArgsConstructor
public class DocumentAnalysisService {
 
    private final AiAnalysisApiClient aiApiClient;
    private final AnalysisResultRepository resultRepository;
 
    public AnalysisResult analyze(Document document) {
        // 스트리밍으로 수신하면서 StringBuilder에 누적
        StringBuilder accumulatedResult = new StringBuilder();
 
        aiApiClient.streamAnalysis(new AnalysisRequest(document.getContent()))
                .doOnNext(chunk -> {
                    // 청크 도착 시마다 실시간 처리 가능 (예: WebSocket으로 전달)
                    accumulatedResult.append(chunk);
                })
                .blockLast(); // MVC 환경이므로 최종 결과를 기다림
 
        // 누적된 결과를 파싱하고 DB에 저장
        AnalysisResult result = AnalysisResult.create(
                document.getId(),
                accumulatedResult.toString()
        );
        return resultRepository.save(result);
    }
}