Dispatching

The "Reactive Gateways" offer a reactive API wrapper around the command, query and event gateways. Most of the operations are similar to those from the non-reactive gateways, simply replacing the CompletableFuture with either a Mono or Flux.

Reactor doesn’t allow null values in streams. Any null value returned from the handler will be mapped to Mono#empty().
Retrying operations

All operations support Reactor’s retry mechanism:

public Mono<OrderSummary> findOrderWithRetries(FindOrderQuery query) {
    return reactiveQueryGateway.query(query, OrderSummary.class)
                               .retry(5);
}

This call will retry sending the query a maximum of five times when it fails.

Configuration

The ReactorCommandGateway, ReactorQueryGateway, and ReactorEventGateway are registered automatically when the axon-reactor artifact is on the classpath. They delegate to the regular CommandGateway, QueryGateway, and EventGateway of your application.

In a Spring Boot application, the gateways are available as beans, as the examples on this page show. In a plain Java application, retrieve a gateway from the AxonConfiguration:

import org.axonframework.common.configuration.AxonConfiguration;
import org.axonframework.extension.reactor.messaging.commandhandling.gateway.ReactorCommandGateway;

class PlainJavaGatewayLookup {

    ReactorCommandGateway reactorCommandGateway(AxonConfiguration configuration) {
        return configuration.getComponent(ReactorCommandGateway.class);
    }
}

Every operation of the gateways is also available with a ProcessingContext parameter. Provide it when you dispatch from within a message handler, to propagate correlation data from one message to another. When dispatching from outside a handler, use the variants without the ProcessingContext.

Reactor command gateway

This section describes the methods on the ReactorCommandGateway.

send

Sends the given command once the caller subscribes to the command result. Returns immediately.

There are two variants:

  • send(command, resultType) returns a Mono<R> with the command’s result, converted to the given resultType.

  • send(command) returns a Mono<Void> that completes once the command has been handled, regardless of its result.

A common pattern is using the REST API to send a command. In this case it is recommended to, for example, use WebFlux, and return the command result Mono directly to the controller:

Sending a command from a Spring WebFlux Controller.
import org.axonframework.extension.reactor.messaging.commandhandling.gateway.ReactorCommandGateway;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;

@RestController
class OrderCommandController {

    private final ReactorCommandGateway reactiveCommandGateway;

    OrderCommandController(ReactorCommandGateway reactiveCommandGateway) {
        this.reactiveCommandGateway = reactiveCommandGateway;
    }

    @PostMapping
    public Mono<OrderConfirmation> placeOrder(@RequestBody PlaceOrderCommand command) {
        return reactiveCommandGateway.send(command, OrderConfirmation.class);
    }
}

If the command’s result is of no interest, use the Mono<Void> variant:

public Mono<Void> ship(ShipOrderCommand command) {
    return reactiveCommandGateway.send(command);
}

Another common pattern is "send and forget":

Function that sends a command and returns immediately without waiting for the result.
class CommandDispatcher {

    private final ReactorCommandGateway reactiveCommandGateway;

    CommandDispatcher(ReactorCommandGateway reactiveCommandGateway) {
        this.reactiveCommandGateway = reactiveCommandGateway;
    }

    public void sendAndForget(ShipOrderCommand command) {
        reactiveCommandGateway.send(command)
                              .subscribe();
    }
}

Reactor query gateway

This section describes the methods on the ReactorQueryGateway.

query

Sends the given query, expecting a response in the form of responseType from a single source.

Recommended way of using the Reactor query gateway within a Spring REST controller.
import org.axonframework.extension.reactor.messaging.queryhandling.gateway.ReactorQueryGateway;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.List;

@RestController
class OrderQueryController {

    private final ReactorQueryGateway reactiveQueryGateway;

    OrderQueryController(ReactorQueryGateway reactiveQueryGateway) {
        this.reactiveQueryGateway = reactiveQueryGateway;
    }

    // The query's Mono is returned to the Spring controller. Subscribe control is given to Spring Framework.
    @GetMapping("/order")
    public Mono<OrderSummary> findOrder(FindOrderQuery query) {
        return reactiveQueryGateway.query(query, OrderSummary.class);
    }

    @GetMapping("/orders")
    public Mono<List<OrderSummary>> findOrders(FindOrdersQuery query) {
        return reactiveQueryGateway.queryMany(query, OrderSummary.class);
    }

    @GetMapping("/orders/stream")
    public Flux<OrderSummary> streamOrders(FindOrdersQuery query) {
        return reactiveQueryGateway.streamingQuery(query, OrderSummary.class);
    }
}

queryMany

Sends the given query, expecting a list of responseType instances as the response. The result is a Mono of that list, as shown in the controller above.

streamingQuery

Sends the given query as a streaming query, expecting the response to be streamed element by element. The result is a Flux of responseType instances, as shown in the controller above.

subscriptionQuery

Sends the given query as a subscription query. The resulting Flux first emits the initial result, and then keeps streaming incremental updates until the subscriber cancels its subscription.

Note that the initial result and the updates are both converted to the same responseType.

class OrderUpdateSubscriber {

    private final ReactorQueryGateway reactiveQueryGateway;

    OrderUpdateSubscriber(ReactorQueryGateway reactiveQueryGateway) {
        this.reactiveQueryGateway = reactiveQueryGateway;
    }

    // Emits the initial result first, followed by every update until the subscriber cancels
    public Flux<OrderSummary> subscribeToOrder(FindOrderQuery query) {
        return reactiveQueryGateway.subscriptionQuery(query, OrderSummary.class);
    }
}

Reactor event gateway

Reactive variation of the EventGateway. Provides support for reactive return types.

publish

Publishes the given events once the caller subscribes to the resulting Mono. The Mono completes when the events have been published.

import org.axonframework.extension.reactor.messaging.eventhandling.gateway.ReactorEventGateway;
import reactor.core.publisher.Mono;

class OrderEventPublisher {

    private final ReactorEventGateway reactiveEventGateway;

    OrderEventPublisher(ReactorEventGateway reactiveEventGateway) {
        this.reactiveEventGateway = reactiveEventGateway;
    }

    public Mono<Void> publishOrderPlaced(String orderId) {
        return reactiveEventGateway.publish(new OrderPlacedEvent(orderId));
    }
}

The messages sent through the Reactor gateways can be intercepted before they are dispatched. See Intercepting for the details.