Package io.axoniq.framework.messaging.multitenancy.eventsourcing

Class MultiTenantEventStorageEngine

java.lang.Object
io.axoniq.framework.messaging.multitenancy.eventsourcing.MultiTenantEventStorageEngine
All Implemented Interfaces:
MultiTenantAwareComponent, DescribableComponent, EventStorageEngine
@Internal public class MultiTenantEventStorageEngine extends Object implements EventStorageEngine, MultiTenantAwareComponent
Tenant-routing EventStorageEngine. Appends and sourcing are routed to the engine of the one tenant resolved from the ProcessingContext, so each tenant's events live in its own store.

Like the other multi-tenant infrastructure components, this engine holds the tenant information, the registration of tenants, and the routing to them. Snapshot writes are routed the same way by MultiTenantSnapshotStore, a component of its own so that resolving an EventStorageEngine or a SnapshotStore by type stays unambiguous.

The SourcingCondition is routed unchanged, so a snapshot sourcing strategy reaches the tenant's own engine rather than being resolved above the fan-out, where no tenant is known yet. Keeping it intact requires the application-wide snapshot composition to be switched off, which the configuration enhancer registering this engine does.

Each tenant's engine is composed once with that tenant's snapshot store through decorate, applying the same rule the event sourcing defaults apply to a single-tenant engine. An engine that is its own snapshot store serves a snapshot sourcing strategy within one call and is left untouched. Any other engine is decorated with that tenant's snapshot store, resolving the snapshot first and sourcing the events following it. Both stay within one tenant.

A tenant-carrying processing context is required for appends and sourcing. When none is available, or the tenant cannot be resolved from it, the operation fails. An append without a context resolves its tenant from the events instead, which must then all belong to the same tenant.

As a MultiTenantAwareComponent this engine follows the TenantProvider: a tenant added at runtime gets its engine on first use, and a removed tenant's composed engine is evicted. Each such registration is announced to the listeners subscribed through subscribe(TenantChangeListener), for a component that has to act on the tenants this engine holds rather than on the tenants the provider knows. Re-registering an already registered tenant is announced too, since that rebuilds the tenant's composed engine.

The read side (stream(org.axonframework.messaging.eventstreaming.StreamingCondition), firstToken(), latestToken(), tokenAt(java.time.Instant)) carries no context and spans all current tenants. It merges the per-tenant streams, tags every event with its tenant, and positions the merged stream with a MultiTenantTrackingToken holding one position per tenant. A tenant added while a processor runs streams from its beginning once the processor re-opens the stream.

