Package org.axonframework.queryhandling

Class SimpleQueryBus

java.lang.Object
org.axonframework.queryhandling.SimpleQueryBus
All Implemented Interfaces:
MessageDispatchInterceptorSupport<QueryMessage<?,?>>, MessageHandlerInterceptorSupport<QueryMessage<?,?>>, QueryBus
public class SimpleQueryBus extends Object implements QueryBus
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 Details

  • Method Details

    • builder

      public static SimpleQueryBus.Builder builder()
      Instantiate a Builder to be able to create a SimpleQueryBus.

      The MessageMonitor is defaulted to NoOpMessageMonitor, TransactionManager to NoTransactionManager, QueryInvocationErrorHandler to LoggingQueryInvocationErrorHandler, the QueryBusSpanFactory defaults to a DefaultQueryBusSpanFactory backed by a NoOpSpanFactory and QueryUpdateEmitter to SimpleQueryUpdateEmitter.

      Returns:
      a Builder to be able to create a SimpleQueryBus

    • subscribe

      public <R> Registration subscribe(@Nonnull String queryName, @Nonnull Type responseType, @Nonnull 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(@Nonnull 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

    • streamingQuery

      public <Q, R> org.reactivestreams.Publisher<QueryResponseMessage<R>> streamingQuery(StreamingQueryMessage<Q,R> query)
      Description copied from interface: QueryBus
      Builds a Publisher of responses to the given query. The actual query is not dispatched until there is a subscription to the result. The query is dispatched to a single query handler. Implementations may opt for invoking several query handlers and then choosing a response from single one for performance or resilience reasons.

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

      Specified by:
      streamingQuery in interface QueryBus
      Type Parameters:
      Q - the payload type of the streaming query
      R - the response type of the streaming query
      Parameters:
      query - the streaming query message
      Returns:
      a Publisher of responses

    • scatterGather

      public <Q, R> Stream<QueryResponseMessage<R>> scatterGather(@Nonnull QueryMessage<Q,R> query, long timeout, @Nonnull 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

      @Deprecated public <Q, I, U> SubscriptionQueryResult<QueryResponseMessage<I>,SubscriptionQueryUpdateMessage<U>> subscriptionQuery(@Nonnull SubscriptionQueryMessage<Q,I,U> query, SubscriptionQueryBackpressure backpressure, int updateBufferSize)
      Deprecated.
      in favor use of {subscriptionQuery(SubscriptionQueryMessage, int)
      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

    • subscriptionQuery

      public <Q, I, U> SubscriptionQueryResult<QueryResponseMessage<I>,SubscriptionQueryUpdateMessage<U>> subscriptionQuery(@Nonnull SubscriptionQueryMessage<Q,I,U> query, 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.

      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
      updateBufferSize - the size of buffer which accumulates updates before subscription to the flux is made
      Returns:
      query result containing initial result and incremental updates

    • queryUpdateEmitter

      public QueryUpdateEmitter queryUpdateEmitter()
      Description copied from interface: QueryBus
      Gets the QueryUpdateEmitter associated with this QueryBus.
      Specified by:
      queryUpdateEmitter in interface QueryBus
      Returns:
      the associated QueryUpdateEmitter

    • getSubscriptions

      protected Map<String,Collection<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(@Nonnull 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).
      Specified by:
      registerHandlerInterceptor in interface MessageHandlerInterceptorSupport<QueryMessage<?,?>>
      Parameters:
      interceptor - the interceptor to invoke before passing a Query to the handler
      Returns:
      handle to deregister the interceptor

    • registerDispatchInterceptor

      @Nonnull public Registration registerDispatchInterceptor(@Nonnull 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.
      Specified by:
      registerDispatchInterceptor in interface MessageDispatchInterceptorSupport<QueryMessage<?,?>>
      Parameters:
      interceptor - the interceptor to invoke when sending a Query
      Returns:
      handle to deregister the interceptor