org.axonframework.eventsourcing.eventstore

Class AbstractEventStorageEngine

java.lang.Object

org.axonframework.eventsourcing.eventstore.AbstractEventStorageEngine

All Implemented Interfaces:

EventStorageEngine

Direct Known Subclasses:

BatchingEventStorageEngine

public abstract class AbstractEventStorageEngine
extends Object
implements EventStorageEngine

Abstract event storage implementation that takes care of event serialization and upcasting.

Author:

Rene de Waele

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	AbstractEventStorageEngine(Serializer serializer, EventUpcaster upcasterChain, PersistenceExceptionResolver persistenceExceptionResolver) Deprecated. The constructor which supplies an additional Serializer for events should be used.
protected	AbstractEventStorageEngine(Serializer snapshotSerializer, EventUpcaster upcasterChain, PersistenceExceptionResolver persistenceExceptionResolver, Serializer eventSerializer) Initializes an EventStorageEngine with given serializer, upcasterChain and persistenceExceptionResolver.
protected	AbstractEventStorageEngine(Serializer snapshotSerializer, EventUpcaster upcasterChain, PersistenceExceptionResolver persistenceExceptionResolver, Serializer eventSerializer, Predicate<? super DomainEventData<?>> snapshotFilter) Initializes an EventStorageEngine with given serializer, upcasterChain and persistenceExceptionResolver.

Method Summary

Modifier and Type	Method and Description
void	appendEvents(List<? extends EventMessage<?>> events) Append a list of events to the event storage.
protected abstract void	appendEvents(List<? extends EventMessage<?>> events, Serializer serializer) Append given events to the backing database.
Serializer	getEventSerializer() Get the serializer used by this storage engine when storing and retrieving events.
Serializer	getSerializer() Get the serializer used by this storage engine when storing and retrieving snapshots.
protected void	handlePersistenceException(Exception exception, EventMessage<?> failedEvent) Invoke when an Exception is raised while persisting an Event or Snapshot.
protected abstract Stream<? extends DomainEventData<?>>	readEventData(String identifier, long firstSequenceNumber) Returns a Stream of serialized event data entries for an aggregate with given identifier.
protected abstract 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.
DomainEventStream	readEvents(String aggregateIdentifier, long firstSequenceNumber) Get a DomainEventStream containing all events published by the aggregate with given aggregateIdentifier starting with the first event having a sequence number that is equal or larger than the given firstSequenceNumber.
Stream<? extends TrackedEventMessage<?>>	readEvents(TrackingToken trackingToken, boolean mayBlock) Open an event stream containing all events stored since given tracking token.
Optional<DomainEventMessage<?>>	readSnapshot(String aggregateIdentifier) Try to load a snapshot event of the aggregate with given aggregateIdentifier.
protected abstract Optional<? extends DomainEventData<?>>	readSnapshotData(String aggregateIdentifier) Returns an optional serialized event entry for given aggregateIdentifier if the backing database contains a snapshot of the aggregate.
void	storeSnapshot(DomainEventMessage<?> snapshot) Store an event that contains a snapshot of an aggregate.
protected abstract void	storeSnapshot(DomainEventMessage<?> snapshot, Serializer serializer) Store the given snapshot of an Aggregate.

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

AbstractEventStorageEngine

@Deprecated
protected AbstractEventStorageEngine(Serializer serializer,
                                                  EventUpcaster upcasterChain,
                                                  PersistenceExceptionResolver persistenceExceptionResolver)

Deprecated. The constructor which supplies an additional Serializer for events should be used.

Initializes an EventStorageEngine with given serializer, upcasterChain and persistenceExceptionResolver.

Parameters:

serializer - Used to serialize and deserialize event payload and metadata. If null a new XStreamSerializer is used.

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.

AbstractEventStorageEngine

protected AbstractEventStorageEngine(Serializer snapshotSerializer,
                                     EventUpcaster upcasterChain,
                                     PersistenceExceptionResolver persistenceExceptionResolver,
                                     Serializer eventSerializer)

Initializes an EventStorageEngine with given serializer, upcasterChain and persistenceExceptionResolver.

Parameters:

snapshotSerializer - Used to serialize and deserialize snapshots. If null a new XStreamSerializer is used.

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 new XStreamSerializer is used.

AbstractEventStorageEngine

protected AbstractEventStorageEngine(Serializer snapshotSerializer,
                                     EventUpcaster upcasterChain,
                                     PersistenceExceptionResolver persistenceExceptionResolver,
                                     Serializer eventSerializer,
                                     Predicate<? super DomainEventData<?>> snapshotFilter)

