Module io.axoniq.connector.axonserver

Package io.axoniq.axonserver.connector.event.impl

Class EventChannelImpl

java.lang.Object

io.axoniq.axonserver.connector.impl.AbstractAxonServerChannel<Void>

io.axoniq.axonserver.connector.event.impl.EventChannelImpl

All Implemented Interfaces:

EventChannel

public class EventChannelImpl
extends AbstractAxonServerChannel<Void>
implements EventChannel

EventChannel implementation, serving as the event connection between AxonServer and a client application.

Constructor Summary

Constructors

Constructor	Description
EventChannelImpl(ClientIdentification clientIdentification, ScheduledExecutorService executor, AxonServerManagedChannel channel)	Constructs a EventChannelImpl.

Method Summary

Modifier and Type	Method	Description
CompletableFuture<Confirmation>	appendSnapshot(Event snapshotEvent)	Store the given snapshotEvent.
CompletableFuture<InstructionAck>	cancelSchedule(String token)	Cancels the scheduled publication of an event for which the given scheduleToken was returned.
void	connect()	Connect this channel with AxonServer.
CompletableFuture<CreateResult>	createPersistentStream(String streamId, PersistentStreamProperties creationProperties)	Opens the persistent stream identified by streamId.
CompletableFuture<Void>	deletePersistentStream(String streamId)	Deletes a persistent stream.
void	disconnect()	Disconnect this channel from AxonServer.
CompletableFuture<Long>	findHighestSequence(String aggregateId)	Find the highest sequence number (i.e. the sequence number of the most recent event) for an aggregate with given aggregateId.
CompletableFuture<Long>	getFirstToken()	Retrieves the Token referring to the first Event in the Event Store.
CompletableFuture<Long>	getLastToken()	Retrieves the Token referring to the most recent Event in the Event Store.
CompletableFuture<Long>	getTokenAt(long instant)	Retrieves the token referencing the first Event in the Event Store closest to the given instant.
boolean	isReady()	Validate whether this channel has all required streams connected with AxonServer.
AggregateEventStream	loadSnapshots(String aggregateIdentifier, long initialSequence, long maxSequence, int maxResults)	Loads Snapshot Events for the given aggregateIdentifier with sequence number between initialSequence and maxSequence (inclusive), returning at most maxResults number of snapshots.
AggregateEventStream	openAggregateStream(String aggregateIdentifier, boolean allowSnapshots)	Opens a stream for consuming Events from a single aggregate, with given allowSnapshots indicating whether the first Event may be a Snapshot Event.
AggregateEventStream	openAggregateStream(String aggregateIdentifier, long initialSequence, long maxSequence)	Opens a stream for consuming Events from a single aggregate, starting with the given initialSequence until the given maxSequence.
PersistentStream	openPersistentStream(String streamId, int bufferSize, int refillBatch, PersistentStreamCallbacks callbacks)	Opens the persistent stream identified by streamId.
PersistentStream	openPersistentStream(String streamId, int bufferSize, int refillBatch, PersistentStreamCallbacks callbacks, PersistentStreamProperties creationProperties)	Opens the persistent stream identified by streamId.
EventStream	openStream(long token, int bufferSize, int refillBatch, boolean forceReadFromLeader)	Open an EventStream, for sequentially consuming events from AxonServer, starting at given token and keeping a local buffer of bufferSize, which is refilled after consuming refillBatch items.
CompletableFuture<List<StreamStatus>>	persistentStreams()	Returns a list of persistent streams with the last confirmed token per segment.
ResultStream<EventQueryResultEntry>	queryEvents(String queryExpression, boolean liveStream, String contextName)	Queries the Event Store for events using given queryExpression.
ResultStream<EventQueryResultEntry>	querySnapshotEvents(String queryExpression, boolean liveStream, String contextName)	Queries the Event Store for snapshot events using given queryExpression.
void	reconnect()	Reconnect this channel with AxonServer.
CompletableFuture<String>	reschedule(String scheduleToken, Instant scheduleTime, Event event)	Convenience method to cancel the scheduled event with given scheduleToken and reschedule the given event to be published at given scheduleTime.
CompletableFuture<Void>	resetPersistentStreamAtHead(String streamId)	Resets the position of a persistent stream to the head (most recent) event in the event store.
CompletableFuture<Void>	resetPersistentStreamAtPosition(String streamId, long position)	Resets the position of a persistent stream to the event at given position (global index) in the event store.
CompletableFuture<Void>	resetPersistentStreamAtTail(String streamId)	Resets the position of a persistent stream to the first (oldest) event in the event store.
CompletableFuture<Void>	resetPersistentStreamAtTime(String streamId, long instant)	Resets the position of a persistent stream to the event at or after the given timestamp in the event store.
CompletableFuture<String>	scheduleEvent(Instant scheduleTime, Event event)	Schedule the given event to be published at given scheduleTime.
CompletableFuture<Void>	setPersistentStreamSegments(String streamId, int segments)	Updates the number of segments for a persistent stream.
AppendEventsTransaction	startAppendEventsTransaction()	Starts a new transaction to append events.
CompletableFuture<Void>	updatePersistentStreamName(String streamId, String streamName)	Updates the name for a persistent stream.

