org.axonframework.mongo.eventsourcing.eventstore

Class AbstractMongoEventStorageStrategy

java.lang.Object

org.axonframework.mongo.eventsourcing.eventstore.AbstractMongoEventStorageStrategy

All Implemented Interfaces:

StorageStrategy

Direct Known Subclasses:

DocumentPerCommitStorageStrategy, DocumentPerEventStorageStrategy

public abstract class AbstractMongoEventStorageStrategy
extends Object
implements StorageStrategy

Abstract implementation of a Mongo StorageStrategy. Implementations only need to provide methods to convert events and snapshots into Documents and vice versa.

Author:

Rene de Waele

Field Summary

Fields

Modifier and Type	Field and Description
protected static int	ORDER_ASC The value to pass to Mongo to fetch documents in ascending order.
protected static int	ORDER_DESC The value to pass to Mongo to fetch documents in descending order.

Constructor Summary

Constructors

Constructor and Description
AbstractMongoEventStorageStrategy(EventEntryConfiguration eventConfiguration, Duration lookBackTime) Initializes a new StorageStrategy for a EventStorageEngine that uses Mongo.

Method Summary

Modifier and Type	Method and Description
void	appendEvents(com.mongodb.client.MongoCollection<org.bson.Document> eventCollection, List<? extends EventMessage<?>> events, Serializer serializer) Appends the given list of events to the given eventCollection.
void	appendSnapshot(com.mongodb.client.MongoCollection<org.bson.Document> snapshotCollection, DomainEventMessage<?> snapshot, Serializer serializer) Append the given aggregate snapshot to the snapshotCollection.
protected abstract Stream<org.bson.Document>	createEventDocuments(List<? extends EventMessage<?>> events, Serializer serializer) Returns a stream of Mongo documents that represent the given batch of events.
protected abstract org.bson.Document	createSnapshotDocument(DomainEventMessage<?> snapshot, Serializer serializer) Returns a Mongo document for given snapshot event.
TrackingToken	createTailToken(com.mongodb.client.MongoCollection<org.bson.Document> eventsCollection) Creates a token that is at the tail of an event stream - that tracks events from the beginning of time.
void	deleteSnapshots(com.mongodb.client.MongoCollection<org.bson.Document> snapshotCollection, String aggregateIdentifier, long sequenceNumber) Delete all snapshots of the aggregate with given aggregateIdentifier from the given snapshotCollection, whose sequence number is lower than the given sequenceNumber.
void	ensureIndexes(com.mongodb.client.MongoCollection<org.bson.Document> eventsCollection, com.mongodb.client.MongoCollection<org.bson.Document> snapshotsCollection) Ensure that the correct indexes are in place.
protected EventEntryConfiguration	eventConfiguration() Returns the EventEntryConfiguration that configures how event entries are to be stored.
protected abstract Stream<? extends DomainEventData<?>>	extractEvents(org.bson.Document object) Retrieves event data from the given Mongo object.
protected Long	extractHighestSequenceNumber(org.bson.Document document) Extract the highest sequence number known from the entry represented by the given document.
protected abstract DomainEventData<?>	extractSnapshot(org.bson.Document object) Retrieves snapshot event data from the given Mongo object.
List<? extends DomainEventData<?>>	findDomainEvents(com.mongodb.client.MongoCollection<org.bson.Document> collection, String aggregateIdentifier, long firstSequenceNumber, int batchSize) Returns a batch of events for an aggregate with given aggregateIdentifier and a sequence number equal or higher than the given firstSequenceNumber.
Optional<? extends DomainEventData<?>>	findLastSnapshot(com.mongodb.client.MongoCollection<org.bson.Document> snapshotCollection, String aggregateIdentifier) Finds the entry containing the last snapshot event for an aggregate with given aggregateIdentifier in the given collection.
List<? extends TrackedEventData<?>>	findTrackedEvents(com.mongodb.client.MongoCollection<org.bson.Document> eventCollection, TrackingToken lastToken, int batchSize) Returns a batch of tracked events with a tracking token above the given lastToken.
Optional<Long>	lastSequenceNumberFor(com.mongodb.client.MongoCollection<org.bson.Document> eventsCollection, String aggregateIdentifier) Return the last known sequence number for an Aggregate with given aggregateIdentifier, whose Events are stored in the given eventsCollection.

