org.axonframework.eventsourcing.eventstore

Class AbstractEventStore

java.lang.Object

org.axonframework.eventhandling.AbstractEventBus

org.axonframework.eventsourcing.eventstore.AbstractEventStore

All Implemented Interfaces:

EventBus, EventStore, MessageDispatchInterceptorSupport<EventMessage<?>>, StreamableMessageSource<TrackedEventMessage<?>>, SubscribableMessageSource<EventMessage<?>>

Direct Known Subclasses:

EmbeddedEventStore

public abstract class AbstractEventStore
extends AbstractEventBus
implements EventStore

Abstract implementation of an EventStore that uses a EventStorageEngine to store and load events.

Author:

Rene de Waele

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	AbstractEventStore(EventStorageEngine storageEngine) Initializes an event store with given storageEngine and NoOpMessageMonitor.
protected	AbstractEventStore(EventStorageEngine storageEngine, MessageMonitor<? super EventMessage<?>> messageMonitor) Initializes an event store with given storageEngine and messageMonitor.

Method Summary

Modifier and Type	Method and Description
TrackingToken	createHeadToken() Creates the token at the end of an event stream.
TrackingToken	createTailToken() Creates the token at the beginning of an event stream.
TrackingToken	createTokenAt(Instant dateTime) Creates a token that tracks all events after given dateTime.
protected Optional<DomainEventMessage<?>>	handleSnapshotReadingError(String aggregateIdentifier, Throwable e) Invoked when an error (Exception or LinkageError) occurs while attempting to read a snapshot event.
Optional<Long>	lastSequenceNumberFor(String aggregateIdentifier) Returns the last known sequence number of an Event for the given aggregateIdentifier.
protected void	prepareCommit(List<? extends EventMessage<?>> events) Process given events while the Unit of Work root is preparing for commit.
DomainEventStream	readEvents(String aggregateIdentifier) Open an event stream containing all domain events belonging to the given aggregateIdentifier.
DomainEventStream	readEvents(String aggregateIdentifier, long firstSequenceNumber) Open an event stream containing all domain events belonging to the given aggregateIdentifier.
protected Stream<? extends DomainEventMessage<?>>	stagedDomainEventMessages(String aggregateIdentifier) Returns a Stream of all DomainEventMessages that have been staged for publication by an Aggregate with given aggregateIdentifier.
protected EventStorageEngine	storageEngine() Returns the EventStorageEngine used by the event store.
void	storeSnapshot(DomainEventMessage<?> snapshot) Stores the given (temporary) snapshot event.

Methods inherited from class org.axonframework.eventhandling.AbstractEventBus

afterCommit, commit, intercept, publish, queuedMessages, registerDispatchInterceptor, subscribe

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.axonframework.eventhandling.EventBus

openStream, publish, publish, registerDispatchInterceptor

Methods inherited from interface org.axonframework.messaging.SubscribableMessageSource

subscribe

Methods inherited from interface org.axonframework.messaging.StreamableMessageSource

createTokenSince

Constructor Detail

AbstractEventStore

protected AbstractEventStore(EventStorageEngine storageEngine)

Initializes an event store with given storageEngine and NoOpMessageMonitor.

Parameters:

storageEngine - The storage engine used to store and load events

AbstractEventStore

protected AbstractEventStore(EventStorageEngine storageEngine,
                             MessageMonitor<? super EventMessage<?>> messageMonitor)

Initializes an event store with given storageEngine and messageMonitor.

Parameters:

storageEngine - The storage engine used to store and load events

messageMonitor - The monitor used to record event publications

Method Detail

prepareCommit

protected void prepareCommit(List<? extends EventMessage<?>> events)

Description copied from class: AbstractEventBus

Process given events while the Unit of Work root is preparing for commit. The default implementation signals the registered MessageMonitor that the given events are ingested and passes the events to each registered event processor.

Overrides:

prepareCommit in class AbstractEventBus

Parameters:

events - Events to be published by this Event Bus

readEvents

public DomainEventStream readEvents(String aggregateIdentifier)

Open an event stream containing all domain events belonging to the given aggregateIdentifier.

The returned stream is finite, ending with the last known event of the aggregate. If the event store holds no events of the given aggregate an empty stream is returned.

This implementation returns a DomainEventStream starting with the last stored snapshot of the aggregate followed by subsequent domain events.

Specified by:

readEvents in interface EventStore

Parameters:

aggregateIdentifier - the identifier of the aggregate whose events to fetch

Returns:

a stream of all currently stored events of the aggregate

handleSnapshotReadingError

protected Optional<DomainEventMessage<?>> handleSnapshotReadingError(String aggregateIdentifier,
                                                                     Throwable e)

