Reactive Programming & WebFlux

Flux.range(1, 1_000_000)
    .onBackpressureBuffer(100)          // buffer overflow strategy — use carefully
    .limitRate(50)                      // prefetch cap — smoother consumer pacing
    .publishOn(Schedulers.boundedElastic())
    .subscribe(
        data -> process(data),          // slow consumer
        Throwable::printStackTrace,
        () -> System.out.println("done"),
        sub -> sub.request(10)          // explicit: pull 10 at a time
    );

Mono.just("hello");
Mono.empty();
Mono.error(new IllegalStateException("not found"));

Flux.just("a", "b", "c");
Flux.fromIterable(List.of(1, 2, 3));
Flux.interval(Duration.ofMillis(100)).take(5);   // tick stream

// Deferred — created per subscription (cold)
Mono.fromCallable(() -> repo.findById(id));

// map — 1:1 sync transform
Flux.just("alice", "bob")
    .map(String::toUpperCase);

// flatMap — 1:N async; inner publishers run concurrently (unordered by default)
Flux.just("user-1", "user-2")
    .flatMap(id -> userClient.fetchUser(id));   // Mono<User> per id

// flatMapSequential — preserve source order with limited concurrency
Flux.just("a", "b", "c")
    .flatMapSequential(this::slowLookup, 2);

// filter
Flux.range(1, 10).filter(n -> n % 2 == 0);

Mono<User> user = userRepo.findById(id);
Mono<Account> account = accountRepo.findByUserId(id);

// zip — pair latest from each when both emit (waits for slowest)
Mono.zip(user, account, (u, a) -> new Dashboard(u, a));

// merge — interleave as they arrive (hot fan-in)
Flux.merge(temperatureSensor(), humiditySensor());

// concat — strictly sequential: first publisher completes, then second
Flux.concat(page1(), page2(), page3());

orderRepo.findById(id)
    .switchIfEmpty(Mono.error(new OrderNotFoundException(id)));

configRepo.findFlag("feature-x")
    .defaultIfEmpty(false);

pricingClient.getRate("USD")
    .timeout(Duration.ofSeconds(2))
    .retryWhen(Retry.backoff(3, Duration.ofMillis(200))
        .filter(ex -> ex instanceof WebClientResponseException))
    .onErrorResume(WebClientResponseException.NotFound.class,
        ex -> Mono.just(FallbackRate.USD_DEFAULT))
    .onErrorMap(TimeoutException.class,
        ex -> new ServiceUnavailableException("pricing down"));

Mono<ExchangeRate> rate = fxClient.fetch("EUR")
    .cache(Duration.ofMinutes(5));   // Reactor 3.4+ timed cache

// Multiple subscribers within 5 min share one HTTP call
rate.subscribe(r -> chargeInEur(r));
rate.subscribe(r -> auditLog(r));

Sinks.Many<MetricEvent> sink = Sinks.many().multicast().onBackpressureBuffer();

public void emit(MetricEvent e) {
  sink.tryEmitNext(e);
}

public Flux<MetricEvent> stream() {
  return sink.asFlux();
}

Mono.fromCallable(() -> legacyBlockingApi())
    .subscribeOn(Schedulers.boundedElastic())  // subscription + source run on elastic
    .map(this::transform)
    .publishOn(Schedulers.parallel())        // downstream operators on parallel
    .subscribe(result -> { });

// ✅ JUnit test — main thread, no event loop
@Test
void findOrder() {
  Order order = orderService.findById(1L).block();
  assertThat(order.id()).isEqualTo(1L);
}

// ✅ Spring MVC controller bridging to reactive client (use sparingly)
@GetMapping("/sync-report")
Report syncReport() {
  return reportClient.fetch().block(Duration.ofSeconds(5));
}

// ❌ WebFlux controller — never block
@GetMapping("/orders/{id}")
Mono<Order> getOrder(@PathVariable Long id) {
  return orderService.findById(id);   // return Mono, don't .block()
}

@RestController
@RequestMapping("/api/orders")
class OrderController {

  private final OrderService orders;

  @GetMapping("/{id}")
  Mono<OrderDto> get(@PathVariable Long id) {
    return orders.findById(id).map(OrderDto::from);
  }