Methods inherited from class io.axoniq.axonserver.connector.impl.AbstractAxonServerChannel

processAck, scheduleImmediateReconnect, scheduleReconnect, sendInstruction

Methods inherited from class java.lang.Object

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

Methods inherited from interface io.axoniq.axonserver.connector.event.EventChannel

appendEvents, loadSnapshot, loadSnapshots, openAggregateStream, openAggregateStream, openStream, openStream, queryEvents, querySnapshotEvents, reschedule, scheduleEvent

Constructor Detail

EventChannelImpl

public EventChannelImpl(ClientIdentification clientIdentification,
                        ScheduledExecutorService executor,
                        AxonServerManagedChannel channel)

Constructs a EventChannelImpl.

Parameters:

executor - a ScheduledExecutorService used to schedule reconnects of this channel

channel - the AxonServerManagedChannel used to form the connection with AxonServer

Method Detail

connect

public void connect()

Description copied from class: AbstractAxonServerChannel

Connect this channel with AxonServer.

Specified by:

connect in class AbstractAxonServerChannel<Void>

reconnect

public void reconnect()

Description copied from class: AbstractAxonServerChannel

Reconnect this channel with AxonServer.

Specified by:

reconnect in class AbstractAxonServerChannel<Void>

disconnect

public void disconnect()

Description copied from class: AbstractAxonServerChannel

Disconnect this channel from AxonServer.

Specified by:

disconnect in class AbstractAxonServerChannel<Void>

isReady

public boolean isReady()

Description copied from class: AbstractAxonServerChannel

Validate whether this channel has all required streams connected with AxonServer. If the state of the channel does not require any active streams, it is considered ready and will return true.

Specified by:

isReady in class AbstractAxonServerChannel<Void>

Returns:

true if this channel is connected with AxonServer or does not require any active connections, false otherwise

startAppendEventsTransaction

public AppendEventsTransaction startAppendEventsTransaction()

Description copied from interface: EventChannel

Starts a new transaction to append events.

Specified by:

startAppendEventsTransaction in interface EventChannel

Returns:

The transaction reference onto which to register events to append

scheduleEvent

public CompletableFuture<String> scheduleEvent(Instant scheduleTime,
                                               Event event)

Description copied from interface: EventChannel

Schedule the given event to be published at given scheduleTime. The returned value can be used to cancel the schedule, or to reschedule the event to another time.

Specified by:

scheduleEvent in interface EventChannel

Parameters:

scheduleTime - The scheduleTime at which to publish the event

event - The event to publish

Returns:

a token used to cancel the schedule

cancelSchedule

public CompletableFuture<InstructionAck> cancelSchedule(String token)

Description copied from interface: EventChannel

Cancels the scheduled publication of an event for which the given scheduleToken was returned.

Specified by:

cancelSchedule in interface EventChannel

Parameters:

token - the token provided when scheduling the event to be cancelled

Returns:

a future reference to the result of the instruction

reschedule

public CompletableFuture<String> reschedule(String scheduleToken,
                                            Instant scheduleTime,
                                            Event event)

Description copied from interface: EventChannel

Convenience method to cancel the scheduled event with given scheduleToken and reschedule the given event to be published at given scheduleTime. Is effectively the same as cancelling and scheduling in separate calls, but this call requires only a single round-trip to the server.

