Flowable<ChatCompletionChunk> streamedChatCompletion = openAiService.streamChatCompletion(requestBody);

AssistantMessage accumulatedMessage = openAiService.mapStreamToAccumulatorWrapper(streamedChatCompletion)
        .doOnNext((chatMessageAccumulatorWrapper -> {
            ChatMessageAccumulator chatMessageAccumulator = chatMessageAccumulatorWrapper.getChatMessageAccumulator();
            if (!chatMessageAccumulator.isFunctionCall()) {
                ChatCompletionChunk chatCompletionChunk = chatMessageAccumulatorWrapper.getChatCompletionChunk();
                String source = chatCompletionChunk.getSource();
                byte[] bytes = source.getBytes(StandardCharsets.UTF_8);
                if (null != bytes) {
                    sseEmitter.send(SseEmitter.event().data(bytes, MediaType.APPLICATION_JSON_UTF8));
                }
            }
        }))
        .doOnError(throwable -> {
            sseEmitter.completeWithError(throwable);
            log.error("streamChatCompletion error", throwable);
        })
        .lastElement()
        .blockingGet()
        .getChatMessageAccumulator()
        .getAccumulatedMessage();