public class SequenceEventStorageEngine extends Object implements EventStorageEngine
New events and snapshots are stored in the active storage.
When fetching snapshots, if a snapshot cannot be found in the active storage it will be obtained from the historic storage.
No mechanism is provided to move events from the active storage to the historic storage engine so clients need to take care of this themselves.
Constructor and Description |
---|
SequenceEventStorageEngine(EventStorageEngine historicStorage,
EventStorageEngine activeStorage)
|
Modifier and Type | Method and Description |
---|---|
void |
appendEvents(List<? extends EventMessage<?>> events)
Append a list of events to the event storage.
|
TrackingToken |
createHeadToken()
Creates a token that is at the head of an event stream - that tracks all new events.
|
TrackingToken |
createTailToken()
Creates a token that is at the tail of an event stream - that tracks events from the beginning of time.
|
TrackingToken |
createTokenAt(Instant dateTime)
Creates a token that tracks all events after given
dateTime . |
Optional<Long> |
lastSequenceNumberFor(String aggregateIdentifier)
Returns the last known sequence number for the given
aggregateIdentifier . |
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 . |
void |
storeSnapshot(DomainEventMessage<?> snapshot)
Store an event that contains a snapshot of an aggregate.
|
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait
appendEvents, readEvents
public SequenceEventStorageEngine(EventStorageEngine historicStorage, EventStorageEngine activeStorage)
historicStorage
- the event storage engine that contains historic events. This can be backed by a read-only
databaseactiveStorage
- the event storage engine that contains 'new' events and to which new events and snapshots
will be writtenpublic void appendEvents(@Nonnull List<? extends EventMessage<?>> events)
EventStorageEngine
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.
appendEvents
in interface EventStorageEngine
events
- Events to append to the event storagepublic void storeSnapshot(@Nonnull DomainEventMessage<?> snapshot)
EventStorageEngine
storeSnapshot
in interface EventStorageEngine
snapshot
- The snapshot event of the aggregate that is to be storedpublic Stream<? extends TrackedEventMessage<?>> readEvents(TrackingToken trackingToken, boolean mayBlock)
EventStorageEngine
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.
readEvents
in interface EventStorageEngine
trackingToken
- Object describing the global index of the last processed event or null
to create a
stream of all events in the storemayBlock
- If true
the storage engine may optionally choose to block to wait for new event
messages if the end of the stream is reached.public DomainEventStream readEvents(@Nonnull String aggregateIdentifier, long firstSequenceNumber)
EventStorageEngine
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.
readEvents
in interface EventStorageEngine
aggregateIdentifier
- The identifier of the aggregatefirstSequenceNumber
- The expected sequence number of the first event in the returned streampublic Optional<DomainEventMessage<?>> readSnapshot(@Nonnull String aggregateIdentifier)
EventStorageEngine
aggregateIdentifier
. If the storage engine has
no snapshot event of the aggregate, an empty Optional is returned.readSnapshot
in interface EventStorageEngine
aggregateIdentifier
- The identifier of the aggregatepublic Optional<Long> lastSequenceNumberFor(@Nonnull String aggregateIdentifier)
EventStorageEngine
aggregateIdentifier
.
While it's recommended to use the sequence numbers from the DomainEventStream
, there are cases where
knowing the sequence number is required, without having read the actual events. In such case, this method is a
viable alternative.
lastSequenceNumberFor
in interface EventStorageEngine
aggregateIdentifier
- The identifier to find the last sequence number forpublic TrackingToken createTailToken()
EventStorageEngine
createTailToken
in interface EventStorageEngine
null
is returnedpublic TrackingToken createHeadToken()
EventStorageEngine
createHeadToken
in interface EventStorageEngine
null
is returnedpublic TrackingToken createTokenAt(@Nonnull Instant dateTime)
EventStorageEngine
dateTime
. If there is an event exactly at the given
dateTime
, it will be tracked too.createTokenAt
in interface EventStorageEngine
dateTime
- The date and time for determining criteria how the tracking token should be created. A tracking
token should point to very first event before this date and time.dateTime
, if there aren't events matching this criteria null
is returnedCopyright © 2010–2023. All rights reserved.