Spring WebFlux & Reactive Programming

// The raw contract (you rarely implement this directly)
public interface Publisher<T> {
    void subscribe(Subscriber<? super T> s);
}

public interface Subscriber<T> {
    void onSubscribe(Subscription s);
    void onNext(T t);
    void onError(Throwable t);
    void onComplete();
}

public interface Subscription {
    void request(long n);  // Backpressure signal
    void cancel();
}

Mono<String> name = userMono.map(User::getName);
Flux<Integer> lengths = namesFlux.map(String::length);

// Mono flatMap: Mono<A> -> (A -> Mono<B>) -> Mono<B>
Mono<Profile> profile = userMono
    .flatMap(user -> profileService.fetchProfile(user.getId()));

// Flux flatMap: subscribes eagerly, interleaves results (NO ordering guarantee)
Flux<Order> orders = userFlux
    .flatMap(user -> orderService.getOrders(user.getId()), 8); // concurrency = 8

// Like Promise.all() — fire concurrently, wait for all results
Mono<DashboardDto> dashboard = Mono.zip(
    userService.findById(id),
    orderService.countByUserId(id),
    notificationService.unreadCount(id)
).map(tuple -> new DashboardDto(tuple.getT1(), tuple.getT2(), tuple.getT3()));

// Like listening to 3 radio stations at once — you hear whichever speaks first
Flux<Event> allEvents = Flux.merge(
    orderEvents,    // may emit first
    paymentEvents,  // may emit first
    shippingEvents
);

// Like a playlist — song 2 starts only after song 1 finishes
Flux<User> allUsers = Flux.concat(
    cacheService.getCachedUsers(),      // Try cache first
    databaseService.getAllUsers()        // Then database
);

Mono<User> user = userRepository.findByEmail(email)
    .switchIfEmpty(Mono.error(new UserNotFoundException(email)));

Mono<Response> resilient = webClient.get()
    .uri("/api/prices")
    .retrieve()
    .bodyToMono(Response.class)
    .timeout(Duration.ofSeconds(3))                           // Fail fast
    .retryWhen(Retry.backoff(3, Duration.ofMillis(500))      // Exponential backoff
        .maxBackoff(Duration.ofSeconds(5))
        .filter(ex -> ex instanceof WebClientResponseException.ServiceUnavailable));

// flatMapMany: Mono -> Flux
Flux<Order> orders = userMono.flatMapMany(user -> orderService.streamOrders(user.getId()));

// concatMap: like flatMap but sequential (preserves order)
Flux<EnrichedOrder> enriched = orderFlux
    .concatMap(order -> enrichmentService.enrich(order)); // one at a time

// collectList: Flux -> Mono<List>
Mono<List<User>> userList = userFlux.collectList();

// distinct / take / skip
Flux<String> unique = namesFlux.distinct();
Flux<Order> firstTen = orderFlux.take(10);

// delayElements: throttle emission rate (useful for rate limiting)
Flux<Event> throttled = eventFlux.delayElements(Duration.ofMillis(100));

// IoT sensors emit 10,000 readings/sec. Your DB can write 500/sec.
public Flux<SensorReading> processSensorData(Flux<SensorReading> sensorFeed) {
    return sensorFeed
        .onBackpressureLatest()             // Keep latest if overwhelmed
        .limitRate(500)                     // Request 500 at a time
        .buffer(50)                         // Batch into groups of 50
        .flatMap(batch -> writeBatchToDb(batch), 4)  // 4 concurrent DB writes
        .doOnDiscard(SensorReading.class, reading ->
            metrics.increment("sensor.readings.dropped"));
}

@RestController
@RequestMapping("/api/users")
@RequiredArgsConstructor
public class UserController {

    private final UserService userService;

    @GetMapping("/{id}")
    public Mono<ResponseEntity<UserDto>> getUser(@PathVariable Long id) {
        return userService.findById(id)
            .map(user -> ResponseEntity.ok(UserDto.from(user)))
            .defaultIfEmpty(ResponseEntity.notFound().build());
    }

    @GetMapping
    public Flux<UserDto> getAllUsers() {
        return userService.findAll().map(UserDto::from);
    }

    @PostMapping
    @ResponseStatus(HttpStatus.CREATED)
    public Mono<UserDto> createUser(@Valid @RequestBody Mono<CreateUserRequest> request) {
        return request.flatMap(userService::create).map(UserDto::from);
    }

    @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<UserDto> streamUsers() {
        return userService.streamAllUsers().map(UserDto::from);
    }
}

@Configuration
public class UserRouter {

