Package org.axonframework.messaging.queryhandling.gateway

Class ShutdownTrackingQueryGateway

java.lang.Object
org.axonframework.messaging.queryhandling.gateway.ShutdownTrackingQueryGateway
All Implemented Interfaces:
DescribableComponent, QueryGateway
public class ShutdownTrackingQueryGateway extends Object implements QueryGateway
A QueryGateway decorator that automatically registers subscription queries and streaming queries with a QueryShutdownManager, ensuring they are canceled during application shutdown.

This decorator wraps any QueryGateway implementation and transparently intercepts subscriptionQuery(java.lang.Object, java.lang.Class<R>, org.axonframework.messaging.core.unitofwork.ProcessingContext, int) and streamingQuery(java.lang.Object, java.lang.Class<R>, org.axonframework.messaging.core.unitofwork.ProcessingContext) calls, passing their results through the respective shutdown managers. All other gateway operations are delegated unchanged.

It is the gateway layer of a layered API: configure it once on the gateway and every dispatched subscription or streaming query is tracked automatically, with no changes needed at call sites.

The recommended way to register this decorator is through MessagingConfigurer.registerQueryGateway(String, java.util.function.Consumer), which creates a named QueryGateway with shutdown tracking applied: 


 MessagingConfigurer.create()
     .registerQueryGateway("reporting", g -> g
         .cancellingSubscriptionQueryOnShutdown()
     );
When using a dependency injection framework, a ShutdownTrackingQueryGateway can also be registered as a named component by constructing it directly with a delegate gateway and the desired QueryShutdownManager instances. Use build(QueryGateway, QueryShutdownManager, QueryShutdownManager) to create new ShutdownTrackingQueryGateway instances at all times.

When call-site tracking is preferred over gateway-level tracking, wrap individual results with QueryShutdownManager.track(org.reactivestreams.Publisher) instead.

