Package io.axoniq.framework.messaging.multitenancy.axonserver.eventstreaming

Class MultiTenantPersistentStreamEventSource

java.lang.Object
io.axoniq.framework.messaging.multitenancy.axonserver.eventstreaming.MultiTenantPersistentStreamEventSource
All Implemented Interfaces:
MultiTenantAwareComponent, DescribableComponent, SubscribableEventSource
@Internal public class MultiTenantPersistentStreamEventSource extends Object implements SubscribableEventSource, MultiTenantAwareComponent
A SubscribableEventSource consuming one persistent stream per tenant, labelling every event it delivers with the tenant it came from.

Each tenant is an Axon Server context, and a persistent stream lives in one context, so this source holds one ordinary PersistentStreamEventSource per tenant and fans a single consumer out to all of them. The stream carries the configured name in every tenant's context, so a tenant's stream appears in Axon Server named exactly as configured.

This source follows the tenant lifecycle only while it has a subscriber. subscribe(BiFunction) subscribes it to the TenantProvider as a MultiTenantAwareComponent, which immediately replays the known tenants and so opens a stream for each, while tenants added or removed later arrive through registerAndStartTenant(TenantDescriptor) and the cancellation of their registration. Cancelling the subscription unsubscribes from the provider again, which closes every tenant's stream and releases its scheduler. Binding the tenant subscription to the consumer's, rather than to a lifecycle phase, keeps the two in step: no stream is ever open without a consumer to feed, and none stays open once the consumer is gone.

Marked Internal as concrete, internal implementation behind the MultiTenantPersistentStreamEventSourceFactory.

Since:
5.3.0
Author:
Jakob Hatzl
See Also:

  • Constructor Details

    • MultiTenantPersistentStreamEventSource

      public MultiTenantPersistentStreamEventSource(String name, io.axoniq.axonserver.connector.event.PersistentStreamProperties properties, Function<String,ScheduledExecutorService> schedulerFactory, int batchSize, Configuration configuration, TenantProvider tenantProvider)
      Constructs a MultiTenantPersistentStreamEventSource for the persistent stream with the given name.
      Parameters:
      name - the name of the persistent stream, used as the stream identifier in every tenant's Axon Server context
      properties - the properties applied when creating each tenant's persistent stream
      schedulerFactory - the factory creating a tenant's ScheduledExecutorService for the pool name given to it, so each tenant's stream runs on threads of its own rather than sharing one pool
      batchSize - the maximum number of events to deliver per batch
      configuration - the configuration supplying the components each tenant's stream is built from
      tenantProvider - the provider whose tenants this source opens a stream for while it has a subscriber
      Throws:
      NullPointerException - if any of the given arguments is null
      IllegalArgumentException - if the given name is empty or batchSize is not positive

  • Method Details

    • subscribe

      public Registration subscribe(BiFunction<List<? extends EventMessage>,ProcessingContext,CompletableFuture<?>> eventsBatchConsumer)
      Subscribes the given eventsBatchConsumer to the persistent stream of every tenant, opening a stream for each tenant known to the TenantProvider and for every tenant added while the subscription lasts.

      Only one consumer can be subscribed at a time, matching the single-tenant behaviour: subscribing the same consumer again is a no-op, while a different one is rejected. Cancelling the returned Registration closes every tenant's stream and releases its scheduler, after which this source can be subscribed again.

      Specified by:
      subscribe in interface SubscribableEventSource
      Parameters:
      eventsBatchConsumer - the consumer receiving batches of events, each with the tenant of its stream on the ProcessingContext
      Returns:
      a registration closing every tenant's stream on cancellation
      Throws:
      IllegalStateException - if another consumer is already subscribed

    • registerTenant

      public Registration registerTenant(TenantDescriptor tenantDescriptor)
      Opens the persistent stream of the given tenantDescriptor and joins it to the subscribed consumer.

      Called by the TenantProvider for each tenant known when this source subscribed to it. The streams of the other tenants are untouched, so a tenant arriving here does not pause their processing.

      Specified by:
      registerTenant in interface MultiTenantAwareComponent
      Parameters:
      tenantDescriptor - the tenant to register with this source
      Returns:
      a registration closing the tenant's stream and releasing its scheduler on cancellation

    • registerAndStartTenant

      public Registration registerAndStartTenant(TenantDescriptor tenantDescriptor)
      Behaves identically to registerTenant(TenantDescriptor), which already opens the tenant's stream: this source is only registered with the TenantProvider while it has a consumer to feed, so there is no registered-but-not-started state to distinguish.
      Specified by:
      registerAndStartTenant in interface MultiTenantAwareComponent
      Parameters:
      tenantDescriptor - the tenant to register and start with this source
      Returns:
      a registration closing the tenant's stream and releasing its scheduler on cancellation

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