org.axonframework.eventsourcing.eventstore

Class BatchingEventStorageEngine

java.lang.Object

org.axonframework.eventsourcing.eventstore.AbstractEventStorageEngine

org.axonframework.eventsourcing.eventstore.BatchingEventStorageEngine

All Implemented Interfaces:

EventStorageEngine

Direct Known Subclasses:

JdbcEventStorageEngine, JpaEventStorageEngine, MongoEventStorageEngine

public abstract class BatchingEventStorageEngine
extends AbstractEventStorageEngine

Abstract EventStorageEngine implementation that fetches events in batches from the backing database.

Author:

Rene de Waele

Constructor Summary

Constructors

Constructor and Description
BatchingEventStorageEngine(Serializer snapshotSerializer, EventUpcaster upcasterChain, PersistenceExceptionResolver persistenceExceptionResolver, Serializer eventSerializer, Integer batchSize) Initializes an EventStorageEngine with given serializer, upcasterChain, persistenceExceptionResolver, eventSerializer and batchSize.
BatchingEventStorageEngine(Serializer snapshotSerializer, EventUpcaster upcasterChain, PersistenceExceptionResolver persistenceExceptionResolver, Serializer eventSerializer, Predicate<? super DomainEventData<?>> snapshotFilter, Integer batchSize) Initializes an EventStorageEngine with given serializer, upcasterChain, persistenceExceptionResolver, eventSerializer and batchSize.

Method Summary

Modifier and Type	Method and Description
int	batchSize() Returns the maximum number of event entries to be fetched per batch.
protected abstract List<? extends DomainEventData<?>>	fetchDomainEvents(String aggregateIdentifier, long firstSequenceNumber, int batchSize) Returns a batch of events published by an aggregate with given aggregateIdentifier.
protected abstract List<? extends TrackedEventData<?>>	fetchTrackedEvents(TrackingToken lastToken, int batchSize) Returns a batch of serialized event data entries in the event storage that have a TrackingToken greater than the given lastToken.
protected Stream<? extends DomainEventData<?>>	readEventData(String identifier, long firstSequenceNumber) Returns a Stream of serialized event data entries for an aggregate with given identifier.
protected Stream<? extends TrackedEventData<?>>	readEventData(TrackingToken trackingToken, boolean mayBlock) Returns a global Stream containing all serialized event data entries in the event storage that have a TrackingToken greater than the given trackingToken.

Methods inherited from class org.axonframework.eventsourcing.eventstore.AbstractEventStorageEngine

appendEvents, appendEvents, getEventSerializer, getSerializer, handlePersistenceException, readEvents, readEvents, readSnapshot, readSnapshotData, storeSnapshot, storeSnapshot

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.axonframework.eventsourcing.eventstore.EventStorageEngine

appendEvents, createHeadToken, createTailToken, createTokenAt, lastSequenceNumberFor, readEvents

Constructor Detail

BatchingEventStorageEngine

public BatchingEventStorageEngine(Serializer snapshotSerializer,
                                  EventUpcaster upcasterChain,
                                  PersistenceExceptionResolver persistenceExceptionResolver,
                                  Serializer eventSerializer,
                                  Integer batchSize)

Initializes an EventStorageEngine with given serializer, upcasterChain, persistenceExceptionResolver, eventSerializer and batchSize.

Parameters:

snapshotSerializer - Used to serialize and deserialize snapshots. If null a XStreamSerializer is instantiated by the AbstractEventStorageEngine.

upcasterChain - Allows older revisions of serialized objects to be deserialized. If null a NoOpEventUpcaster is used.

persistenceExceptionResolver - Detects concurrency exceptions from the backing database. If null persistence exceptions are not explicitly resolved.

eventSerializer - Used to serialize and deserialize event payload and metadata. If null a XStreamSerializer is instantiated by the AbstractEventStorageEngine.

batchSize - The number of events that should be read at each database access. When more than this number of events must be read to rebuild an aggregate's state, the events are read in batches of this size. If null a batch size of 100 is used. Tip: if you use a snapshotter, make sure to choose snapshot trigger and batch size such that a single batch will generally retrieve all events required to rebuild an aggregate's state.

BatchingEventStorageEngine