Initializes an EventStorageEngine with given serializer, upcasterChain and persistenceExceptionResolver. The given snapshotFilter is used to define which snapshots are accepted for this version of the application.

Parameters:

snapshotSerializer - Used to serialize and deserialize snapshots. If null a new XStreamSerializer is used.

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 new XStreamSerializer is used.

snapshotFilter - Decides whether to use a snapshot or not. If null, every snapshot is accepted as viable.

Method Detail

readEvents

public Stream<? extends TrackedEventMessage<?>> readEvents(TrackingToken trackingToken,
                                                           boolean mayBlock)

Description copied from interface: EventStorageEngine

Open an event stream containing all events stored since given tracking token. The returned stream is comprised of events from aggregates as well as other application events. Pass a trackingToken of null to open a stream containing all available events.

If the value of the given mayBlock is true the returned stream is allowed to block while waiting for new event messages if the end of the stream is reached.

Specified by:

readEvents in interface EventStorageEngine

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

readEvents

public DomainEventStream readEvents(String aggregateIdentifier,
                                    long firstSequenceNumber)

Description copied from interface: EventStorageEngine

Get a DomainEventStream containing all events published by the aggregate with given aggregateIdentifier starting with the first event having a sequence number that is equal or larger than the given firstSequenceNumber.

The returned stream is finite, i.e. it should not block to wait for further events if the end of the event stream of the aggregate is reached.

Specified by:

readEvents in interface EventStorageEngine

Parameters:

aggregateIdentifier - The identifier of the aggregate

firstSequenceNumber - The expected sequence number of the first event in the returned stream

Returns:

A non-blocking DomainEventStream of the given aggregate

readSnapshot

public Optional<DomainEventMessage<?>> readSnapshot(String aggregateIdentifier)

Description copied from interface: EventStorageEngine

Try to load a snapshot event of the aggregate with given aggregateIdentifier. If the storage engine has no snapshot event of the aggregate, an empty Optional is returned.

Specified by:

readSnapshot in interface EventStorageEngine

Parameters:

aggregateIdentifier - The identifier of the aggregate

Returns:

An optional with a snapshot of the aggregate

appendEvents

public void appendEvents(List<? extends EventMessage<?>> events)

Description copied from interface: EventStorageEngine

Append a list of events to the event storage. Events will be appended in the order that they are offered in.

Note that all events should have a unique event identifier. When storing domain events events should also have a unique combination of aggregate id and sequence number.

Specified by:

appendEvents in interface EventStorageEngine

Parameters:

events - Events to append to the event storage

storeSnapshot

public void storeSnapshot(DomainEventMessage<?> snapshot)

Description copied from interface: EventStorageEngine

Store an event that contains a snapshot of an aggregate. If the event storage already contains a snapshot for the same aggregate, then it will be replaced with the given snapshot.

Specified by:

storeSnapshot in interface EventStorageEngine

Parameters:

snapshot - The snapshot event of the aggregate that is to be stored

handlePersistenceException

protected void handlePersistenceException(Exception exception,
                                          EventMessage<?> failedEvent)

Invoke when an Exception is raised while persisting an Event or Snapshot.

Parameters:

exception - The exception raised while persisting an Event

failedEvent - The EventMessage that could not be persisted

appendEvents

protected abstract void appendEvents(List<? extends EventMessage<?>> events,
                                     Serializer serializer)

Append given events to the backing database. Use the given serializer to serialize the event's payload and metadata.

Parameters:

events - Events to append to the database

serializer - Serializer used to convert the events to a suitable format for storage

storeSnapshot

protected abstract void storeSnapshot(DomainEventMessage<?> snapshot,
                                      Serializer serializer)

Store the given snapshot of an Aggregate. Implementations may override any existing snapshot of the Aggregate with the given snapshot.

Parameters:

snapshot - Snapshot Event of the aggregate

serializer - Serializer used to convert the snapshot event to a suitable format for storage

readEventData

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

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.

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 abstract 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.

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

readSnapshotData

protected abstract Optional<? extends DomainEventData<?>> readSnapshotData(String aggregateIdentifier)

Returns an optional serialized event entry for given aggregateIdentifier if the backing database contains a snapshot of the aggregate.

Parameters:

aggregateIdentifier - The aggregate identifier to fetch a snapshot for

Returns:

An optional with a serialized snapshot of the aggregate

getSerializer

public Serializer getSerializer()

Get the serializer used by this storage engine when storing and retrieving snapshots.

Returns:

the serializer used by this storage

getEventSerializer

public Serializer getEventSerializer()

Get the serializer used by this storage engine when storing and retrieving events.

Returns:

the serializer used by this storage