Package 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, JpaEventStorageEngine

public abstract class BatchingEventStorageEngine
extends AbstractEventStorageEngine

AbstractEventStorageEngine implementation that fetches events in batches from the backing database.

Since:

3.0

Author:

Rene de Waele

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	BatchingEventStorageEngine.Builder	Abstract Builder class to instantiate a BatchingEventStorageEngine.

Field Summary

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

upcasterChain

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	BatchingEventStorageEngine(BatchingEventStorageEngine.Builder builder)	Instantiate a BatchingEventStorageEngine based on the fields contained in the BatchingEventStorageEngine.Builder.

Method Summary

Modifier and Type	Method	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 boolean	fetchForAggregateUntilEmpty()	Specifies whether the readEventData(String, long) should proceed fetching events for an aggregate until an empty batch is returned.
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, getSnapshotSerializer, 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 Details

BatchingEventStorageEngine

protected BatchingEventStorageEngine(BatchingEventStorageEngine.Builder builder)

Instantiate a BatchingEventStorageEngine based on the fields contained in the BatchingEventStorageEngine.Builder.

Will assert that the event and snapshot Serializer are not null, and will throw an AxonConfigurationException if any of them is null.

Parameters:

builder - the BatchingEventStorageEngine.Builder used to instantiate a BatchingEventStorageEngine instance

Method Details

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. Implementations for which this is not always the case should override fetchForAggregateUntilEmpty() to return true and preferably configure a better BatchingEventStorageEngine.Builder.finalAggregateBatchPredicate(Predicate) to provide a better heuristic for detecting the last batch in a stream.

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

fetchForAggregateUntilEmpty

protected boolean fetchForAggregateUntilEmpty()

Specifies whether the readEventData(String, long) should proceed fetching events for an aggregate until an empty batch is returned. Defaults to false, as Aggregate event batches typically do not have gaps in them.

Returns:

a boolean specifying whether readEventData(String, long) should proceed fetching events for an aggregate until an empty batch is returned

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