Package io.axoniq.framework.axonserver.connector.event

Class PersistentStreamEventSource

java.lang.Object

io.axoniq.framework.axonserver.connector.event.PersistentStreamEventSource

All Implemented Interfaces:

SubscribableEventSource

public class PersistentStreamEventSource
extends Object
implements SubscribableEventSource

A SubscribableEventSource that receives events from a persistent stream on Axon Server.

The persistent stream is identified by a unique name, which acts as the stream identifier in Axon Server. Using the same name for different instances will join the same server-side stream. Each instance owns an exclusive PersistentStreamConnection and may have at most one active subscriber at a time.

Since:

5.2.0

Author:

Marc Gathier, Jakob Hatzl

Constructor Summary

Constructors

Constructor	Description
PersistentStreamEventSource(String name, AxonServerConnectionManager connectionManager, AxonServerConfiguration serverConfig, EventConverter converter, EventTypeResolver eventTypeResolver, io.axoniq.axonserver.connector.event.PersistentStreamProperties persistentStreamProperties, ScheduledExecutorService scheduler, UnitOfWorkFactory unitOfWorkFactory, int batchSize)	Instantiates a PersistentStreamEventSource.
PersistentStreamEventSource(String name, AxonServerConnectionManager connectionManager, AxonServerConfiguration serverConfig, EventConverter converter, EventTypeResolver eventTypeResolver, io.axoniq.axonserver.connector.event.PersistentStreamProperties persistentStreamProperties, ScheduledExecutorService scheduler, UnitOfWorkFactory unitOfWorkFactory, int batchSize, @Nullable String context)	Instantiates a PersistentStreamEventSource.

Method Summary

Modifier and Type	Method	Description
Registration	subscribe(BiFunction<List<? extends EventMessage>,ProcessingContext,CompletableFuture<?>> eventsBatchConsumer)	Subscribe the given eventsBatchConsumer to this event source.

Methods inherited from class java.lang.Object

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

Constructor Details

PersistentStreamEventSource

public PersistentStreamEventSource(String name,
 AxonServerConnectionManager connectionManager,
 AxonServerConfiguration serverConfig,
 EventConverter converter,
 EventTypeResolver eventTypeResolver,
 io.axoniq.axonserver.connector.event.PersistentStreamProperties persistentStreamProperties,
 ScheduledExecutorService scheduler,
 UnitOfWorkFactory unitOfWorkFactory,
 int batchSize)

Instantiates a PersistentStreamEventSource.

Parameters:

name - the name of the persistent stream; acts as the unique stream identifier in Axon Server

connectionManager - the Axon Server connection manager

serverConfig - the Axon Server configuration

converter - the event converter used to deserialize event payloads

eventTypeResolver - the event type resolver used to resolve the type on inbound events

persistentStreamProperties - the properties for the persistent stream

scheduler - the scheduler thread pool to schedule tasks

unitOfWorkFactory - the UnitOfWorkFactory used to create a unit of work to span message processing

batchSize - the maximum number of events to collect per batch

PersistentStreamEventSource

public PersistentStreamEventSource(String name,
 AxonServerConnectionManager connectionManager,
 AxonServerConfiguration serverConfig,
 EventConverter converter,
 EventTypeResolver eventTypeResolver,
 io.axoniq.axonserver.connector.event.PersistentStreamProperties persistentStreamProperties,
 ScheduledExecutorService scheduler,
 UnitOfWorkFactory unitOfWorkFactory,
 int batchSize,
 @Nullable String context)

Instantiates a PersistentStreamEventSource.

Parameters:

name - the name of the persistent stream; acts as the unique stream identifier in Axon Server

connectionManager - the Axon Server connection manager

serverConfig - the Axon Server configuration

converter - the event converter used to deserialize event payloads

eventTypeResolver - the event type resolver used to resolve the type on inbound events

persistentStreamProperties - the properties for the persistent stream

scheduler - the scheduler thread pool to schedule tasks

unitOfWorkFactory - the UnitOfWorkFactory used to create a unit of work to span message processing

batchSize - the maximum number of events to collect per batch

context - the Axon Server context in which this stream exists, or null to use the context from AxonServerConfiguration.getContext()

Method Details

subscribe

public Registration subscribe(BiFunction<List<? extends EventMessage>,ProcessingContext,CompletableFuture<?>> eventsBatchConsumer)

Description copied from interface: SubscribableEventSource

Subscribe the given eventsBatchConsumer to this event source. When subscribed, it will receive all events published to this source since the subscription.

If the given eventsBatchConsumer is already subscribed, nothing happens.

Note on ProcessingContext: The ProcessingContext parameter passed to the consumer may be null. When null, it is the responsibility of the registered eventsBatchConsumer to create an appropriate ProcessingContext as needed for processing the events.

Specified by:

subscribe in interface SubscribableEventSource

Parameters:

eventsBatchConsumer - The event batches consumer to subscribe.

Returns:

A Registration handle to unsubscribe the eventsBatchConsumer. When unsubscribed, it will no longer receive events.