A Multi represents a stream of data.
A stream can emit 0, 1, n, or an infinite number of items.
You will rarely create instances of Multi yourself but instead use a reactive client that exposes a Mutiny API.
Still, just like Uni there exists a rich API for creating Multi objects.
The Multi type
A Multi<T> is a data stream that:
-
emits
0..nitem events -
emits a failure event
-
emits a completion event for bounded streams
|
Failures are terminal events: after having received a failure no further item will be emitted. |
Multi<T> provides many operators that create, transform, and orchestrate Multi sequences.
The operators can be used to define a processing pipeline.
The events flow in this pipeline, and each operator can process or transform the events.
Multis are lazy by nature.
To trigger the computation, you must subscribe.
The following snippet provides a simple example of pipeline using Multi:
Multi.createFrom().items(1, 2, 3, 4, 5)
.onItem().transform(i -> i * 2)
.transform().byTakingFirstItems(3)
.onFailure().recoverWithItem(0)
.subscribe().with(System.out::println);You can see the Multi javadoc here.
Subscribing to a Multi
Remember, if you don’t subscribe, nothing is going to happen. Also, the pipeline is materialized for each subscription.
When subscribing to a Multi, you can pass an item callback (invoked when the item is emitted), or pass two callbacks, one receiving the item and one receiving the failure, or three callbacks to handle respectively the item, failure and completion events.
Cancellable cancellable = multi
.subscribe().with(
item -> System.out.println(item),
failure -> System.out.println("Failed with " + failure),
() -> System.out.println("Completed"));Note the returned Cancellable: this object allows canceling the stream if need be.
Creating Multi from items
There are many ways to create Multi instances.
Use Multi.createFrom() to see all the possibilities.
For instance you can create a Multi from known items or from an Iterable:
Multi<Integer> multiFromItems = Multi.createFrom().items(1, 2, 3, 4);
Multi<Integer> multiFromIterable = Multi.createFrom().iterable(Arrays.asList(1, 2, 3, 4, 5));Every subscriber receives the same set of items (1, 2… 5) just after the subscription.
You can also use Suppliers:
AtomicInteger counter = new AtomicInteger();
Multi<Integer> multi = Multi.createFrom().items(() ->
IntStream.range(counter.getAndIncrement(), counter.get() * 2).boxed());The Supplier is called for every subscriber, so each of them will get different values.
|
You can create ranges using |
Creating failing Multis
Streams can also fail.
Failures are used to indicate to the downstream subscribers that the source encountered a terrible error and cannot continue emitting items.
Create failed Multi instances with:
// Pass an exception directly:
Multi<Integer> failed1 = Multi.createFrom().failure(new Exception("boom"));
// Pass a supplier called for every subscriber:
Multi<Integer> failed2 = Multi.createFrom().failure(() -> new Exception("boom"));Creating empty Multis
Unlike Uni, Multi streams don’t send null items (this is forbidden in reactive streams).
Instead Multi streams send completion events indicating that there are no more items to consume.
Of course, the completion event can happen even if there are no items, creating an empty stream.
You can create such a stream using:
Multi<String> multi = Multi.createFrom().empty();Creating Multis using an emitter (advanced)
You can create a Multi using an emitter.
This approach is useful when integrating callback-based APIs:
Multi<Integer> multi = Multi.createFrom().emitter(em -> {
em.emit(1);
em.emit(2);
em.emit(3);
em.complete();
});The emitter can also send a failure. It can also get notified of cancellation to, for example, stop the work in progress.