Support Flux<...> as QueryHandler return type

[raphael-jungers]

When using Spring Data with reactive repositories, the methods return types are Mono<...> or Flux<...>. The QueryGateway doesn't seem to support these types as QueryHandlers return types.

I'm mostly interested in using the subscriptionQuery method with a Flux as the initial response type.

[smcvb]

Thanks for your feature request @raphael-jungers.
I've marked this ticket as under discussion for now.
Any suggestions from the team's side will be shared in this thread, so stay tuned. :-)

[raphael-jungers]

I finally managed to use spring-data-cassandra-reactive with cassandra as the query database without any change to the framework.

Here's a code example for those interested :

The Controller :

    fun getTransactions(payload: Mono<String>): Flux<Payload> {
        return payload.map {
            ObjectMapper().readValue(it, SearchTransactionsQuery::class.java)
        }.map { query ->
            val result = queryGateway.subscriptionQuery(query, ResponseTypes.multipleInstancesOf(Transaction::class.java), ResponseTypes.instanceOf(Transaction::class.java))
            Flux.concat(result.initialResult(), result.updates())
                    .map {
                        DefaultPayload.create(ObjectMapper().writeValueAsString(it))
                    }
        }.flatMapMany { flux -> flux }
    }

The QueryHandler :

    @QueryHandler
    fun handle(query: SearchTransactionsQuery): Iterable<Transaction> {
        return transactionRepository.findAll().toIterable()
    }

The EventHandler (emitting updates)

@EventHandler
    fun on(event: TransactionExecutedEvent) {

        val transaction = Transaction(event.id, event.exampleData)

        transactionRepository.save(transaction).doOnNext { t ->
            queryUpdateEmitter.emit(SearchTransactionsQuery::class.java,
                    { query -> true },
                    t)
        }.subscribe()
    }

[smcvb]

Great you got that working! :)
Iterable is indeed one of the types the ResponseType class, specifically the MultipleInstancesResponseType implementation, can cope with.
FYI: the multiple instance of ResponseType should also be able to cope with Streams and arrays, either of those three with or without generics and wildcards.

[UkonnRa]

+1, and maybe kotlin coroutine should also be supported

[smcvb]

@UkonnRa please maintain Kotlin specific requests to the Kotlin Extension Repository. We know you've dropped an issue request there, but simply reiterating the fact here does not increase prioritization on the matter. We first need to figure out an implementation tactic for the matter (hence the "Under Discussion" tag) prior to initiation work on it.

Of course, implementation suggestion or a pull request are very much valued from any users of Axon. Thus if you are up for the task, I'd be looking forward to reviewing your PR!

[psoares]

Just stumbled upon this use case and ended up doing what looks like a dirty workaround:

Flux<Foo> orderedProducts = Mono.fromFuture(queryGateway.query(new FindFoos(), ResponseTypes.multipleInstancesOf(Foo.class)))
                                      .flatMapMany(Flux::fromIterable);

[smcvb]

I think we've missed adding this issue to milestone 4.6.0, when we started our efforts on the Streaming Query support we introduced in #2001.

With that in place you can use QueryGateway#streamingQuery(Q query, Class<R> responseType) to retrieve a Publisher as a response.
Furthermore, it allows your query handlers to be written in this format:

@QueryHandler
public Flux<Order> handle(FindAllOrdersQuery query) {
    return Flux.of(...
}

Given the nature of this issue, we feel the aforementioned support is thus present as of 4.6.0 of Axon Framework.
Please give it a try and let us know what you think!

[gklijs]

Please also note that using the reactor extension, we do offer a subscription query with a flux return type.