Reactive Programming ‐ Practical WebFlux Patterns with Spring Boot & Building Applications with Spring Boot and WebFlux - thought-corner/backend-roadmap GitHub Wiki

@RestController
public class ProductController {

    // 단일 객체: Mono<T>
    @GetMapping("/products/first")
    public Mono<Product> getFirst() {
        return Mono.just(new Product("1", "노트북", 1_500_000));
    }

    // 여러 객체: Flux<T> → 기본적으로 JSON 배열로 직렬화되어 응답
    @GetMapping("/products")
    public Flux<Product> getAll() {
        return Flux.just(
                new Product("1", "노트북", 1_500_000),
                new Product("2", "마우스", 50_000),
                new Product("3", "키보드", 120_000));
    }

    // 데이터 유무로 상태 코드 분기: 있으면 200, 없으면 404
    @GetMapping("/products/available")
    public Mono<ResponseEntity<Product>> getAvailable() {
        return findAvailableProduct()                            // Mono<Product> (없으면 empty)
                .map(ResponseEntity::ok)                         // 있으면 200 OK
                .defaultIfEmpty(ResponseEntity.notFound().build());  // 비었으면 404
    }

    // 실제로는 repository.findById(...) 등 논블로킹 조회. 여기선 재고 유무만 흉내
    private Mono<Product> findAvailableProduct() {
        boolean inStock = true;
        return inStock ? Mono.just(new Product("1", "노트북", 1_500_000)) : Mono.empty();
    }
}

@RestController
public class SseController {

    // ⚠️ 원본은 Flux<String>이라 id/event/retry 필드를 못 쓴다.
    //    ServerSentEvent로 반환해야 아래 4개 필드를 실제로 활용할 수 있다.
    @GetMapping(value = "/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> streamEvents() {
        return Flux.interval(Duration.ofSeconds(1))
                .map(tick -> ServerSentEvent.<String>builder()
                        .id(String.valueOf(tick))       // 이벤트 고유 ID (재연결 시 Last-Event-ID로 전달)
                        .event("time")                  // 이벤트 타입 이름
                        .data("이벤트 #" + tick)         // 실제 데이터
                        .retry(Duration.ofSeconds(3))   // 클라이언트 재연결 대기 시간
                        .build())
                .doOnCancel(() -> log.info("클라이언트 연결 종료 → interval 정리"));
    }
}

// 핸들러 인터페이스
public interface WebSocketHandler {

    // 핸드셰이크 시 협상할 서브 프로토콜 목록 (기본: 미사용)
    default List<String> getSubProtocols() {
        return Collections.emptyList();
    }

    // WebSocket 세션 처리. 반환된 Mono가 완료되면 세션 종료, 에러면 비정상 종료.
    Mono<Void> handle(WebSocketSession session);
}

// 구현 — 서버 시간 push + 클라이언트 메시지 수신
@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()));

        // 클라이언트 → 서버: 수신 로그만 남기고 완료 신호(then)만 전달
        Mono<Void> clientToServer = session.receive()
                .doOnNext(msg -> log.info("클라이언트: {}", msg.getPayloadAsText()))
                .then();

        // 두 방향을 함께 실행 — 한쪽이 끝나면 zip이 종료되며 세션이 닫힌다
        return Mono.zip(session.send(serverToClient), clientToServer).then();
    }
}

// 여러 구독자에게 이벤트를 뿌리는 hot 스트림
private final Sinks.Many<String> sink = Sinks.many().multicast().onBackpressureBuffer();

// 어딘가에서 이벤트 발생 시 push
public void publish(String event) {
    sink.tryEmitNext(event);
}

// SSE 컨트롤러는 이 sink를 구독
@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> stream() {
    return sink.asFlux();   // 여러 클라이언트가 같은 이벤트 흐름을 공유(hot)
}