public BatchingEventStorageEngine(Serializer snapshotSerializer,
                                  EventUpcaster upcasterChain,
                                  PersistenceExceptionResolver persistenceExceptionResolver,
                                  Serializer eventSerializer,
                                  Predicate<? super DomainEventData<?>> snapshotFilter,
                                  Integer batchSize)

Initializes an EventStorageEngine with given serializer, upcasterChain, persistenceExceptionResolver, eventSerializer and batchSize.

Parameters:

snapshotSerializer - Used to serialize and deserialize snapshots. If null a XStreamSerializer is instantiated by the AbstractEventStorageEngine.

upcasterChain - Allows older revisions of serialized objects to be deserialized. If null a NoOpEventUpcaster is used.

persistenceExceptionResolver - Detects concurrency exceptions from the backing database. If null persistence exceptions are not explicitly resolved.

eventSerializer - Used to serialize and deserialize event payload and metadata. If null a XStreamSerializer is instantiated by the AbstractEventStorageEngine.

snapshotFilter - Filter describing which snapshots should be considered. If null, all snapshots will be considered viable.

batchSize - The number of events that should be read at each database access. When more than this number of events must be read to rebuild an aggregate's state, the events are read in batches of this size. If null a batch size of 100 is used. Tip: if you use a snapshotter, make sure to choose snapshot trigger and batch size such that a single batch will generally retrieve all events required to rebuild an aggregate's state.

Method Detail

fetchTrackedEvents

protected abstract List<? extends TrackedEventData<?>> fetchTrackedEvents(TrackingToken lastToken,
                                                                          int batchSize)

Returns a batch of serialized event data entries in the event storage that have a TrackingToken greater than the given lastToken. Event entries in the stream should be ordered by tracking token. If the lastToken is null a stream containing all events should be returned.

Only if the returned List is empty the event storage assumes that the backing database holds no further applicable entries.

Parameters:

lastToken - Object describing the global index of the last processed event or null to create a stream of all events in the store

batchSize - The maximum number of events that should be returned

Returns:

A batch of tracked event messages stored since the given tracking token

fetchDomainEvents

protected abstract List<? extends DomainEventData<?>> fetchDomainEvents(String aggregateIdentifier,
                                                                        long firstSequenceNumber,
                                                                        int batchSize)

Returns a batch of events published by an aggregate with given aggregateIdentifier.

The sequence numbers in the returned batch should be ordered by sequence number. The first event in the batch should have a sequence number equal to or larger than given firstSequenceNumber. Implementations should make sure the returned batch does not contain gaps between events due to uncommitted storage transactions.

If the returned number of entries is smaller than the given batchSize it is assumed that the storage holds no further applicable entries.

Parameters:

aggregateIdentifier - The identifier of the aggregate to open a stream for

firstSequenceNumber - The sequence number of the first excepted event entry

batchSize - The maximum number of events that should be returned

Returns:

a batch of serialized event entries for the given aggregate

readEventData

protected Stream<? extends DomainEventData<?>> readEventData(String identifier,
                                                             long firstSequenceNumber)

Description copied from class: AbstractEventStorageEngine

Returns a Stream of serialized event data entries for an aggregate with given identifier. The events should be ordered by aggregate sequence number and have a sequence number starting from the given firstSequenceNumber.

Specified by:

readEventData in class AbstractEventStorageEngine

Parameters:

identifier - The identifier of the aggregate to open a stream for

firstSequenceNumber - The sequence number of the first excepted event entry

Returns:

a Stream of serialized event entries for the given aggregate

readEventData

protected Stream<? extends TrackedEventData<?>> readEventData(TrackingToken trackingToken,
                                                              boolean mayBlock)

Returns a global Stream containing all serialized event data entries in the event storage that have a TrackingToken greater than the given trackingToken. Event entries in the stream should be ordered by tracking token. If the trackingToken is null a stream containing all events should be returned.

If the end of the stream is reached and mayBlock is true the stream may block to wait for new events.

This implementation produces non-blocking event streams.

Specified by:

readEventData in class AbstractEventStorageEngine

Parameters:

trackingToken - Object describing the global index of the last processed event or null to create a stream of all events in the store

mayBlock - If true the storage engine may optionally choose to block to wait for new event messages if the end of the stream is reached.

Returns:

A stream containing all tracked event messages stored since the given tracking token

batchSize

public int batchSize()

Returns the maximum number of event entries to be fetched per batch.

Returns:

the fetch batch size