Since:
5.3.0
Author:
Jakob Hatzl, Laura Devriendt

  • Constructor Details

  • Method Details

    • appendEvents

      public CompletableFuture<EventStorageEngine.AppendTransaction<?>> appendEvents(AppendCondition condition, @Nullable ProcessingContext context, List<TaggedEventMessage<?>> events)
      Description copied from interface: EventStorageEngine
      Appends a List of events to the underlying storage solution.

      Events will be appended in the order that they are offered in, validating the given condition before being stored. Note that all events should have a unique event identifier. Tags paired with the events will be stored as well.

      Implementations may be able to detect conflicts during the append stage. In such case, the returned completable future will complete exceptionally, indicating such conflict. Other implementations may delay such checks until the EventStorageEngine.AppendTransaction.commit() is called.

      Called during the PREPARE_COMMIT phase.

      Specified by:
      appendEvents in interface EventStorageEngine
      Parameters:
      condition - The condition describing the transactional requirements for the append transaction
      context - The current ProcessingContext, if any.
      events - The List of events to append to the underlying storage solution.
      Returns:
      A transaction instance that can be committed or rolled back.

    • source

      public MessageStream<EventMessage> source(SourcingCondition condition, @Nullable ProcessingContext context)
      Description copied from interface: EventStorageEngine
      Creates a finite MessageStream of events matching the given condition, carrying along the active ProcessingContext.

      Behaves identically to EventStorageEngine.source(SourcingCondition); the context is passed to decorators (for example snapshot-loading and tracing decorators) that need to correlate the sourcing operation with the surrounding unit of work.

      Specified by:
      source in interface EventStorageEngine
      Parameters:
      condition - the SourcingCondition dictating the stream of events to source
      context - the ProcessingContext active while sourcing; may be null
      Returns:
      a finite MessageStream of events matching the given condition

    • tenants

      public List<TenantDescriptor> tenants()
      Returns the tenants currently registered with this engine.
      Returns:
      the tenants currently registered with this engine

    • subscribe

      @Internal public Registration subscribe(TenantChangeListener listener)
      Subscribes the given listener to the changes of the tenants this engine holds.

      A change is announced only once it is visible through tenants(), which is what makes acting on one safe. A listener must therefore subscribe here rather than to the TenantProvider: the provider notifies its subscribers in turn, so a subscriber of the provider can observe this engine before it registered the tenant.

      Every call adds a listener of its own, so a listener subscribed twice is invoked twice for one change. Each returned registration removes what its own call added.

      Internal, because the only listeners are the module's own components, and what they are told is a detail of how this engine holds its tenants.

      Announcing happens on the thread applying the change, so a listener must return promptly and hand off any work of its own. A listener cancelled while an announcement is in flight may still be invoked for that announcement, so it has to tolerate running once more after its own cancellation.

      Parameters:
      listener - the listener to invoke after every change to the tenants this engine holds
      Returns:
      a registration whose cancellation stops the given listener from being invoked further
      Throws:
      NullPointerException - if the given listener is null

    • registerTenant

      public Registration registerTenant(TenantDescriptor tenantDescriptor)
      Description copied from interface: MultiTenantAwareComponent
      Registers the given tenantDescriptor as a known tenant with this multi-tenant aware component.

      The caller must retain the returned Registration and cancel it when the tenant is removed, since releasing the component's per-tenant resources rides on that cancellation.

      Specified by:
      registerTenant in interface MultiTenantAwareComponent
      Parameters:
      tenantDescriptor - The TenantDescriptor to register with this component.
      Returns:
      A Registration used to deregister the given tenantDescriptor.

    • registerAndStartTenant

      public Registration registerAndStartTenant(TenantDescriptor tenantDescriptor)
      Description copied from interface: MultiTenantAwareComponent
      Registers the given tenantDescriptor as a known tenant with this multi-tenant aware component. If applicable, this task will construct a tenant segment and start it.
      Specified by:
      registerAndStartTenant in interface MultiTenantAwareComponent
      Parameters:
      tenantDescriptor - The TenantDescriptor to register with this component.
      Returns:
      A Registration used to deregister the given tenantDescriptor.

    • stream

      public MessageStream<EventMessage> stream(StreamingCondition condition)
      Description copied from interface: EventStorageEngine
      Creates an infinite MessageStream of events matching the given condition.

      The condition may dictate the StreamingCondition.position() to start streaming from, as well as define filter criteria for the returned MessageStream.

      Specified by:
      stream in interface EventStorageEngine
      Parameters:
      condition - The StreamingCondition dictating the StreamingCondition.position() to start streaming from, as well as the filter criteria used for the returned MessageStream.
      Returns:
      An infinite MessageStream of events matching the given condition.

    • firstToken

      public CompletableFuture<TrackingToken> firstToken()
      Returns a token holding no per-tenant position, so a processor resuming from it opens every tenant at the beginning of its own store.

      The beginning of a tenant's store is not assumed to be the zero position. Streaming from this token fills in each tenant's own first token when it opens that tenant's stream, so a tenant whose early events were pruned still opens at its real first event. Naming those positions in this token instead would make it disagree with a token written before a tenant existed, and a streaming processor compares the two to recognize an event it already handled, so they are left out.

      Specified by:
      firstToken in interface EventStorageEngine
      Returns:
      A CompletableFuture of a TrackingToken at the first position of an event stream.

    • latestToken

      public CompletableFuture<TrackingToken> latestToken()
      Description copied from interface: EventStorageEngine
      Creates a TrackingToken that is at the latest position of an event stream.

      In other words, a token that tracks all new events from this point forward.

      Specified by:
      latestToken in interface EventStorageEngine
      Returns:
      A CompletableFuture of a TrackingToken at the latest position of an event stream.

    • tokenAt

      public CompletableFuture<TrackingToken> tokenAt(Instant at)
      Description copied from interface: EventStorageEngine
      Creates a TrackingToken that tracks all events after the given at.

      If there is an event exactly at the given at, it will be tracked too.

      Specified by:
      tokenAt in interface EventStorageEngine
      Parameters:
      at - The Instant determining how the TrackingToken should be created. A tracking token should point to very first event before this Instant.
      Returns:
      A CompletableFuture of a TrackingToken at the given at, if there aren't events matching this criteria null is returned

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