    @Bean
    public RouterFunction<ServerResponse> userRoutes(UserHandler handler) {
        return RouterFunctions.route()
            .path("/api/users", builder -> builder
                .GET("/{id}", handler::getUser)
                .GET("", handler::getAllUsers)
                .POST("", handler::createUser)
            )
            .filter((request, next) -> {
                log.info("Request: {} {}", request.method(), request.path());
                return next.handle(request);
            })
            .build();
    }
}

@Component
@RequiredArgsConstructor
public class UserHandler {

    private final UserService userService;

    public Mono<ServerResponse> getUser(ServerRequest request) {
        Long id = Long.parseLong(request.pathVariable("id"));
        return userService.findById(id)
            .flatMap(user -> ServerResponse.ok()
                .contentType(MediaType.APPLICATION_JSON)
                .bodyValue(UserDto.from(user)))
            .switchIfEmpty(ServerResponse.notFound().build());
    }

    public Mono<ServerResponse> getAllUsers(ServerRequest request) {
        return ServerResponse.ok()
            .contentType(MediaType.APPLICATION_JSON)
            .body(userService.findAll().map(UserDto::from), UserDto.class);
    }

    public Mono<ServerResponse> createUser(ServerRequest request) {
        return request.bodyToMono(CreateUserRequest.class)
            .flatMap(userService::create)
            .flatMap(user -> ServerResponse.status(HttpStatus.CREATED)
                .bodyValue(UserDto.from(user)));
    }
}

@Table("users")
public record User(
    @Id Long id,
    String email,
    String name,
    Instant createdAt
) {}

public interface UserRepository extends ReactiveCrudRepository<User, Long> {

    Mono<User> findByEmail(String email);

    @Query("SELECT * FROM users WHERE created_at > :since ORDER BY created_at DESC")
    Flux<User> findRecentUsers(Instant since);
}

public interface ProductRepository extends ReactiveMongoRepository<Product, String> {

    Flux<Product> findByCategory(String category);

    @Tailable  // For capped collections — streams new documents as they arrive
    Flux<Product> findWithTailableCursorBy();
}

@Bean
public WebClient paymentServiceClient() {
    return WebClient.builder()
        .baseUrl("https://payment-service.internal")
        .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
        .filter(ExchangeFilterFunctions.basicAuthentication("user", "secret"))
        .codecs(config -> config.defaultCodecs().maxInMemorySize(2 * 1024 * 1024))
        .build();
}

// GET — single item
public Mono<User> fetchUser(String userId) {
    return webClient.get()
        .uri("/api/users/{id}", userId)
        .retrieve()
        .bodyToMono(User.class);
}

// GET — streaming multiple items
public Flux<User> searchUsers(String name) {
    return webClient.get()
        .uri(b -> b.path("/api/users").queryParam("name", name).build())
        .retrieve()
        .bodyToFlux(User.class);
}

// POST with error handling + retry
public Mono<PaymentResponse> charge(PaymentRequest request) {
    return webClient.post()
        .uri("/api/payments/charge")
        .bodyValue(request)
        .retrieve()
        .onStatus(HttpStatusCode::is4xxClientError, resp ->
            resp.bodyToMono(ErrorResponse.class)
                .flatMap(err -> Mono.error(new ValidationException(err.getMessage()))))
        .onStatus(HttpStatusCode::is5xxServerError, resp ->
            Mono.error(new ServiceUnavailableException()))
        .bodyToMono(PaymentResponse.class)
        .timeout(Duration.ofSeconds(5))
        .retryWhen(Retry.backoff(3, Duration.ofMillis(500))
            .filter(ex -> ex instanceof ServiceUnavailableException));
}

@GetMapping(value = "/api/stocks/ticker", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<StockPrice>> streamPrices() {
    return Flux.interval(Duration.ofMillis(500))
        .map(i -> {
            String symbol = SYMBOLS[random.nextInt(SYMBOLS.length)];
            BigDecimal price = BigDecimal.valueOf(100 + random.nextDouble() * 200);
            return ServerSentEvent.<StockPrice>builder()
                .id(String.valueOf(i))
                .event("price-update")
                .data(new StockPrice(symbol, price, Instant.now()))
                .retry(Duration.ofSeconds(5))
                .build();
        });
}

@Configuration
@EnableWebFlux
public class WebSocketConfig {

    @Bean
    public HandlerMapping webSocketMapping(ChatWebSocketHandler handler) {
        Map<String, WebSocketHandler> map = Map.of("/ws/chat", handler);
        SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
        mapping.setUrlMap(map);
        mapping.setOrder(-1);
        return mapping;
    }
}

@Component
public class ChatWebSocketHandler implements WebSocketHandler {

    private final Sinks.Many<String> sink = Sinks.many().multicast().onBackpressureBuffer();

    @Override
    public Mono<Void> handle(WebSocketSession session) {
        // Receive messages from client → broadcast to all
        Mono<Void> input = session.receive()
            .map(WebSocketMessage::getPayloadAsText)
            .doOnNext(msg -> sink.tryEmitNext(msg))
            .then();

        // Send broadcast messages to this client
        Mono<Void> output = session.send(
            sink.asFlux().map(session::textMessage));

        return Mono.zip(input, output).then();
    }
}

public Mono<Order> placeOrder(OrderRequest request) {
    return validateOrder(request)
        .flatMap(this::saveOrder)
        .flatMap(order -> paymentClient.charge(order)
            .map(payment -> order.withPaymentId(payment.getId()))
            .onErrorResume(PaymentUnavailableException.class, ex -> {
                log.error("Payment down, queuing order: {}", order.getId());
                return queueForRetry(order).thenReturn(order.withStatus(PENDING));
            })
        )
        .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
            .maxBackoff(Duration.ofSeconds(10))
            .filter(ex -> ex instanceof TransientException))
        .doOnError(ex -> metrics.increment("order.failed"))
        .onErrorMap(ex -> new OrderProcessingException("Order failed", ex));
}