Specified by:

reschedule in interface EventChannel

Parameters:

scheduleToken - the token of the event to cancel

scheduleTime - the point in time to publish the new event

event - the event to publish

Returns:

a future reference to the token for the new schedule

findHighestSequence

public CompletableFuture<Long> findHighestSequence(String aggregateId)

Description copied from interface: EventChannel

Find the highest sequence number (i.e. the sequence number of the most recent event) for an aggregate with given aggregateId.

If no events for an aggregate with given identifier have been found, the returned CompletableFuture will resolve to a null value.

Specified by:

findHighestSequence in interface EventChannel

Parameters:

aggregateId - the identifier of the aggregate to find the sequence number for

Returns:

a CompletableFuture providing the sequence number found, or resolving to null when no event was found.

openStream

public EventStream openStream(long token,
                              int bufferSize,
                              int refillBatch,
                              boolean forceReadFromLeader)

Description copied from interface: EventChannel

Open an EventStream, for sequentially consuming events from AxonServer, starting at given token and keeping a local buffer of bufferSize, which is refilled after consuming refillBatch items. The forceReadFromLeader parameter can be used to enforce this stream to read events from the RAFT leader.

A value for bufferSize smaller than 64 items will result in a buffer of 64. A value for refillBatch smaller than 16 will result in a refill batch of 16. A value larger than the bufferSize will be reduced to match the given bufferSize. While this will work, it is not recommended. The refillBatch should be sufficiently small to allow for a constant flow of messages to consume.

The stream of Events starts immediately upon the invocation of this method, making the first messages available for consumption as soon as they have arrived from AxonServer.

Specified by:

openStream in interface EventChannel

Parameters:

token - the token representing the position to start the stream, or -1 to start from beginning

bufferSize - the number of events to buffer locally

refillBatch - the number of events to be consumed prior to refilling the buffer

forceReadFromLeader - a boolean defining whether Events must be read from the leader

Returns:

the Stream for consuming the requested Events

See Also:

to use a sensible default for refill batch value.

createPersistentStream

public CompletableFuture<CreateResult> createPersistentStream(String streamId,
                                                              PersistentStreamProperties creationProperties)

Description copied from interface: EventChannel

Opens the persistent stream identified by streamId. If the stream does not exist it will be created with the properties specified in creationProperties.

Specified by:

createPersistentStream in interface EventChannel

Parameters:

streamId - the unique identification of a persistent stream

creationProperties - properties to initialize the persistent stream if it does not exist yet

Returns:

a result indicating if the stream has been created or it already existed

openPersistentStream

public PersistentStream openPersistentStream(String streamId,
                                             int bufferSize,
                                             int refillBatch,
                                             PersistentStreamCallbacks callbacks)

Description copied from interface: EventChannel

Opens the persistent stream identified by streamId. Fails if the stream does not exist.

Specified by:

openPersistentStream in interface EventChannel

Parameters:

streamId - the unique identification of a persistent stream

bufferSize - the number of events to buffer locally

refillBatch - the number of events to be consumed prior to refilling the buffer

callbacks - the callbacks that are invoked on persistent stream events

Returns:

a PersistedStream streaming events per segment

openPersistentStream

public PersistentStream openPersistentStream(String streamId,
                                             int bufferSize,
                                             int refillBatch,
                                             PersistentStreamCallbacks callbacks,
                                             PersistentStreamProperties creationProperties)

Description copied from interface: EventChannel

Opens the persistent stream identified by streamId. If the stream does not exist it will be created with the properties specified in creationProperties.

Specified by:

openPersistentStream in interface EventChannel

Parameters:

streamId - the unique identification of a persistent stream

bufferSize - the number of events to buffer locally

refillBatch - the number of events to be consumed prior to refilling the buffer

callbacks - the callbacks that are invoked on persistent stream events

creationProperties - properties to initialize the persistent stream if it does not exist yet

Returns:

a PersistentStream streaming events per segment

deletePersistentStream

public CompletableFuture<Void> deletePersistentStream(String streamId)

Description copied from interface: EventChannel

Deletes a persistent stream. If the stream does not exist the operation completes successfully.

Specified by:

deletePersistentStream in interface EventChannel

