org.axonframework.queryhandling

Class SimpleQueryBus

java.lang.Object

org.axonframework.queryhandling.SimpleQueryBus

All Implemented Interfaces:

QueryBus, QueryUpdateEmitter

public class SimpleQueryBus
extends Object
implements QueryBus, QueryUpdateEmitter

Implementation of the QueryBus that dispatches queries to the handlers within the JVM. Any timeouts are ignored by this implementation, as handlers are considered to answer immediately.

In case multiple handlers are registered for the same query and response type, the query(QueryMessage) method will invoke one of these handlers. Which one is unspecified.

Since:

3.1

Author:

Marc Gathier, Allard Buijze, Steven van Beelen, Milan Savic

Constructor Summary

Constructors

Constructor and Description
SimpleQueryBus() Initialize the query bus without monitoring on messages and a LoggingQueryInvocationErrorHandler.
SimpleQueryBus(MessageMonitor<? super QueryMessage<?,?>> messageMonitor, MessageMonitor<? super SubscriptionQueryUpdateMessage<?>> updateMessageMonitor, TransactionManager transactionManager, QueryInvocationErrorHandler errorHandler) Initialize the query bus with the given messageMonitor, updateMessageMonitor, transactionManager and given errorHandler.
SimpleQueryBus(MessageMonitor<? super QueryMessage<?,?>> messageMonitor, TransactionManager transactionManager, QueryInvocationErrorHandler errorHandler) Initialize the query bus with the given messageMonitor and given errorHandler.
SimpleQueryBus(TransactionManager transactionManager) Initialize the query bus using given transactionManager to manage transactions around query execution with.

Method Summary

Modifier and Type	Method and Description
Set<SubscriptionQueryMessage<?,?,?>>	activeSubscriptions() Provides the set of running subscription queries.
void	complete(Predicate<SubscriptionQueryMessage<?,?,?>> filter) Completes subscription queries matching given filter.
void	completeExceptionally(Predicate<SubscriptionQueryMessage<?,?,?>> filter, Throwable cause) Completes with an error subscription queries matching given filter.
<U> void	emit(Predicate<SubscriptionQueryMessage<?,?,U>> filter, SubscriptionQueryUpdateMessage<U> update) Emits incremental update (as return value of provided update function) to subscription queries matching given filter.
protected Map<String,Collection<org.axonframework.queryhandling.QuerySubscription>>	getSubscriptions() Returns the subscriptions for this query bus.
<Q,R> CompletableFuture<QueryResponseMessage<R>>	query(QueryMessage<Q,R> query) Dispatch the given query to a single QueryHandler subscribed to the given query's queryName and responseType.
Registration	registerDispatchInterceptor(MessageDispatchInterceptor<? super QueryMessage<?,?>> interceptor) Registers an interceptor that intercepts Queries as they are sent.
Registration	registerHandlerInterceptor(MessageHandlerInterceptor<? super QueryMessage<?,?>> interceptor) Registers an interceptor that is used to intercept Queries before they are passed to their respective handlers.
<Q,R> Stream<QueryResponseMessage<R>>	scatterGather(QueryMessage<Q,R> query, long timeout, TimeUnit unit) Dispatch the given query to all QueryHandlers subscribed to the given query's queryName/responseType.
<R> Registration	subscribe(String queryName, Type responseType, MessageHandler<? super QueryMessage<?,R>> handler) Subscribe the given handler to queries with the given queryName and responseType.
<Q,I,U> SubscriptionQueryResult<QueryResponseMessage<I>,SubscriptionQueryUpdateMessage<U>>	subscriptionQuery(SubscriptionQueryMessage<Q,I,U> query, SubscriptionQueryBackpressure backpressure, int updateBufferSize) Dispatch the given query to a single QueryHandler subscribed to the given query's queryName/initialResponseType/updateResponseType.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.axonframework.queryhandling.QueryBus

subscriptionQuery

