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.

close() still delivers every event already buffered locally at the moment it is called to the consumer, and only retires that consumer, in favor of a no-op, once all of it has been delivered.

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 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, PersistentStreamContextCustomizer contextCustomizer, int batchSize, @Nullable String context)
      Instantiates a PersistentStreamConnection placing additional resources on the ProcessingContext of every batch through the given contextCustomizer.
      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
      contextCustomizer - the customizer placing resources on the ProcessingContext of every batch, invoked once per batch before any of its events is consumed, returning the context that batch is consumed with
      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.

      Events already buffered locally for a segment at the moment this method is called are still delivered to the consumer supplied to open(BiFunction). The consumer is only retired, in favor of a no-op, once every currently open segment has confirmed its local buffer is fully drained. Otherwise, those buffered events would be acknowledged to Axon Server without ever reaching a real consumer.