An Emitter lets imperative code send a message into a MicroProfile Reactive Messaging channel. Inject an Emitter with @Channel, then call send with the message payload. This is useful when a JAX-RS endpoint needs to start asynchronous work without becoming a message producer itself. section 9.3.4 section 9.4
Listing 9.3 An Emitter example
@Inject
@Channel("requests")
Emitter<String> requestsEmitter;
@POST
@Path("/request")
public String request(String body) {
requestsEmitter.send(body);
return "Processing " + body;
}
A POST endpoint injects Emitter<String> for the requests channel and sends the request body before returning a response. section 9.3.4
send returns a CompletionStage that completes when the message is acknowledged. A negative acknowledgment completes it exceptionally; a message that is never acknowledged leaves it incomplete. Simply calling send does not wait for either result. To wait synchronously, join the stage, but that blocks the thread. A reactive pipeline can instead incorporate the acknowledgment using MutinyEmitter and Uni. section 9.3.4 section 9.4
Requesting billing for a reservation
Add quarkus-messaging, inject Emitter<Invoice> with @Channel("invoices"), and send an invoice after persisting a reservation. An @Incoming("invoices") method can initially act as a billing stub and print the received invoice. When receipt matters, handle the CompletionStage failure rather than assuming that a successful send call means billing received the request. section 9.4
Listing 9.6 ReservationResource producing messages to channel
public static final double STANDARD_RATE_PER_DAY = 19.99;
@Inject
@Channel("invoices")
Emitter<Invoice> invoiceEmitter; #1
@Consumes(MediaType.APPLICATION_JSON)
@POST
@WithTransaction
public Uni<Reservation> make(Reservation reservation) {
...
return reservation.<Reservation>persist().onItem()
.call(persistedReservation -> {
Log.info("Successfully reserved reservation " +
persistedReservation);
invoiceEmitter.send(new Invoice(reservation,
computePrice(reservation))); #2
...
}
private double computePrice(Reservation reservation) {
return (ChronoUnit.DAYS.between(reservation.startDay,
reservation.endDay) + 1) * STANDARD_RATE_PER_DAY;
}
The reservation endpoint constructs an invoice from the reservation and its calculated price, then sends it to the invoices channel. section 9.4
The same pattern can send invoice adjustments when a rental ends on a different date than reserved: inject Emitter<InvoiceAdjust> for the invoices-adjust channel. With the Kafka messaging extension, that channel is managed by the Kafka connector; a console consumer can inspect the resulting topic messages. section 9.6.1