Methods inherited from class java.lang.Object

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

Field Detail

ORDER_ASC

protected static final int ORDER_ASC

The value to pass to Mongo to fetch documents in ascending order.

See Also:

Constant Field Values

ORDER_DESC

protected static final int ORDER_DESC

The value to pass to Mongo to fetch documents in descending order.

See Also:

Constant Field Values

Constructor Detail

AbstractMongoEventStorageStrategy

public AbstractMongoEventStorageStrategy(EventEntryConfiguration eventConfiguration,
                                         Duration lookBackTime)

Initializes a new StorageStrategy for a EventStorageEngine that uses Mongo.

Parameters:

eventConfiguration - configuration of the event entry 'schema'. If null the schema with default values is used.

lookBackTime - the maximum time to look back when fetching new events while tracking. If null a 10 second interval is used.

Method Detail

appendEvents

public void appendEvents(com.mongodb.client.MongoCollection<org.bson.Document> eventCollection,
                         List<? extends EventMessage<?>> events,
                         Serializer serializer)

Description copied from interface: StorageStrategy

Appends the given list of events to the given eventCollection. Use the given serializer to serialize the payload and metadata of the events.

Specified by:

appendEvents in interface StorageStrategy

Parameters:

eventCollection - the document collection that contains event entries

events - the event messages to append to the event collection

serializer - the serializer used to serialize the event payload and metadata

createEventDocuments

protected abstract Stream<org.bson.Document> createEventDocuments(List<? extends EventMessage<?>> events,
                                                                  Serializer serializer)

Returns a stream of Mongo documents that represent the given batch of events. The given list of events represents events produced in the context of a single unit of work. Uses the given serializer to serialize event payload and metadata.

Parameters:

events - the events to convert to Mongo documents

serializer - the serializer to convert the events' payload and metadata

Returns:

stream of Mongo documents from the given event batch

appendSnapshot

public void appendSnapshot(com.mongodb.client.MongoCollection<org.bson.Document> snapshotCollection,
                           DomainEventMessage<?> snapshot,
                           Serializer serializer)

Description copied from interface: StorageStrategy

Append the given aggregate snapshot to the snapshotCollection. Use the given serializer to serialize the payload and metadata of the snapshot.

Specified by:

appendSnapshot in interface StorageStrategy

Parameters:

snapshotCollection - the document collection that contains aggregate snapshot entries

snapshot - the snapshot event message to append to the event collection

serializer - the serializer used to serialize the payload and metadata of the snapshot event

createSnapshotDocument

protected abstract org.bson.Document createSnapshotDocument(DomainEventMessage<?> snapshot,
                                                            Serializer serializer)

Returns a Mongo document for given snapshot event. Uses the given serializer to serialize event payload and metadata.

Parameters:

snapshot - the snapshot to convert

serializer - the to convert the snapshot's payload and metadata

Returns:

a Mongo documents from given snapshot

deleteSnapshots

public void deleteSnapshots(com.mongodb.client.MongoCollection<org.bson.Document> snapshotCollection,
                            String aggregateIdentifier,
                            long sequenceNumber)

Description copied from interface: StorageStrategy

Delete all snapshots of the aggregate with given aggregateIdentifier from the given snapshotCollection, whose sequence number is lower than the given sequenceNumber.

Specified by:

deleteSnapshots in interface StorageStrategy

Parameters:

snapshotCollection - the document collection that contains aggregate snapshot entries

aggregateIdentifier - the identifier of the aggregate for which to delete all snapshots

sequenceNumber - The sequenceNumber representing the upper bound (exclusive) of the snapshots to delete

findDomainEvents

public List<? extends DomainEventData<?>> findDomainEvents(com.mongodb.client.MongoCollection<org.bson.Document> collection,
                                                           String aggregateIdentifier,
                                                           long firstSequenceNumber,
                                                           int batchSize)

Description copied from interface: StorageStrategy

Returns a batch of events for an aggregate with given aggregateIdentifier and a sequence number equal or higher than the given firstSequenceNumber. The returned documents should be ordered chronologically (typically by using the sequence number). The size of the returned list should not exceed the given batchSize.

Specified by:

findDomainEvents in interface StorageStrategy