Methods inherited from interface org.axonframework.queryhandling.QueryUpdateEmitter

complete, completeExceptionally, emit, emit, emit

Constructor Detail

SimpleQueryBus

public SimpleQueryBus()

Initialize the query bus without monitoring on messages and a LoggingQueryInvocationErrorHandler.

SimpleQueryBus

public SimpleQueryBus(TransactionManager transactionManager)

Initialize the query bus using given transactionManager to manage transactions around query execution with. No monitoring is applied to messages and a LoggingQueryInvocationErrorHandler is used to log errors on handlers during a scatter-gather query.

Parameters:

transactionManager - The transaction manager to manage transactions around query execution with

SimpleQueryBus

public SimpleQueryBus(MessageMonitor<? super QueryMessage<?,?>> messageMonitor,
                      TransactionManager transactionManager,
                      QueryInvocationErrorHandler errorHandler)

Initialize the query bus with the given messageMonitor and given errorHandler.

Parameters:

messageMonitor - The message monitor notified for incoming messages and their result

transactionManager - The transaction manager to manage transactions around query execution with

errorHandler - The error handler to invoke when query handler report an error

SimpleQueryBus

public SimpleQueryBus(MessageMonitor<? super QueryMessage<?,?>> messageMonitor,
                      MessageMonitor<? super SubscriptionQueryUpdateMessage<?>> updateMessageMonitor,
                      TransactionManager transactionManager,
                      QueryInvocationErrorHandler errorHandler)

Initialize the query bus with the given messageMonitor, updateMessageMonitor, transactionManager and given errorHandler.

Parameters:

messageMonitor - The message monitor notified for incoming messages and their result

updateMessageMonitor - The message monitor notified for incoming update message in regard to subscription queries

transactionManager - The transaction manager to manage transactions around query execution with

errorHandler - The error handler to invoke when query handler report an error

Method Detail

subscribe

public <R> Registration subscribe(String queryName,
                                  Type responseType,
                                  MessageHandler<? super QueryMessage<?,R>> handler)

Description copied from interface: QueryBus

Subscribe the given handler to queries with the given queryName and responseType. Multiple handlers may subscribe to the same combination of queryName/responseType.

Specified by:

subscribe in interface QueryBus

Parameters:

queryName - the name of the query request to subscribe

responseType - the type of response the subscribed component answers with

handler - a handler that implements the query

Returns:

a handle to un-subscribe the query handler

query

public <Q,R> CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q,R> query)

Description copied from interface: QueryBus

Dispatch the given query to a single QueryHandler subscribed to the given query's queryName and responseType. This method returns all values returned by the Query Handler as a Collection. This may or may not be the exact collection as defined in the Query Handler.

If the Query Handler defines a single return object (i.e. not a collection or array), that object is returned as the sole entry in a singleton collection.

When no handlers are available that can answer the given query, the returned CompletableFuture will be completed with a NoHandlerForQueryException.

Specified by:

query in interface QueryBus

Type Parameters:

Q - the payload type of the query

R - the response type of the query

Parameters:

query - the query

Returns:

a CompletableFuture that resolves when the response is available

scatterGather

public <Q,R> Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q,R> query,
                                                           long timeout,
                                                           TimeUnit unit)

Description copied from interface: QueryBus

Dispatch the given query to all QueryHandlers subscribed to the given query's queryName/responseType. Returns a stream of results which blocks until all handlers have processed the request or when the timeout occurs.

If no handlers are available to provide a result, or when all available handlers throw an exception while attempting to do so, the returned Stream is empty.

Note that any terminal operation (such as Stream.forEach(Consumer)) on the Stream may cause it to block until the timeout has expired, awaiting additional data to include in the stream.

Specified by:

scatterGather in interface QueryBus

Type Parameters:

Q - the payload type of the query

R - the response type of the query

Parameters:

query - the query

timeout - time to wait for results

unit - unit for the timeout

Returns:

stream of query results