Since:
5.4.0
Author:
Allard Buijze

  • Method Details

    • build

      public static ShutdownTrackingQueryGateway build(QueryGateway delegate, @Nullable QueryShutdownManager subscriptionQueryShutdownManager, @Nullable QueryShutdownManager streamingQueryShutdownManager)
      Builds a ShutdownTrackingQueryGateway, delegating to the given delegate while using the given subscriptionQueryShutdownManager and streamingQueryShutdownManager to automatically shutdown subscription- and straeming queries respectively.

      If both the subscriptionQueryShutdownManager and streamingQueryShutdownManager are null, an AxonConfigurationException is thrown.

      Parameters:
      delegate - the QueryGateway to delegate all operations to
      subscriptionQueryShutdownManager - the manager to track subscription queries with, or null to skip subscription query tracking
      streamingQueryShutdownManager - the manager to track streaming queries with, or null to skip streaming query tracking
      Returns:
      a tracking wrapper around delegate
      Throws:
      AxonConfigurationException - when both the subscriptionQueryShutdownManager and streamingQueryShutdownManager are null

    • query

      public <R> CompletableFuture<R> query(Object query, Class<R> responseType, @Nullable ProcessingContext context)
      Description copied from interface: QueryGateway
      Sends given query over the QueryBus, expecting a single response with the given responseType from a single source.

      Execution may be asynchronous, depending on the QueryBus implementation.

      The given query is wrapped as the payload of the QueryMessage that is eventually posted on the QueryBus, unless the query already implements Message. In that case, a QueryMessage is constructed from that message's payload and Metadata.

      Specified by:
      query in interface QueryGateway
      Type Parameters:
      R - the generic type of the expected response
      Parameters:
      query - the query to be sent
      responseType - a Class describing the desired response type
      context - the processing context, if any, to dispatch the given query in
      Returns:
      a CompletableFuture containing a single query response of type responseType

    • queryMany

      public <R> CompletableFuture<List<R>> queryMany(Object query, Class<R> responseType, @Nullable ProcessingContext context)
      Description copied from interface: QueryGateway
      Sends given query over the QueryBus, expecting multiple responses in the form of responseType from a single source.

      Execution may be asynchronous, depending on the QueryBus implementation.

      The given query is wrapped as the payload of the QueryMessage that is eventually posted on the QueryBus, unless the query already implements Message. In that case, a QueryMessage is constructed from that message's payload and Metadata.

      Specified by:
      queryMany in interface QueryGateway
      Type Parameters:
      R - the generic type of the expected response(s)
      Parameters:
      query - the query to be sent
      responseType - a Class describing the desired response type
      context - the processing context, if any, to dispatch the given query in
      Returns:
      a CompletableFuture containing a list of query responses of type responseType

    • streamingQuery

      public <R> org.reactivestreams.Publisher<R> streamingQuery(Object query, Class<R> responseType, @Nullable ProcessingContext context)
      Description copied from interface: QueryGateway
      Sends given query over the QueryBus, expecting a response as a Publisher of responseType.

      The query is sent once the Publisher is subscribed to. The streaming query allows a client to stream large result sets. Usage of this method requires Project Reactor on the class path.

      The given query is wrapped as the payload of the QueryMessage that is eventually posted on the QueryBus, unless the query already implements Message. In that case, a QueryMessage is constructed from that message's payload and Metadata.

      Publisher is used for backwards compatibility reason, for clients that don't have Project Reactor on class path. Check Reactor Extension for native Flux type and more. Use Flux.from(publisher) to convert to Flux stream.

      Specified by:
      streamingQuery in interface QueryGateway
      Type Parameters:
      R - the generic type of the expected response(s)
      Parameters:
      query - the query to be sent
      responseType - a Class describing the desired response type
      context - the processing context, if any, to dispatch the given query in
      Returns:
      a Publisher streaming the results as dictated by the given responseType

    • subscriptionQuery

      public <R> org.reactivestreams.Publisher<R> subscriptionQuery(Object query, Class<R> responseType, @Nullable ProcessingContext context, int updateBufferSize)
      Description copied from interface: QueryGateway
      Sends given query over the QueryBus as a subscription query, combining the initial result and emitted update as a Publisher of the given responseType.

      The query is sent once the Publisher is subscribed to. Furthermore, updates are received at the moment the query is sent, and until it is cancelled by the caller or closed by the emitting side.

      The given query is wrapped as the payload of the QueryMessage that is eventually posted on the QueryBus, unless the query already implements Message. In that case, a QueryMessage is constructed from that message's payload and Metadata.

      Note that any null results, on the initial result or the update, will be filtered out by the gateway. If you require the null to be returned for the initial and update results, we suggest using the QueryBus instead.

      The returned Publisher must be subscribed to and consumed from before the buffer fills up. Once the buffer is full, any attempt to add an update will complete the stream with an exception.

      Specified by:
      subscriptionQuery in interface QueryGateway
      Type Parameters:
      R - the type of all the responses
      Parameters:
      query - the query to be sent
      responseType - the response type returned by this query as the initial result and the update
      context - the processing context, if any, to dispatch the given query in
      updateBufferSize - the size of buffer which accumulates updates before a subscription to the Flux is made
      Returns:
      a Publisher streaming the results as dictated by the given responseType

    • subscriptionQuery

      public <R> org.reactivestreams.Publisher<R> subscriptionQuery(Object query, Class<R> responseType, Function<QueryResponseMessage,R> mapper, @Nullable ProcessingContext context, int updateBufferSize)
      Description copied from interface: QueryGateway
      Sends given query over the QueryBus and returns a Publisher supplying the initial update followed by the update. The given mapper is used to map the QueryResponseMessage to the desired object type. To distinguish between the initial result and the update, the given mapper can check whether the given responseMessage is an instance of SubscriptionQueryUpdateMessage. In that case the message is considered an update, otherwise it is considered the initial result.

      The query is sent upon invocation of this method. Furthermore, updates are received at the moment the query is sent, and until the subscription to the Publisher is canceled by the caller or closed by the emitting side.

      The given query is wrapped as the payload of the QueryMessage that is eventually posted on the QueryBus, unless the query already implements Message. In that case, a QueryMessage is constructed from that message's payload and Metadata.

      Note that any null results, on the initial result or the update, will be filtered out by the gateway. If you require the null to be returned for the initial and update results, we suggest using the QueryBus instead.

      Specified by:
      subscriptionQuery in interface QueryGateway
      Type Parameters:
      R - the type payload to map the responses to
      Parameters:
      query - the query to be sent
      responseType - the response type returned by this query as the initial result
      mapper - a Function that maps the QueryResponseMessage to the desired response; messages for which the mapper returns a null value are filtered out
      context - the processing context, if any, to dispatch the given query in
      updateBufferSize - the size of the buffer which accumulates updates to be processed
      Returns:
      a Publisher which can be used to cancel receiving update
      See Also:

    • describeTo

      public void describeTo(ComponentDescriptor descriptor)
      Description copied from interface: DescribableComponent
      Describe the properties of this DescribableComponent with the given descriptor.

      Components should call the appropriate describeProperty methods on the descriptor to register their properties. The descriptor is responsible for determining how these properties are formatted and structured in the final output.

      Best Practices: As a general rule, all relevant fields of a DescribableComponent implementation should be described in this method. However, developers have discretion to include only the fields that make sense in the context. Not every field may be meaningful for description purposes, especially internal implementation details. Furthermore, components might want to expose different information based on their current state. The final decision on what properties to include lies with the person implementing the describeTo method, who should focus on providing information that is useful for understanding the component's configuration and state.

      Example implementation: 

       public void describeTo(ComponentDescriptor descriptor) {
     descriptor.describeProperty("name", this.name);
     descriptor.describeProperty("enabled", this.enabled);
     descriptor.describeProperty("configuration", this.configuration); // A nested component
     descriptor.describeProperty("handlers", this.eventHandlers);      // A collection
 }
      Specified by:
      describeTo in interface DescribableComponent
      Parameters:
      descriptor - The component descriptor to describe this DescribableComponentn its properties in.