Invoked when an error (Exception or LinkageError) occurs while attempting to read a snapshot event. This method can be overridden to change the default behavior, which is to log the exception (warn level) and ignore the snapshot.

Overriding implementations may choose to return normally, or raise an exception. Exceptions raised from this method are propagated to the caller of the readEvents(String) or readEvents(String, long) methods.

Returning an empty Optional will force the initialization of the aggregate to happen based on the entire event stream of that aggregate.

Parameters:

aggregateIdentifier - The identifier of the aggregate for which an snapshot failed to load

e - The exception or error that occurred while loading or deserializing the snapshot

Returns:

An optional DomainEventMessage to use as the snapshot for this aggregate

Throws:

RuntimeException - any runtimeException to fail loading the

stagedDomainEventMessages

protected Stream<? extends DomainEventMessage<?>> stagedDomainEventMessages(String aggregateIdentifier)

Returns a Stream of all DomainEventMessages that have been staged for publication by an Aggregate with given aggregateIdentifier.

Parameters:

aggregateIdentifier - The identifier of the aggregate to get staged events for

Returns:

a Stream of DomainEventMessage of the identified aggregate

readEvents

public DomainEventStream readEvents(String aggregateIdentifier,
                                    long firstSequenceNumber)

Description copied from interface: EventStore

Open an event stream containing all domain events belonging to the given aggregateIdentifier.

The returned stream is finite, ending with the last known event of the aggregate. If the event store holds no events of the given aggregate an empty stream is returned.

The default implementation invokes EventStore.readEvents(String) and then filters out events with a sequence number smaller than firstSequenceNumber.

Specified by:

readEvents in interface EventStore

Parameters:

aggregateIdentifier - the identifier of the aggregate whose events to fetch

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

Returns:

a stream of all currently stored events of the aggregate

storeSnapshot

public void storeSnapshot(DomainEventMessage<?> snapshot)

Description copied from interface: EventStore

Stores the given (temporary) snapshot event. This snapshot replaces the segment of the event stream identified by the snapshot's Aggregate Identifier up to (and including) the event with the snapshot's sequence number.

These snapshots will only affect the DomainEventStream returned by the EventStore.readEvents(String) method. They do not change the events returned by EventBus.openStream(TrackingToken) or those received by using SubscribableMessageSource.subscribe(java.util.function.Consumer).

Note that snapshots are considered a temporary replacement for Events, and are used as performance optimization. Event Store implementations may choose to ignore or delete snapshots.

Specified by:

storeSnapshot in interface EventStore

Parameters:

snapshot - The snapshot to replace part of the DomainEventStream.

storageEngine

protected EventStorageEngine storageEngine()

Returns the EventStorageEngine used by the event store.

Returns:

The event storage engine used by this event store

lastSequenceNumberFor

public Optional<Long> lastSequenceNumberFor(String aggregateIdentifier)

Description copied from interface: EventStore

Returns the last known sequence number of an Event for the given aggregateIdentifier.

It is preferred to retrieve the last known sequence number from the Domain Event Stream when sourcing an Aggregate from events. However, this method provides an alternative in cases no events have been read. For example when using state storage.

Specified by:

lastSequenceNumberFor in interface EventStore

Parameters:

aggregateIdentifier - The identifier of the aggregate to find the highest sequence for

Returns:

An optional containing the highest sequence number found, or an empty optional is no events are found for this aggregate

createTailToken

public TrackingToken createTailToken()

Description copied from interface: StreamableMessageSource

Creates the token at the beginning of an event stream. The beginning of an event stream in this context means the token of very first event in the stream.

The default behavior for this method is to return null, which always represents the tail position of a stream. However, implementations are encouraged to return an instance that explicitly represents the tail of the stream.

Specified by:

createTailToken in interface StreamableMessageSource<TrackedEventMessage<?>>

Returns:

the token at the beginning of an event stream

createHeadToken

public TrackingToken createHeadToken()

Description copied from interface: StreamableMessageSource

Creates the token at the end of an event stream. The end of an event stream in this context means the token of very last event in the stream.

Specified by:

createHeadToken in interface StreamableMessageSource<TrackedEventMessage<?>>

Returns:

the token at the end of an event stream

createTokenAt

public TrackingToken createTokenAt(Instant dateTime)

Description copied from interface: StreamableMessageSource

Creates a token that tracks all events after given dateTime. If there is an event exactly at the given dateTime, it will be tracked too.

Specified by:

createTokenAt in interface StreamableMessageSource<TrackedEventMessage<?>>

Parameters:

dateTime - The date and time for determining criteria how the tracking token should be created. A tracking token should point at very first event before this date and time.

Returns:

a tracking token at the given dateTime, if there aren't events matching this criteria null is returned