Package io.axoniq.framework.messaging.multitenancy.queryhandling

Class TenantAwareQueryBus

java.lang.Object
io.axoniq.framework.messaging.multitenancy.queryhandling.TenantAwareQueryBus
All Implemented Interfaces:
DescribableComponent, QueryBus, QueryHandlerRegistry<QueryBus>
@Internal public class TenantAwareQueryBus extends Object implements QueryBus
QueryBus decorator scoping subscription-query update emission and completion to the tenant resolved from the current ProcessingContext.

Subscription-query dispatch is already tenant-routed by the per-tenant QueryBusConnector composition (see MultiTenantAxonServerQueryBusConnector), but update emission is not: a distributed QueryBus tracks every active subscription in a single registry and matches emitUpdate(Predicate, Supplier, ProcessingContext)/ completeSubscriptions(Predicate, ProcessingContext)/ completeSubscriptionsExceptionally(Predicate, Throwable, ProcessingContext) calls against all of them, regardless of tenant. This decorator closes that gap by ANDing a tenant clause onto the caller's filter before delegating, so a matching QueryMessage only fires when its tenantId metadata equals the tenant resolved from the given context. Because QueryUpdateEmitter.forContext(ProcessingContext) always resolves its QueryBus from the context, every emitter obtained either by parameter injection or by a direct forContext call is scoped this way, regardless of how it was constructed.

query(QueryMessage, ProcessingContext) additionally rejects queries for a tenant that is not (or no longer) served, before delegating. A distributed QueryBus may serve a query from its local segment whenever a local handler is subscribed for it, bypassing the tenant-routing connector entirely, which would otherwise answer queries for a tenant that was never registered or whose context has been removed. Validating here keeps the outcome identical regardless of whether the query is served locally or dispatched through the connector.

The remaining dispatching methods (subscriptionQuery(QueryMessage, ProcessingContext, int), subscribeToUpdates(QueryMessage, int)) and subscribe(QualifiedName, QueryHandler) are pure pass-through: subscription queries always travel through the connector, which resolves the tenant itself.

Registered as a decorator on QueryBus by MultiTenancyConfigurationDefaults, positioned between DistributedQueryBus and InterceptingQueryBus in the decoration chain. Not intended to be instantiated directly by applications.

