SmallRye Mutiny is the reactive API exposed by Quarkus for asynchronous, event-driven applications. Code reacts when a result, failure, or stream item arrives rather than waiting for it. Mutiny implements Reactive Streams and includes backpressure, which lets consumers signal producers to slow down. section 8.2
Choose a type by the shape of the result:
- Uni represents one asynchronously computed item or a failure. Subscribing triggers its computation; unlike a full Reactive Streams publisher, it can also handle null. section 8.2.2
- Multi represents a stream of items, potentially without an end. Its fluent methods build a pipeline whose stages can transform, filter, or react to events. section 8.2.2
A pipeline starts when something subscribes. Each stage then runs as the preceding stage emits an event. Mutiny’s fluent API makes it possible to assemble those stages incrementally, using IDE autocompletion and JavaDoc to find operations. section 8.2.2
Listing 8.1 An example use of the Mutiny API
Multi.createFrom().items("a", "b", "c")
.onItem()
.transform(String::toUpperCase)
.onItem()
.invoke(s -> System.out.println("Intermediate stage " + s))
.onItem()
.transform(s -> s + " item")
.filter(s -> !s.startsWith("B"))
.onCompletion()
.invoke(() -> System.out.println("Stream completed"))
.subscribe()
.with(s -> System.out.println("Subscriber received " + s));
The pipeline creates three items, transforms them to uppercase, logs each intermediate item, appends text, and filters out the item beginning with B. Subscription delivers the remaining items and triggers completion handling. section 8.2.2
Quarkus’s gRPC extension generates service APIs with Mutiny types alongside a blocking API. Its Mutiny-specific interface replaces the default StreamObserver approach with types consistent with other Quarkus extensions. A gRPC request stream can map naturally to Multi: an inventory service maps incoming requests to model objects, persists them, and maps them to response items. section 4.7 section 8.4.2
Listing 8.5 The GrpcInventoryService#add method
@Override
@Blocking
public Multi<CarResponse> add(Multi<InsertCarRequest> requests) {
return requests
.map(request -> {
Car car = new Car();
car.setLicensePlateNumber(request.getLicensePlateNumber());
car.setManufacturer(request.getManufacturer());
car.setModel(request.getModel());
return car;
}).onItem().invoke(car -> {
QuarkusTransaction.requiringNew().run( () -> {
carRepository.persist(car);
Log.info("Persisting " + car);
});
}).map(car -> CarResponse.newBuilder()
.setLicensePlateNumber(car.getLicensePlateNumber())
.setManufacturer(car.getManufacturer())
.setModel(car.getModel())
.setId(car.getId())
.build());
}
The inventory method shows the important constraint: returning Multi does not make blocking database I/O nonblocking. Its @Blocking annotation keeps that work off the event-loop thread. section 8.4.2
