Creating a Flux or Mono sets up a source of data, but does not make the data flow. Calling subscribe() attaches a subscriber and starts consumption—like turning on the tap after connecting a garden hose. section 11.3.1 section 11.2
Consume a Flux
Flux<String> fruitFlux = Flux.just("Apple", "Orange", "Grape"); fruitFlux.subscribe(fruit -> System.out.println("Here's some fruit: " + fruit)); The lambda consumes each item. Reactor uses it to create a Reactive Streams Subscriber. With no intermediate operations, the fruit goes directly from the Flux to that subscriber. section 11.3.1
Operations can sit between the source and subscribe(). For example, Mono.just("Craig").map(String::toUpperCase).subscribe(System.out::println) transforms the value before printing it. Do not confuse subscribe() with subscribeOn(): the latter specifies how a subscription is handled concurrently, rather than starting consumption itself. section 11.2 section 11.3.3
For a WebClient request, subscribe() can take two consumers: one for a returned value and another for an error. A missing ingredient may send a WebClientResponseException to the error consumer; onStatus() can instead map that response to a domain-specific exception. Applying timeout() before subscribe() likewise sends a timeout to the error consumer if the wait exceeds the chosen duration. section 12.4.1 section 12.4.4
Flux<Ingredient> ingredients = webclient
.get()
.uri("/ingredients")
.retrieve()
.bodyToFlux(Ingredient.class);
ingredients
.timeout(Duration.ofSeconds(1))
.subscribe(
i -> { ... },
e -> {
// handle timeout error
});
This WebClient example places a one-second limit before subscribe() and supplies separate handlers for ingredients and errors. section 12.4.1
In tests, StepVerifier is often more useful than printing values: it subscribes, checks the items as they arrive, and verifies completion. At the Reactive Streams interface level, a subscriber receives onSubscribe(), then can request items or cancel; items, errors, and completion arrive as distinct signals. section 11.3.1 section 11.1.1
