Reactive Programming ‐ WebFlux Patterns with Spring Boot - thought-corner/Backend-PlayGround GitHub Wiki

@RestController
public class ProductController {

    @GetMapping("/products/first")
    public Mono<Product> getFirst() {
        return Mono.just(new Product("1", "노트북", 1500000));
    }

    @GetMapping("/products")
    public Flux<Product> getAll() {
        return Flux.just(
                new Product("1", "노트북", 1500000),
                new Product("2", "마우스", 50000),
                new Product("3", "키보드", 120000));
    }

    @GetMapping("/products/available")
    public Mono<ResponseEntity<Product>> getAvailable() {
        // 예시로 DB를 조회하고, 여기서 재고가 있으면 true, 없다면 false
        boolean inStock = true;

        Mono<Product> found = inStock
                ? Mono.just(new Product("1", "노트북", 1500))
                : Mono.empty();

        return found
                .map(ResponseEntity::ok)
                .switchIfEmpty(Mono.just(ResponseEntity.notFound().build()));
    }

}

@RestController
public class SseController {

    @GetMapping(value = "/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> streamEvents() {
        return Flux.interval(Duration.ofSeconds(1)).map(tick -> "이벤트 #" + tick);
    }

    /*
     * 1. data : 보내는 데이터
     * 2. id : 이벤트에 대한 고유 ID ( 클라이언트 재연결 -> Last Event ID 헤더로 전달)
     * 3. event : 이벤트 타입에 대한 이름
     * 4. retry : 재연결에 대한 대기 시간
     */
}

public interface WebSocketHandler {

    /**
     * 이 핸들러가 지원하는 서브 프로토콜 목록을 반환합니다.
     * 핸드셰이크 시 클라이언트가 요청한 서브 프로토콜과 협상하는 데 사용됩니다.
     * 기본값은 빈 리스트(서브 프로토콜 미사용)입니다.
     *
     * @return 지원하는 서브 프로토콜 목록
     */
    default List<String> getSubProtocols() {
        return Collections.emptyList();
    }

    /**
     * WebSocket 세션을 처리합니다.
     * 반환된 {@link Mono}가 완료되면 세션이 종료됩니다.
     * 에러로 완료될 경우 세션은 비정상 종료됩니다.
     *
     * @param session 연결된 WebSocket 세션
     * @return 세션 처리가 완료되면 완료되는 {@link Mono}
     */
    Mono<Void> handle(WebSocketSession session);
}

@Component
public class TimeWebSocketHandler implements WebSocketHandler {

    @Override
    public Mono<Void> handle(WebSocketSession session) {

        // 클라이언트로 1초마다 시간 전송
        Flux<WebSocketMessage> serverToClient = Flux.interval(Duration.ofSeconds(1))
                .map(tick -> session.textMessage("서버 시간 : " + LocalDateTime.now().toString()));

        // 클라이언트에서 오는 메시지도 수신(소비만 하고, 종료 신호를 전달)
        Mono<Void> clientToServer = session.receive()
                .doOnNext(msg -> System.out.println("클라이언트: " + msg.getPayloadAsText()))
                .then();

        // 두 스트림을 함께 실행
        return Mono.zip(session.send(serverToClient), clientToServer).then();

        /*
         * 1. session.send -> Flux.interval 기반, 무한 스트림, 자체적으로 완료되지는 않음
         * 2. clientToServer -> session.receive() 기반, 클라이언트가 끊을 떄 완료가 됨
         */
    }

}