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

Class PersistentStreamConnection

java.lang.Object

io.axoniq.framework.axonserver.connector.event.PersistentStreamConnection

@Internal
public class PersistentStreamConnection
extends Object

A connection instance receiving the events for a persistent stream to pass on in batches to an event consumer.

Opens a gRPC connection to Axon Server via open(BiFunction) and invokes the supplied BiFunction for each batch. The batch consumer returns a CompletableFuture that must complete before the token for the last event in the batch is acknowledged to Axon Server. On consumer failure the batch is retried with exponential back-off.

This is an internal helper for the PersistentStreamEventSource managing the gRPC based persistent stream with Axon Server. This class is usually not used directly by users.

Since:

5.2.0

Author:

Marc Gathier, Jakob Hatzl

Constructor Summary

Constructors

Constructor	Description
PersistentStreamConnection(String streamId, AxonServerConnectionManager connectionManager, AxonServerConfiguration serverConfig, EventConverter converter, io.axoniq.axonserver.connector.event.PersistentStreamProperties persistentStreamProperties, ScheduledExecutorService scheduler, UnitOfWorkFactory unitOfWorkFactory, int batchSize, @Nullable String context)	Instantiates a PersistentStreamConnection falling back to the default event type resolver for message type resolution.
PersistentStreamConnection(String streamId, 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 PersistentStreamConnection.

Method Summary

Modifier and Type	Method	Description
void	close()	Closes the persistent stream connection to Axon Server.
void	open(BiFunction<List<? extends EventMessage>,ProcessingContext,CompletableFuture<?>> consumer)	Initiates the connection to Axon Server and starts delivering events to the given consumer.

Methods inherited from class java.lang.Object

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

Constructor Details

PersistentStreamConnection

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

Instantiates a PersistentStreamConnection falling back to the default event type resolver for message type resolution.

Parameters:

streamId - the unique identifier of the persistent stream

connectionManager - the Axon Server connection manager

serverConfig - the Axon Server configuration

converter - the event converter used to deserialize event payloads

persistentStreamProperties - the properties for the persistent stream

scheduler - the scheduler thread pool to schedule tasks

unitOfWorkFactory - the unit of work factory

batchSize - the maximum number of events to collect per batch

context - the Axon Server context to connect to, or null to use AxonServerConfiguration.getContext()

PersistentStreamConnection

public PersistentStreamConnection(String streamId,
 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 PersistentStreamConnection.

Parameters:

streamId - the unique identifier of the persistent stream

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 unit of work factory

batchSize - the maximum number of events to collect per batch

context - the Axon Server context to connect to, or null to use AxonServerConfiguration.getContext()

Method Details

open

public void open(BiFunction<List<? extends EventMessage>,ProcessingContext,CompletableFuture<?>> consumer)

Initiates the connection to Axon Server and starts delivering events to the given consumer.

The stream can be opened with only a single consumer at a time. After a previous close(), the stream may be reopened by calling this method again.

Parameters:

consumer - the consumer of batches of event messages; to allow providing tracking and replay information per event, it receives each event in a single callback ProcessingContext enriched with the current TrackingToken and, when available, aggregate identity information, and must return a CompletableFuture that completes when the event has been processed; events are still processed in batches as configured through the batchSize constructor argument, the persistent stream connection takes care of creating and spanning a unit of work over all events belonging to a single batch

Throws:

IllegalStateException - if the stream was already opened

close

public void close()

Closes the persistent stream connection to Axon Server.