Flux is Project Reactor’s publisher type for asynchronously emitting zero or more ordered elements, including potentially infinite streams. It implements Reactive Streams’ Publisher and forms a pipeline through which data flows without assuming that every operation runs on the same thread. section 11.2
Create Flux instances, transform their values, combine streams, and subscribe to start receiving data. Mono handles datasets containing no more than one element; Flux handles broader streams and shares many of Mono’s operations. section 11.2 section 11.3
Common Flux composition patterns include:
- Merge streams with mergeWith() so values arrive according to source timing rather than a guaranteed alternating order. section 11.3.2
- Zip streams so each output combines corresponding elements from both sources, either as tuples or transformed messages. section 11.3.2
- Select the first source to emit with firstWithSignal(), then continue receiving values from that source. section 11.3.2
Merge character and food streams
A character Flux containing Garfield, Kojak, and Barbossa can be merged with a food Flux containing Lasagna, Lollipops, and Apples. Delays make the emissions interleave, and StepVerifier checks the resulting sequence. section 11.3.2
@Test
public void mergeFluxes() {
Flux<String> characterFlux = Flux
.just("Garfield", "Kojak", "Barbossa")
.delayElements(Duration.ofMillis(500));
Flux<String> foodFlux = Flux
.just("Lasagna", "Lollipops", "Apples")
.delaySubscription(Duration.ofMillis(250))
.delayElements(Duration.ofMillis(500));
Flux<String> mergedFlux = characterFlux.mergeWith(foodFlux);
StepVerifier.create(mergedFlux)
.expectNext("Garfield")
.expectNext("Lasagna")
.expectNext("Kojak")
.expectNext("Lollipops")
.expectNext("Barbossa")
.expectNext("Apples")
.verifyComplete();
}
Shows a complete mergeWith() test, including delayed emissions and StepVerifier assertions. section 11.3.2