@Test
void findById_existingUser_returnsUser() {
    User user = new User(1L, "alice@test.com", "Alice", Instant.now());
    when(userRepository.findById(1L)).thenReturn(Mono.just(user));

    StepVerifier.create(userService.findById(1L))
        .assertNext(result -> assertThat(result.email()).isEqualTo("alice@test.com"))
        .verifyComplete();
}

@Test
void findById_nonExistent_completesEmpty() {
    when(userRepository.findById(99L)).thenReturn(Mono.empty());

    StepVerifier.create(userService.findById(99L))
        .verifyComplete();  // No items, just onComplete
}

@Test
void create_duplicateEmail_emitsError() {
    when(userRepository.findByEmail("alice@test.com")).thenReturn(Mono.just(existingUser));

    StepVerifier.create(userService.create(new CreateUserRequest("alice@test.com", "X")))
        .expectError(DuplicateEmailException.class)
        .verify();
}

@Test
void retryWithBackoff_respectsDelay() {
    StepVerifier.withVirtualTime(() ->
            failingService.call()
                .retryWhen(Retry.backoff(3, Duration.ofSeconds(2)))
        )
        .expectSubscription()
        .thenAwait(Duration.ofSeconds(2))   // First retry
        .thenAwait(Duration.ofSeconds(4))   // Second retry (exponential)
        .expectNext(expectedResult)
        .verifyComplete();
}

@SpringBootTest(webEnvironment = RANDOM_PORT)
class UserControllerIT {

    @Autowired WebTestClient webTestClient;

    @Test
    void getUser_returns200() {
        webTestClient.get().uri("/api/users/1")
            .exchange()
            .expectStatus().isOk()
            .expectBody(UserDto.class)
            .value(user -> assertThat(user.email()).isEqualTo("alice@test.com"));
    }

    @Test
    void streamUsers_returnsSSE() {
        webTestClient.get().uri("/api/users/stream")
            .accept(MediaType.TEXT_EVENT_STREAM)
            .exchange()
            .expectStatus().isOk()
            .returnResult(UserDto.class)
            .getResponseBody()
            .as(StepVerifier::create)
            .expectNextCount(3)
            .thenCancel()
            .verify();
    }
}

// WRONG: blocks the event loop
Mono<User> user = Mono.fromCallable(() -> jdbcTemplate.queryForObject(...));

// RIGHT: offload to bounded elastic scheduler
Mono<User> user = Mono.fromCallable(() -> jdbcTemplate.queryForObject(...))
    .subscribeOn(Schedulers.boundedElastic());

// WRONG: IllegalStateException on Netty thread, or deadlock
@GetMapping("/user/{id}")
public Mono<User> getUser(@PathVariable Long id) {
    User user = userRepository.findById(id).block(); // NEVER
    return Mono.just(user);
}

// WRONG: this does NOTHING — no subscriber
userRepository.save(user);

// RIGHT: return it so the framework subscribes
return userRepository.save(user);

// Development: enable debug agent (bytecode instrumentation, lower overhead)
ReactorDebugAgent.init(); // Call once at startup

// Targeted debugging: checkpoint()
Flux<Order> orders = orderService.getOrders()
    .checkpoint("after-getOrders")
    .filter(Order::isPaid)
    .checkpoint("after-filter-paid");

// Write context at subscription point
Mono<User> result = userService.findById(id)
    .contextWrite(ctx -> ctx.put("traceId", traceId));

// Read context inside the chain
Mono<String> trace = Mono.deferContextual(ctx -> Mono.just(ctx.get("traceId")));