Parameters:

collection - The collection in which to find the events

aggregateIdentifier - The identifier of the aggregate to query

firstSequenceNumber - The sequence number of the first event to return

batchSize - The maximum number of event entries to fetch

Returns:

a list of matching domain event entries

extractEvents

protected abstract Stream<? extends DomainEventData<?>> extractEvents(org.bson.Document object)

Retrieves event data from the given Mongo object.

Parameters:

object - the object to convert to event data

Returns:

stream of events from given document

findTrackedEvents

public List<? extends TrackedEventData<?>> findTrackedEvents(com.mongodb.client.MongoCollection<org.bson.Document> eventCollection,
                                                             TrackingToken lastToken,
                                                             int batchSize)

Description copied from interface: StorageStrategy

Returns a batch of tracked events with a tracking token above the given lastToken. If lastToken is null the first (oldest) batch of entries in the store should be returned.

The returned documents should be ordered chronologically (typically by using the timestamp of the event). The size of the returned list should not exceed the given batchSize.

Specified by:

findTrackedEvents in interface StorageStrategy

Parameters:

eventCollection - The collection in which to find the events

lastToken - the token of the last event in the previous batch or null to load the oldest batch

batchSize - The maximum number of event entries to fetch

Returns:

a list of matching tracked event entries

findLastSnapshot

public Optional<? extends DomainEventData<?>> findLastSnapshot(com.mongodb.client.MongoCollection<org.bson.Document> snapshotCollection,
                                                               String aggregateIdentifier)

Description copied from interface: StorageStrategy

Finds the entry containing the last snapshot event for an aggregate with given aggregateIdentifier in the given collection.

Specified by:

findLastSnapshot in interface StorageStrategy

Parameters:

snapshotCollection - The collection to find the last snapshot event in

aggregateIdentifier - The identifier of the aggregate to find a snapshot for

Returns:

an optional with DomainEventData of the snapshot if found

lastSequenceNumberFor

public Optional<Long> lastSequenceNumberFor(com.mongodb.client.MongoCollection<org.bson.Document> eventsCollection,
                                            String aggregateIdentifier)

Description copied from interface: StorageStrategy

Return the last known sequence number for an Aggregate with given aggregateIdentifier, whose Events are stored in the given eventsCollection.

Specified by:

lastSequenceNumberFor in interface StorageStrategy

Parameters:

eventsCollection - The Collection in which to search for Events

aggregateIdentifier - The aggregate to find the last sequence number for

Returns:

An optional with the last sequence number, or an empty optional if no Events are present for the given aggregate.

createTailToken

public TrackingToken createTailToken(com.mongodb.client.MongoCollection<org.bson.Document> eventsCollection)

Description copied from interface: StorageStrategy

Creates a token that is at the tail of an event stream - that tracks events from the beginning of time.

Specified by:

createTailToken in interface StorageStrategy

Returns:

a tracking token at the tail of an event stream, if event stream is empty null is returned

extractSnapshot

protected abstract DomainEventData<?> extractSnapshot(org.bson.Document object)

Retrieves snapshot event data from the given Mongo object.

Parameters:

object - the object to convert to snapshot data

Returns:

snapshot data contained in given document

extractHighestSequenceNumber

protected Long extractHighestSequenceNumber(org.bson.Document document)

Extract the highest sequence number known from the entry represented by the given document.

This implementation takes the sequenceNumberProperty defined in the eventConfiguration. Implementations that allow storage of multiple events in a single document should override this method.

Parameters:

document - The document representing the entry stored in Mongo

Returns:

a Long representing the highest sequence number found

ensureIndexes

public void ensureIndexes(com.mongodb.client.MongoCollection<org.bson.Document> eventsCollection,
                          com.mongodb.client.MongoCollection<org.bson.Document> snapshotsCollection)

Description copied from interface: StorageStrategy

Ensure that the correct indexes are in place.

Specified by:

ensureIndexes in interface StorageStrategy

Parameters:

eventsCollection - The collection containing the documents representing commits and events.

snapshotsCollection - The collection containing the document representing snapshots

eventConfiguration

protected EventEntryConfiguration eventConfiguration()

Returns the EventEntryConfiguration that configures how event entries are to be stored.

Returns:

the event entry configuration