  @GetMapping
  Flux<OrderDto> list(@RequestParam(defaultValue = "0") int page) {
    return orders.findPage(page).map(OrderDto::from);
  }

  @PostMapping
  Mono<ResponseEntity<OrderDto>> create(@Valid @RequestBody Mono<CreateOrderRequest> body) {
    return body
        .flatMap(orders::create)
        .map(dto -> ResponseEntity.status(HttpStatus.CREATED).body(dto));
  }

  @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
  Flux<OrderEvent> stream() {
    return orders.eventStream();
  }
}

@Configuration
class OrderRoutes {

  @Bean
  RouterFunction<ServerResponse> routes(OrderHandler handler) {
    return RouterFunctions.route()
        .GET("/api/orders/{id}", handler::getOrder)
        .GET("/api/orders", handler::listOrders)
        .POST("/api/orders", handler::createOrder)
        .build();
  }
}

@Component
class OrderHandler {
  Mono<ServerResponse> getOrder(ServerRequest req) {
    Long id = Long.parseLong(req.pathVariable("id"));
    return orderService.findById(id)
        .map(dto -> ServerResponse.ok().bodyValue(dto))
        .switchIfEmpty(ServerResponse.notFound().build());
  }
}

@Bean
WebClient paymentClient(WebClient.Builder builder) {
  return builder
      .baseUrl("https://api.payments.example")
      .defaultHeader(HttpHeaders.ACCEPT, MediaType.APPLICATION_JSON_VALUE)
      .filter(ExchangeFilterFunction.ofRequestProcessor(req ->
          Mono.just(ClientRequest.from(req)
              .header("X-Request-Id", UUID.randomUUID().toString())
              .build())))
      .build();
}

// retrieve — idiomatic
Mono<PaymentResult> charge = client.post()
    .uri("/charges")
    .bodyValue(new ChargeRequest(100, "USD"))
    .retrieve()
    .onStatus(HttpStatusCode::is4xxClientError,
        resp -> resp.bodyToMono(ErrorBody.class)
            .map(body -> new PaymentRejectedException(body.message())))
    .bodyToMono(PaymentResult.class);

// exchangeToMono — full control
Mono<RateQuote> quote = client.get()
    .uri("/rates/{ccy}", "EUR")
    .exchangeToMono(resp -> {
      if (resp.statusCode().is2xxSuccessful()) {
        return resp.bodyToMono(RateQuote.class);
      }
      if (resp.statusCode().equals(HttpStatus.NOT_FOUND)) {
        return Mono.just(RateQuote.fallback("EUR"));
      }
      return resp.createException().flatMap(Mono::error);
    });

public Mono<InventorySnapshot> fetchInventory(String sku) {
  return client.get()
      .uri("/inventory/{sku}", sku)
      .retrieve()
      .bodyToMono(InventorySnapshot.class)
      .timeout(Duration.ofSeconds(3))
      .retryWhen(Retry.backoff(3, Duration.ofMillis(100))
          .maxBackoff(Duration.ofSeconds(2))
          .filter(ex -> ex instanceof WebClientRequestException))
      .onErrorResume(WebClientResponseException.NotFound.class,
          ex -> Mono.just(InventorySnapshot.empty(sku)));
}

Flux<LogLine> tailLogs = client.get()
    .uri("/logs/stream")
    .accept(MediaType.TEXT_EVENT_STREAM)
    .retrieve()
    .bodyToFlux(LogLine.class);

// Proxy stream through your controller
@GetMapping(value = "/logs", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
Flux<LogLine> proxyLogs() {
  return tailLogs;
}

implementation "org.springframework.boot:spring-boot-starter-data-r2dbc"
runtimeOnly "org.postgresql:r2dbc-postgresql"

@Table("orders")
record Order(@Id Long id, String status, Instant createdAt) {}

interface OrderRepository extends ReactiveCrudRepository<Order, Long> {
  Flux<Order> findByStatus(String status);
}

@Service
class OrderService {
  Mono<Order> ship(Long id) {
    return orders.findById(id)
        .switchIfEmpty(Mono.error(new OrderNotFoundException(id)))
        .flatMap(o -> orders.save(o.withStatus("SHIPPED")));
  }
}

spring:
  threads:
    virtual:
      enabled: true