Parameters:

streamId - the unique identification of a persistent stream

Returns:

a CompletableFuture that completes when the persistent stream is deleted

updatePersistentStreamName

public CompletableFuture<Void> updatePersistentStreamName(String streamId,
                                                          String streamName)

Description copied from interface: EventChannel

Updates the name for a persistent stream.

Specified by:

updatePersistentStreamName in interface EventChannel

Parameters:

streamId - the unique identification of a persistent stream

streamName - the new logical name for the stream

Returns:

a CompletableFuture that completes when the persistent stream is updated

setPersistentStreamSegments

public CompletableFuture<Void> setPersistentStreamSegments(String streamId,
                                                           int segments)

Description copied from interface: EventChannel

Updates the number of segments for a persistent stream.

Specified by:

setPersistentStreamSegments in interface EventChannel

Parameters:

streamId - the unique identification of a persistent stream

segments - the requested number of segments for the persistent stream

Returns:

a CompletableFuture that completes when the persistent stream is updated

persistentStreams

public CompletableFuture<List<StreamStatus>> persistentStreams()

Description copied from interface: EventChannel

Returns a list of persistent streams with the last confirmed token per segment.

Specified by:

persistentStreams in interface EventChannel

Returns:

a list of persistent streams

resetPersistentStreamAtHead

public CompletableFuture<Void> resetPersistentStreamAtHead(String streamId)

Description copied from interface: EventChannel

Resets the position of a persistent stream to the head (most recent) event in the event store.

Specified by:

resetPersistentStreamAtHead in interface EventChannel

Parameters:

streamId - the unique identification of a persistent stream

resetPersistentStreamAtTail

public CompletableFuture<Void> resetPersistentStreamAtTail(String streamId)

Description copied from interface: EventChannel

Resets the position of a persistent stream to the first (oldest) event in the event store.

Specified by:

resetPersistentStreamAtTail in interface EventChannel

Parameters:

streamId - the unique identification of a persistent stream

resetPersistentStreamAtPosition

public CompletableFuture<Void> resetPersistentStreamAtPosition(String streamId,
                                                               long position)

Description copied from interface: EventChannel

Resets the position of a persistent stream to the event at given position (global index) in the event store.

Specified by:

resetPersistentStreamAtPosition in interface EventChannel

Parameters:

streamId - the unique identification of a persistent stream

position - the global index of the first event to be returned

resetPersistentStreamAtTime

public CompletableFuture<Void> resetPersistentStreamAtTime(String streamId,
                                                           long instant)

Description copied from interface: EventChannel

Resets the position of a persistent stream to the event at or after the given timestamp in the event store.

Specified by:

resetPersistentStreamAtTime in interface EventChannel

Parameters:

streamId - the unique identification of a persistent stream

instant - the timestamp of the first event to be included in the stream

openAggregateStream

public AggregateEventStream openAggregateStream(String aggregateIdentifier,
                                                boolean allowSnapshots)

Description copied from interface: EventChannel

Opens a stream for consuming Events from a single aggregate, with given allowSnapshots indicating whether the first Event may be a Snapshot Event. When given allowSnapshots is false, this method will return events starting at the first available sequence number of the aggregate (typically 0).

Note that this method does not have any form of flow control. All messages are buffered locally. When expecting large streams of events, consider using EventChannel.openAggregateStream(String, long, long) to retrieve chunks of the event stream instead.

Specified by:

openAggregateStream in interface EventChannel

Parameters:

aggregateIdentifier - the identifier of the Aggregate to load events for

allowSnapshots - a boolean whether to allow a snapshot event as first event, or not

Returns:

a stream of Events for the given Aggregate

openAggregateStream

public AggregateEventStream openAggregateStream(String aggregateIdentifier,
                                                long initialSequence,
                                                long maxSequence)

Description copied from interface: EventChannel

Opens a stream for consuming Events from a single aggregate, starting with the given initialSequence until the given maxSequence. This method will not return a Snapshot Event as first event.

Note: a maxSequence of 0 will result in all events after the initial sequence to be returned.

Specified by:

openAggregateStream in interface EventChannel

Parameters:

aggregateIdentifier - the identifier of the Aggregate to load events for

initialSequence - the sequence number of the first event to return

