Reactive Streaming

  Spring WebFlux는 SSE(Server-Sent Events)를 이용해 데이터를 Streaming 할 수 있습니다. SSE는 Spring 4.2 버전부터 지원되었으며, Spring 5 버전부터 Reactor의 Publisher 타입인 Flux를 이용해 조금 더 편리한 방법으로 SSE를 사용할 수 있게 되었습니다.

 

Note. SSE(Server-Sent Events)

  클라이언트가 HTTP 연결을 통해 서버로부터 전송되는 업데이트 데이터를 지속적으로 수신할 수 있는 단방향 서버 푸시 기술입니다. SSE는 주로 클라이언트 측에서 서버로부터 전송되는 이벤트 스트림을 자동으로 수신하기 위해 사용됩니다.

 

 다음은 Streaming으로 처리될 데이터를 데이터베이스에서 조회하는 BookService 클래스의 코드입니다. Spring WebFlux에서 Streaming 방식으로 데이터를 전송하기 위한 response body의 타입은 Flux입니다.

@Slf4j
@Validated
@Service
@RequiredArgsConstructor
public class BookService {
    private final @NonNull R2dbcEntityTemplate template;
    private final @NonNull CustomBeanUtils<Book> beanUtils;

    ...

    public Flux<Book> streamingBooks() {
        return template
                .select(Book.class)
                .all()
                .delayElements(Duration.ofSeconds(2L));
    }

    ...
}

 

  다음은 BookService에서 리턴되는 Flux<Book>을 response body로 전송하기 위한 BookRouter 코드입니다.

@Configuration
public class BookRouter {
    ...

    @Bean
    public RouterFunction<?> routeStreamingBook(BookService bookService,
                                                BookMapper mapper) {
        return route(RequestPredicates.GET("/v11/streaming-books"),
                request -> ServerResponse
                        .ok()
                        .contentType(MediaType.TEXT_EVENT_STREAM)
                        .body(bookService
                                        .streamingBooks()
                                        .map(book -> mapper.bookToResponse(book))
                                ,
                                BookDto.Response.class));
    }
}

• RequestPredicates.GET(): 클라이언트로부터 전송되는 request를 라우팅하기 위해 사용합니다. 파라미터로 지정한 request URI에 매치되는 request를 라우팅합니다.
• .contentType(MediaType.TEXT_EVENT_STREAM): SSE를 이용해 스트림을 클라이언트로 전송하기 위해서는 Content Type이 'text/event-stream'이어야 합니다.

 

  다음은 Reactive Streaming 데이터를 수신하기 위한 클라이언트 역할을 하는 BookWebClient 코드입니다.

@Slf4j
@Configuration
public class BookWebClient {
    @Bean
    public ApplicationRunner streamingBooks() {
        return (ApplicationArguments arguments) -> {
            WebClient webClient = WebClient.create("http://localhost:8080");
            Flux<BookDto.Response> response =
                    webClient
                            .get()
                            .uri("http://localhost:8080/v11/streaming-books")
                            .retrieve()
                            .bodyToFlux(BookDto.Response.class);

            response.subscribe(book -> {
                        log.info("bookId: {}", book.getBookId());
                        log.info("titleKorean: {}", book.getTitleKorean());
                        log.info("titleEnglish: {}", book.getTitleEnglish());
                        log.info("description: {}", book.getDescription());
                        log.info("author: {}", book.getAuthor());
                        log.info("isbn: {}", book.getIsbn());
                        log.info("publishDate: {}", book.getPublishDate());
                        log.info("=======================================");
                    },
                    error -> log.error("# error happened: ", error));
        };
    }
}

• bodyToFlux(): Streaming 데이터를 수신하기 위해 사용합니다. 파라미터로 수신한 데이터를 디코딩할 데이터 타입을 지정합니다.

 

Note. Reactive Streaming 데이터를 Javascript 기반의 웹 앱에서 수신하기 위해서는 Javascript API인 EventSource 객체를 이용하면 됩니다. 다음을 참고하시면 됩니다.

https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events

Using server-sent events - Web APIs | MDN

 

 

 

[참고 정보]

• 스프링으로 시작하는 리액티브 프로그래밍
• 전체 소스코드