Since:
5.3.0
Author:
Jakob Hatzl, Laura Devriendt

  • Constructor Details

    • TenantAwareQueryBus

      public TenantAwareQueryBus(QueryBus delegate, TenantResolver tenantResolver, TenantDescriptors tenantDescriptors)
      Constructs a TenantAwareQueryBus, delegating all operations to the given delegate.
      Parameters:
      delegate - the QueryBus to delegate all operations to
      tenantResolver - the TenantResolver used for tenant resolution from query messages
      tenantDescriptors - the currently served TenantDescriptors, used to reject queries for tenants that are not served

  • Method Details

    • subscribe

      public TenantAwareQueryBus subscribe(QualifiedName queryName, QueryHandler queryHandler)
      Description copied from interface: QueryHandlerRegistry
      Subscribe the given queryHandler for queries and response of the given queryName.

      If a subscription already exists for the queryName, the behavior is undefined. Implementations may throw an exception to refuse duplicate subscription or alternatively decide whether the existing or new handler gets the subscription.

      Specified by:
      subscribe in interface QueryHandlerRegistry<QueryBus>
      Parameters:
      queryName - The fully qualified name of the query
      queryHandler - The handler instance that handles queries for the given queryName.
      Returns:
      This registry for fluent interfacing.

    • query

      public MessageStream<QueryResponseMessage> query(QueryMessage query, @Nullable ProcessingContext context)
      Dispatches the given query for the tenant resolved from it.

      Rejects the query with a TenantNotResolvedException when its resolved tenant is not among the currently served tenants, rather than leaving that verdict to the tenant-routing connector: a distributed QueryBus serves queries with a locally subscribed handler from its local segment, never reaching the connector.

      Specified by:
      query in interface QueryBus
      Parameters:
      query - the query to dispatch
      context - the processing context under which the query is dispatched (can be null)
      Returns:
      a MessageStream of the responses for the given query
      Throws:
      TenantNotResolvedException - if no tenant can be resolved from the given query, or the resolved tenant is not served

    • subscriptionQuery

      public MessageStream<QueryResponseMessage> subscriptionQuery(QueryMessage query, @Nullable ProcessingContext context, 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 update, the query bus will queue all update which happen after the subscription query is done and once the subscription to the flux is made, these update will be emitted.

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

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

      If a subscription query with the same query identifier is already registered, the returned MessageStream will be failed with a SubscriptionQueryAlreadyRegisteredException instead of throwing the exception. This allows callers to handle double subscription scenarios gracefully through the stream API.

      Specified by:
      subscriptionQuery in interface QueryBus
      Parameters:
      query - the subscription query to dispatch
      context - the processing context under which the query is being published (can be null)
      updateBufferSize - the size of the buffer which accumulates update
      Returns:
      query result containing initial result and incremental update, or a failed MessageStream with SubscriptionQueryAlreadyRegisteredException if a subscription with the same query identifier already exists

    • subscribeToUpdates

      public MessageStream<SubscriptionQueryUpdateMessage> subscribeToUpdates(QueryMessage query, int updateBufferSize)
      Description copied from interface: QueryBus
      Subscribes the given query with the given updateBufferSize, and returns the MessageStream that provides the update of the subscription query.

      Can be used directly instead when fine-grained control of update handlers is required. If using the update directly is not mandatory for your use case, we strongly recommend using QueryBus.subscriptionQuery(QueryMessage, ProcessingContext, int) instead.

      Note that the returned MessageStream must be 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.

      If a subscription query with the same query identifier is already registered, the returned MessageStream will be failed with a SubscriptionQueryAlreadyRegisteredException instead of throwing the exception.

      Specified by:
      subscribeToUpdates in interface QueryBus
      Parameters:
      query - the subscription query for which we register an update handler
      updateBufferSize - the size of the buffer that accumulates update
      Returns:
      a MessageStream of update for the given subscription query, or a failed MessageStream with SubscriptionQueryAlreadyRegisteredException if a subscription with the same query identifier already exists

    • emitUpdate

      public CompletableFuture<Void> emitUpdate(Predicate<QueryMessage> filter, Supplier<SubscriptionQueryUpdateMessage> updateSupplier, @Nullable ProcessingContext context)
      Emits the outcome of the updateSupplier to subscription queries matching the given queryName and given filter.

      ANDs a tenant clause onto the given filter, matching only QueryMessages whose resolved tenant equals the tenant resolved from the given context.

      Specified by:
      emitUpdate in interface QueryBus
      Parameters:
      filter - a predicate filtering on QueryMessages; the updateSupplier is only sent to subscription queries matching this filter
      updateSupplier - the update supplier to emit for subscription queries matching the given filter
      context - the processing context under which the updateSupplier is being emitted; this is needed to resolve the actual tenant; a null context submitted here will produce a TenantNotResolvedException immediately
      Returns:
      a future completing whenever the updateSupplier has been emitted
      Throws:
      TenantNotResolvedException - in case no context is supplied

    • emitUpdateAndCount

      public CompletableFuture<OptionalInt> emitUpdateAndCount(Predicate<QueryMessage> filter, Supplier<SubscriptionQueryUpdateMessage> updateSupplier, @Nullable ProcessingContext context)
      Emits the outcome of the updateSupplier to subscription queries matching the given queryName and given filter, returning the number of subscription queries the update was emitted to.

      Implementations that cannot determine this number return OptionalInt.empty() instead.

      ANDs a tenant clause onto the given filter, matching only QueryMessages whose resolved tenant equals the tenant resolved from the given context.

      Specified by:
      emitUpdateAndCount in interface QueryBus
      Parameters:
      filter - a predicate filtering on QueryMessages. The updateSupplier will only be sent to subscription queries matching this filter
      updateSupplier - the update supplier to emit for subscription queries matching the given filter
      context - the processing context under which the updateSupplier is being emitted; this is needed to resolve the actual tenant; a null context submitted here will produce a TenantNotResolvedException immediately
      Returns:
      a future completing with the number of subscription queries the update was emitted to as an OptionalInt, which is empty when we couldn't match
      Throws:
      TenantNotResolvedException - in case no context is supplied

    • completeSubscriptions

      public CompletableFuture<Void> completeSubscriptions(Predicate<QueryMessage> filter, @Nullable ProcessingContext context)
      Completes subscription queries matching the given filter.

      To be used whenever there are no subsequent update to emit left.

      ANDs a tenant clause onto the given filter, matching only QueryMessages whose resolved tenant equals the tenant resolved from the given context.

      Specified by:
      completeSubscriptions in interface QueryBus
      Parameters:
      filter - a predicate filtering on QueryMessages; subscription queries matching this filter are completed
      context - the processing context within which to complete subscription queries (can be null); this is needed to resolve the actual tenant; a null context submitted here will produce a TenantNotResolvedException immediately
      Returns:
      a future completing whenever all matching subscription queries have been completed
      Throws:
      TenantNotResolvedException - in case no context is supplied

    • completeSubscriptionsAndCount

      public CompletableFuture<OptionalInt> completeSubscriptionsAndCount(Predicate<QueryMessage> filter, @Nullable ProcessingContext context)
      Completes subscription queries matching the given filter, returning the number of subscription queries that were completed.

      Implementations that cannot determine this number return OptionalInt.empty() instead.

      ANDs a tenant clause onto the given filter, matching only QueryMessages whose resolved tenant equals the tenant resolved from the given context.

      Specified by:
      completeSubscriptionsAndCount in interface QueryBus
      Parameters:
      filter - a predicate filtering on QueryMessages. Subscription queries matching this filter will be completed
      context - the processing context within which to complete subscription queries (can be null); this is needed to resolve the actual tenant; a null context submitted here will produce a TenantNotResolvedException immediately
      Returns:
      a future completing with the number of subscription queries that were completed as an OptionalInt, which is empty when we couldn't match
      Throws:
      TenantNotResolvedException - in case no context is supplied

    • completeSubscriptionsExceptionally

      public CompletableFuture<Void> completeSubscriptionsExceptionally(Predicate<QueryMessage> filter, Throwable cause, @Nullable ProcessingContext context)
      Completes subscription queries matching the given filter exceptionally with the given cause.

      To be used whenever emitting update should be stopped due to some exception.

      ANDs a tenant clause onto the given filter, matching only QueryMessages whose resolved tenant equals the tenant resolved from the given context.

      Specified by:
      completeSubscriptionsExceptionally in interface QueryBus
      Parameters:
      filter - a predicate filtering on QueryMessages; subscription queries matching this filter are completed exceptionally
      cause - the cause of an error
      context - the processing context within which to complete subscription queries exceptionally (can be null); this is needed to resolve the actual tenant; a null context submitted here will produce a TenantNotResolvedException immediately
      Returns:
      a future completing whenever all matching subscription queries have been completed exceptionally
      Throws:
      TenantNotResolvedException - in case no context is supplied

    • completeSubscriptionsExceptionallyAndCount

      public CompletableFuture<OptionalInt> completeSubscriptionsExceptionallyAndCount(Predicate<QueryMessage> filter, Throwable cause, @Nullable ProcessingContext context)
      Completes subscription queries matching the given filter exceptionally with the given cause, returning the number of subscription queries that were completed exceptionally.

      Implementations that cannot determine this number return OptionalInt.empty() instead.

      ANDs a tenant clause onto the given filter, matching only QueryMessages whose resolved tenant equals the tenant resolved from the given context.

      Specified by:
      completeSubscriptionsExceptionallyAndCount in interface QueryBus
      Parameters:
      filter - a predicate filtering on QueryMessages. Subscription queries matching this filter will be completed exceptionally
      cause - the cause of an error
      context - the processing context within which to complete subscription queries exceptionally (can be null); this is needed to resolve the actual tenant; a null context submitted here will produce a TenantNotResolvedException immediately
      Returns:
      a future completing with the number of subscription queries that were completed exceptionally as an OptionalInt, which is empty when we couldn't match
      Throws:
      TenantNotResolvedException - in case no context is supplied

    • 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.