• 클라이언트에서 대기하는 시간이 길다면 실시간성이 떨어지고, 대기하는 시간이 짧다면 서버에 부담이 간다.

• 요청과 응답 모두 독립적이기 때문에 header를 모두 포함합니다.
  --> 원래는 하나의 http 응답입니다. 공통되는 요소를 반복해서 보내야하기 때문에 오버헤드가 발생합니다.
• 클라이언트와 서버 모두 TCP/IP 연결을 연상태로 대기합니다.
  -> 한정된 커넥션 풀과 관련된 리소스를 신경써야합니다.
• 클라이언트에게 제공할 이벤트가 큐에 쌓이면 각각의 이벤트를 단건으로 여러개의 long poll 요청에 나눠서 전달해야합니다.
• 브라우저, gateway 등 다른 구성요소의 timeout을 고려하여 대기 시간을 설정해야합니다.

• HTTP/1.1또는 HTTP/2를 사용할 수 있습니다

• HTTP/2의 경우 멀티플렉싱을 이용할 수 있습니다.

• 클라이언트가 서버에 요청을 보냅니다.

• 서버가 전달할 이벤트, 데이터 등이 있다면 응답의 일부분을 전달합니다.

• 요청이나 연결을 닫지 않고 이벤트, 데이터를 전달할 때까지 대기합니다.

동적으로 content를 생성하는 경우 정확한 Content-Length 를 미리 제공할 수 없기 때문에 아래의 방식으로 HTTP Streaming을 구현합니다.

관련된 기술로서 가장 먼저 생각나는 것은 웹소켓입니다. Websocket이 아닌, SSE를 선택한 이유는 크게 두가지였습니다.

웹소켓을 구현할 경우 고려하고 관리해야하는 범위가 늘어납니다.

1. 웹소켓 핸드 셰이크를 위한 config 클래스
2. stomp 환경에서 작동할 수 있는 메시징 브로커 그 자체
   1. 메시징 브로커와 관련된 Config 클래스들
3. 채팅 publish와 로그 저장, 채팅 로그 조회 관련된 클래스

하지만 SSE를 활용하여 구현할 경우 Websocket보다는 고려하고 관리해야하는 범위가 좁습니다.

1. 핸드 셰이크가 필요없이 단일한 REST API 하나를 이용해서 응답을 주고 받습니다.
2. 메시징 브로커를 쓰지 않습니다.
3. 채팅 publish를 할 필요가 없으며 로그 저장 및 채팅 로그 조회 관련 클래스만 구현하면 됩니다,

0~N개의 데이터 항목을 비동기적으로 스트리밍하는 Publisher를 나타냅니다.
데이터 스트림을 처리하고 변환하는 다양한 연산자를 제공합니다.

Flux.create와 함께 사용하며, Flux의 데이터를 push 방식으로 제공할 수 있게끔합니다. 또한 Flux 스트림 내에서 데이터를 동적으로 생성하고 내보낼 수 있습니다.

• next, complete, error 메서드를 오버라이딩하여 데이터를 전송하거나 스트림의 완료, 에러 시 처리를 커스텀할 수 있습니다.

두 객체를 활용해 스트림을 동적으로 생성하고 데이터를 push할 수 있습니다.
Flux.create 메서드는 FluxSink를 인자로 받아 데이터를 스트림에 공급할 수 있는 Flux를 생성합니다.

클라이언트로 부터 유저 질의를 받아 실시간으로 생성되는 챗봇의 응답을 스트리밍하는 컨트롤러 입니다.

  