maxSequence - the sequence number of the last event to return

Returns:

a stream of Events for the given Aggregate

appendSnapshot

public CompletableFuture<Confirmation> appendSnapshot(Event snapshotEvent)

Description copied from interface: EventChannel

Store the given snapshotEvent.

Specified by:

appendSnapshot in interface EventChannel

Parameters:

snapshotEvent - the Snapshot Event to store

Returns:

a CompletableFuture providing the confirmation upon storage of the snapshot

loadSnapshots

public AggregateEventStream loadSnapshots(String aggregateIdentifier,
                                          long initialSequence,
                                          long maxSequence,
                                          int maxResults)

Description copied from interface: EventChannel

Loads Snapshot Events for the given aggregateIdentifier with sequence number between initialSequence and maxSequence (inclusive), returning at most maxResults number of snapshots.

Note that Snapshot Events are returned in reverse order of their sequence number.

Specified by:

loadSnapshots in interface EventChannel

Parameters:

aggregateIdentifier - the identifier of the Aggregate to retrieve snapshots for

initialSequence - the lowest sequence number of snapshots to return

maxSequence - the highest sequence number of snapshots to return

maxResults - the maximum allowed number of snapshots to return

Returns:

a stream containing Snapshot Events within the requested bounds

getLastToken

public CompletableFuture<Long> getLastToken()

Description copied from interface: EventChannel

Retrieves the Token referring to the most recent Event in the Event Store. Using this token to open a stream will only yield events that were appended after execution of this call.

Specified by:

getLastToken in interface EventChannel

Returns:

a completable future resolving the Token of the last (i.e. most recent) event

getFirstToken

public CompletableFuture<Long> getFirstToken()

Description copied from interface: EventChannel

Retrieves the Token referring to the first Event in the Event Store. Using this token to open a stream will yield all events that available in the event store.

Specified by:

getFirstToken in interface EventChannel

Returns:

a completable future resolving the Token of the first (i.e. oldest) event

getTokenAt

public CompletableFuture<Long> getTokenAt(long instant)

Description copied from interface: EventChannel

Retrieves the token referencing the first Event in the Event Store closest to the given instant.

The timestamp of the referenced Event may be on or after the given instant when the instant occurred within the Event Store's lifecycle. Will return the most recent token when the instant is placed into the future. Will return the earliest token if the instant is placed in the past (read: before the existence of this Event Store).

Using this token to open a stream will yield all events with a timestamp starting at that instance but may also yield events created before this instant, whose append transaction completed after the instant.

Specified by:

getTokenAt in interface EventChannel

Parameters:

instant - The point in time for which to retrieve a token referencing an event within the Event Store. If the given instant resides outside the Event Store its existing time frame, the latest or earliest token is returned for a future or past instant respectively.

Returns:

A completable future resolving the Token of the first (i.e. oldest) event with timestamp on or after given instant.

queryEvents

public ResultStream<EventQueryResultEntry> queryEvents(String queryExpression,
                                                       boolean liveStream,
                                                       String contextName)

Description copied from interface: EventChannel

Queries the Event Store for events using given queryExpression. The given liveStream indicates whether the query should complete when the end of the Event Stream is reached, or if the query should continue processing events as they are stored. When a contextName is provided the query is executed for that context, instead of the main context for the channel.

Specified by:

queryEvents in interface EventChannel

Parameters:

queryExpression - a valid Event Stream Query Language expression

liveStream - whether to continue processing live events

contextName - the name of the context of the event store to query

Returns:

a ResultStream containing the query results

querySnapshotEvents

public ResultStream<EventQueryResultEntry> querySnapshotEvents(String queryExpression,
                                                               boolean liveStream,
                                                               String contextName)

Description copied from interface: EventChannel

Queries the Event Store for snapshot events using given queryExpression. The given liveStream indicates whether the query should complete when the end of the Snapshot Event Stream is reached, or if the query should continue processing snapshot events as they are stored. When a contextName is provided the query is executed for that context, instead of the main context for the channel.

Specified by:

querySnapshotEvents in interface EventChannel

Parameters:

queryExpression - a valid Event Stream Query Language expression

liveStream - whether to continue processing live snapshot events

contextName - the name of the context of the event store to query

Returns:

a ResultStream containing the query results