subscriptionQuery

public <Q,I,U> SubscriptionQueryResult<QueryResponseMessage<I>,SubscriptionQueryUpdateMessage<U>> subscriptionQuery(SubscriptionQueryMessage<Q,I,U> query,
                                                                                                                    SubscriptionQueryBackpressure backpressure,
                                                                                                                    int updateBufferSize)

Description copied from interface: QueryBus

Dispatch the given query to a single QueryHandler subscribed to the given query's queryName/initialResponseType/updateResponseType. The result is lazily created and there will be no execution of the query handler before there is a subscription to the initial result. In order not to miss updates, the query bus will queue all updates which happen after the subscription query is done and once the subscription to the flux is made, these updates will be emitted.

If there is an error during retrieving or consuming initial result, stream for incremental updates is NOT interrupted.

If there is an error during emitting an update, subscription is cancelled causing further emits not reaching the destination.

Provided backpressure mechanism will be used to deal with fast emitters.

Specified by:

subscriptionQuery in interface QueryBus

Type Parameters:

Q - the payload type of the query

I - the response type of the query

U - the incremental response types of the query

Parameters:

query - the query

backpressure - the backpressure mechanism to be used for emitting updates

updateBufferSize - the size of buffer which accumulates updates before subscription to the flux is made

Returns:

query result containing initial result and incremental updates

emit

public <U> void emit(Predicate<SubscriptionQueryMessage<?,?,U>> filter,
                     SubscriptionQueryUpdateMessage<U> update)

Description copied from interface: QueryUpdateEmitter

Emits incremental update (as return value of provided update function) to subscription queries matching given filter.

Specified by:

emit in interface QueryUpdateEmitter

Type Parameters:

U - the type of the update

Parameters:

filter - predicate on subscription query message used to filter subscription queries

update - incremental update message

complete

public void complete(Predicate<SubscriptionQueryMessage<?,?,?>> filter)

Description copied from interface: QueryUpdateEmitter

Completes subscription queries matching given filter.

Specified by:

complete in interface QueryUpdateEmitter

Parameters:

filter - predicate on subscription query message used to filter subscription queries

completeExceptionally

public void completeExceptionally(Predicate<SubscriptionQueryMessage<?,?,?>> filter,
                                  Throwable cause)

Description copied from interface: QueryUpdateEmitter

Completes with an error subscription queries matching given filter.

Specified by:

completeExceptionally in interface QueryUpdateEmitter

Parameters:

filter - predicate on subscription query message used to filter subscription queries

cause - the cause of an error

activeSubscriptions

public Set<SubscriptionQueryMessage<?,?,?>> activeSubscriptions()

Provides the set of running subscription queries. If there are changes to subscriptions they will be reflected in the returned set of this method.

Returns:

the set of running subscription queries

getSubscriptions

protected Map<String,Collection<org.axonframework.queryhandling.QuerySubscription>> getSubscriptions()

Returns the subscriptions for this query bus. While the returned map is unmodifiable, it may or may not reflect changes made to the subscriptions after the call was made.

Returns:

the subscriptions for this query bus

registerHandlerInterceptor

public Registration registerHandlerInterceptor(MessageHandlerInterceptor<? super QueryMessage<?,?>> interceptor)

Registers an interceptor that is used to intercept Queries before they are passed to their respective handlers. The interceptor is invoked separately for each handler instance (in a separate unit of work).

Parameters:

interceptor - the interceptor to invoke before passing a Query to the handler

Returns:

handle to unregister the interceptor

registerDispatchInterceptor

public Registration registerDispatchInterceptor(MessageDispatchInterceptor<? super QueryMessage<?,?>> interceptor)

Registers an interceptor that intercepts Queries as they are sent. Each interceptor is called once, regardless of the type of query (point-to-point or scatter-gather) executed.

Parameters:

interceptor - the interceptor to invoke when sending a Query

Returns:

handle to unregister the interceptor