@PostMapping(produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> generateChat(@RequestBody ChatReq question,   
								 @AuthenticationPrincipal Member member) {

	
	return chatBotService.question(question, member.getId());

}

Copy

produces = MediaType.TEXT_EVENT_STREAM_VALUE
이 API의 반환 값이 text/event-stream MIME 타입임을 명시합니다.

Flux
스트리밍 방식으로 여러개의 문자를 전송하기 위해서 Project Reactor의 Flux 타입을 사용합니다.

Open AI의 Streaming Chat Model을 사용하여 실시간으로 데이터를 받아들이고, 이를 Reactive Streams의 Flux로 변환하여 반환합니다.

  

@Override
public Flux<String> generateStreamingChat(String template, 
										  Map<String, Object> variables, 
										  String userId, Chatbot chatbot) {

  

	Prompt prompt = getPrompt(template, variables);
	
	
	if (openAiStreamingChatModel == null) {
	openAiStreamingChatModel = OpenAiStreamingChatModel.builder()
								.apiKey(openAiKey)
								.modelName(GPT_3_5_TURBO)
								.build();
	}

  
  
  

	return Flux.create(emitter ->
			openAiStreamingChatModel.generate(prompt.text(), 
			new StreamingResponseHandler<>() {
			
			  
			
			@Override
			public void onNext(String token) {
				emitter.next(token);
			}
			
			  
			
			@Override
			public void onComplete(Response<AiMessage> response) {
				chatQueryService.saveChat(userId, 
										 . chatbot, 
										   . response.content().text(), 
											false
										);
			
				emitter.complete();
			}
			
			  
			  
			
			@Override
			public void onError(Throwable error) {	
				log.error("[OpenAiChatService generateStreamingChat] 에러 발생 ::{}", error);
				emitter.next("에러가 발생했습니다. 관리자에게 문의하세요.");
				emitter.complete();
	
			}));

}
Copy

Flux.create
Flux 스트림을 생성합니다. emitter는 FluxSink 객체로, 데이터 스트림을 내보낼 수 있습니다. Flux.create는 데이터를 생성하고 Flux.sink를 통해 비즈니스 로직을 제공받아 가공된 Flux를 생성합니다.

emitter
FluxSink 인터페이스의 인스턴스입니다. 데이터를 Flux 스트림으로 내보냅니다. next, complete, error 등의 메서드를 오버라이딩하여 스트림을 제어할 수 있습니다.

StreamingResponseHandler
Open AI의 스트리밍 응답을 처리하기 위해 해당 StreamingResponseHandler를 사용합니다. 이 핸들러를 사용하기 위해서는 onNext, onComplete, onError 3가지 메서드를 구현해야합니다.

onNext 메서드

실시간으로 생성되는 데이터를 수신했을 때 onNext 메서드를 호출합니다.
emitter.next(token) 을 호출해 받은 데이터를 Flux 스트림으로 전달합니다.

@Override
public void onNext(String token) {
	emitter.next(token);
}

Copy

onComplete 메서드

스트리밍이 완료되었을 때 호출합니다.

@Override
public void onComplete(Response<AiMessage> response) {
    chatQueryService.saveChat(userId, 
							  chatbot, 
                              response.content().text(), 
                              false
                             );

    emitter.complete();
}

Copy

1. 다음 접속시에 채팅 내역을 제공해야하기 때문에 Dynamo DB를 활용해 채팅 내용을 저장합니다.
2. emitter.complete() 를 호출하여 Flux 스트림을 활용합니다.

onError 메서드

스트리밍 중 에러가 발생했을 때 호출합니다.

@Override
public void onError(Throwable error) {	
	log.error("[OpenAiChatService generateStreamingChat] 에러 발생 ::{}", error);
	emitter.next("에러가 발생했습니다. 관리자에게 문의하세요.");
	emitter.complete();
}));

Copy

원래는 emitter.error() 를 활용해 Publisher에서 에러를 발생시켜야합니다. 하지만, 특정 에러 응답을 반환하기 보다는, 서버에서 에러 로깅 후 클라이언트 대화창에서 바로 에러메시지를 출력하기로 구현 스펙을 결정했기 때문에 다음과 같은 비즈니스 로직으로 구현하였습니다.

1. 에러를 로깅한다.
2. 대화 로그에 에러 발생 메시지를 포함시킨다.
3. 스트리밍을 종료한다.