Package org.axonframework.messaging.eventstreaming

Class MultiStreamableEventSource

java.lang.Object

org.axonframework.messaging.eventstreaming.MultiStreamableEventSource

All Implemented Interfaces:

StreamableEventSource, TrackingTokenSource

public class MultiStreamableEventSource
extends Object
implements StreamableEventSource

A StreamableEventSource implementation that allows streaming event processors to process events from multiple underlying event sources. This is useful when events need to be consumed from:

• Multiple event stores
• Multiple Axon Server contexts
• Different storage types (for example, an Event Store and a Kafka topic)

Events from the different sources are merged using a MergedMessageStream, with the order of event consumption determined by a configurable Comparator. The default comparator returns the oldest event available (based on the event's timestamp). Each source is tracked independently using a MultiSourceTrackingToken, which maintains the position for each individual source.

Since:

5.1.0

Author:

Allard Buijze, Greg Woods

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static interface	MultiStreamableEventSource.SourceCollector	Intermediate builder for collecting event sources before creating a MultiStreamableEventSource.

Field Summary

Fields

Modifier and Type	Field	Description
final Context.ResourceKey<String>	SOURCE_ID_RESOURCE	Resource key used to identify which source an event originated from.

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	MultiStreamableEventSource(Map<String,StreamableEventSource> eventSources, Comparator<MessageStream.Entry<EventMessage>> eventComparator)	Constructs a MultiStreamableEventSource from the collected sources and comparator.

Method Summary

Modifier and Type	Method	Description
static MultiStreamableEventSource.SourceCollector	combining(String sourceName, StreamableEventSource source)	Creates a new MultiStreamableEventSource by combining multiple event sources.
CompletableFuture<TrackingToken>	firstToken(@Nullable ProcessingContext context)	Creates a TrackingToken representing the first position of the event stream.
CompletableFuture<TrackingToken>	latestToken(@Nullable ProcessingContext context)	Creates a TrackingToken representing the latest position, thus pointing at the next event of the event stream.
MessageStream<EventMessage>	open(StreamingCondition condition, @Nullable ProcessingContext context)	Open an event stream containing all events matching the given condition.
CompletableFuture<TrackingToken>	tokenAt(Instant at, @Nullable ProcessingContext context)	Creates a TrackingToken tracking all events after the given at from an event stream.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

SOURCE_ID_RESOURCE

public final Context.ResourceKey<String> SOURCE_ID_RESOURCE

Resource key used to identify which source an event originated from. This key can be accessed in the eventComparator to determine the source of an event and apply custom ordering logic, such as giving precedence to events from a specific source.

Constructor Details

MultiStreamableEventSource

protected MultiStreamableEventSource(Map<String,StreamableEventSource> eventSources,
 Comparator<MessageStream.Entry<EventMessage>> eventComparator)

Constructs a MultiStreamableEventSource from the collected sources and comparator.

Parameters:

eventSources - The map of event sources, keyed by their unique names.

eventComparator - The comparator to use when deciding which message to process first.

Method Details

combining

public static MultiStreamableEventSource.SourceCollector combining(String sourceName,
 StreamableEventSource source)

Creates a new MultiStreamableEventSource by combining multiple event sources. This is the starting point for the fluent builder API.

Example usage:

 // With default timestamp-based comparison
 MultiStreamableEventSource source = MultiStreamableEventSource
     .combining("eventStore1", eventSource1)
     .and("eventStore2", eventSource2)
     .comparingTimestamps();

 // With custom comparator
 MultiStreamableEventSource source = MultiStreamableEventSource
     .combining("eventStore1", eventSource1)
     .and("eventStore2", eventSource2)
     .comparingUsing(customComparator);
 

Parameters:

sourceName - A unique name identifying the first source.

source - The first event source to include.

Returns:

A SourceCollector for adding more sources and configuring the comparator.

open

public MessageStream<EventMessage> open(StreamingCondition condition,
 @Nullable ProcessingContext context)

Description copied from interface: StreamableEventSource

Open an event stream containing all events matching the given condition.

To retrieve the position of the returned events, the TrackingToken.fromContext(Context) operation should be used by providing the entire MessageStream.Entry wrapping the returned events.

Note that the returned stream is infinite, so beware of applying terminal operations to the returned stream.

When all events are of interest during streaming, then use EventCriteria.havingAnyTag() as the condition criteria.

Specified by:

open in interface StreamableEventSource

Parameters:

condition - The StreamingCondition defining the starting position of the stream and event criteria to filter the stream with.

context - The current ProcessingContext, if any.

Returns:

An event stream matching the given condition.

firstToken

public CompletableFuture<TrackingToken> firstToken(@Nullable ProcessingContext context)

Description copied from interface: TrackingTokenSource

Creates a TrackingToken representing the first position of the event stream.

As the retrieved token represents the point from which to

invalid reference

open

the event stream, the first event to be streamed when opening is the one right after the returned token.

Subsequent invocation of this method will yield the same result, unless the stream's initial values are deleted.

Specified by:

firstToken in interface TrackingTokenSource

Parameters:

context - The current ProcessingContext, if any.

Returns:

A CompletableFuture of a TrackingToken representing the first event of the event stream.

latestToken

public CompletableFuture<TrackingToken> latestToken(@Nullable ProcessingContext context)

Description copied from interface: TrackingTokenSource

Creates a TrackingToken representing the latest position, thus pointing at the next event of the event stream.

As the retrieved token represents the point from which to

invalid reference

open

the event stream, the first event to be streamed when opening is the one right after the returned token.

Since the event stream of this source is theoretically infinite, subsequent invocation of this operation typically return a different token. Only if this StreamableEventSource is idle, will several latestToken() invocations result in the same TrackingToken.

Specified by:

latestToken in interface TrackingTokenSource

Parameters:

context - The current ProcessingContext, if any.

Returns:

A CompletableFuture of a TrackingToken representing the latest event, thus pointing at the next event of the event stream.

tokenAt

public CompletableFuture<TrackingToken> tokenAt(Instant at,
 @Nullable ProcessingContext context)

Description copied from interface: TrackingTokenSource

Creates a TrackingToken tracking all events after the given at from an event stream.

When there is an EventMessage exactly at the given dateTime, it will be tracked too.

Specified by:

tokenAt in interface TrackingTokenSource

Parameters:

at - The Instant determining how the TrackingToken should be created. The returned token points at very first event before this Instant.

context - The current ProcessingContext, if any.

Returns:

A CompletableFuture of TrackingToken pointing at the very first event before the given at of the event stream.