public class EmbeddedEventStore extends AbstractEventStore
EventStore that stores and fetches events using an EventStorageEngine. If
supported by its storage engine the embedded event store provides event tracking and replaying capabilities.
The event store can be tracked by multiple event processors simultaneously. To prevent that each event processor
needs to read from the storage engine individually the embedded event store contains a cache of the most recent
events. This cache is shared between the streams of various event processors. So, assuming an event processor
processes events fast enough and is not far behind the head of the event log it will not need a private connection
to the underlying data store. The size of the cache (in number of events) is configurable.
This 'event consumption optimization' might in some scenarios not be desirable, as it will spin up additional threads
and perform some locking operations. Hence it is switchable by using the
EmbeddedEventStore(EventStorageEngine, MessageMonitor, int, long, long, TimeUnit, ThreadFactory, boolean)
constructor and provided a false for the last parameter. Additionally, this can also be turned off by
providing a system property with key optimize-event-consumption.
The embedded event store automatically fetches new events from the store if there is at least one registered tracking event processor present. It will do so after new events are committed to the store, as well as periodically as events may have been committed by other nodes or applications. This periodic fetch delay is configurable.
| Constructor and Description |
|---|
EmbeddedEventStore(EventStorageEngine storageEngine)
Initializes an
EmbeddedEventStore with given storageEngine and default settings. |
EmbeddedEventStore(EventStorageEngine storageEngine,
MessageMonitor<? super EventMessage<?>> monitor)
|
EmbeddedEventStore(EventStorageEngine storageEngine,
MessageMonitor<? super EventMessage<?>> monitor,
int cachedEvents,
long fetchDelay,
long cleanupDelay,
TimeUnit timeUnit)
|
EmbeddedEventStore(EventStorageEngine storageEngine,
MessageMonitor<? super EventMessage<?>> monitor,
int cachedEvents,
long fetchDelay,
long cleanupDelay,
TimeUnit timeUnit,
ThreadFactory threadFactory)
|
EmbeddedEventStore(EventStorageEngine storageEngine,
MessageMonitor<? super EventMessage<?>> monitor,
int cachedEvents,
long fetchDelay,
long cleanupDelay,
TimeUnit timeUnit,
ThreadFactory threadFactory,
boolean optimizeEventConsumption)
|
| Modifier and Type | Method and Description |
|---|---|
protected void |
afterCommit(List<? extends EventMessage<?>> events)
Process given
events after the Unit of Work has been committed. |
TrackingEventStream |
openStream(TrackingToken trackingToken)
Open an event stream containing all events since given tracking token.
|
void |
shutDown()
Method to invoke when the application shuts down.
|
createHeadToken, createTailToken, createTokenAt, handleSnapshotReadingError, lastSequenceNumberFor, prepareCommit, readEvents, readEvents, stagedDomainEventMessages, storageEngine, storeSnapshotcommit, intercept, publish, queuedMessages, registerDispatchInterceptor, subscribeclone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, waitpublish, publish, registerDispatchInterceptorsubscribecreateTokenSincepublic EmbeddedEventStore(EventStorageEngine storageEngine)
EmbeddedEventStore with given storageEngine and default settings.storageEngine - the storage engine to usepublic EmbeddedEventStore(EventStorageEngine storageEngine, MessageMonitor<? super EventMessage<?>> monitor)
storageEngine - the storage engine to usemonitor - the metrics monitor that tracks how many events are ingested by the event storepublic EmbeddedEventStore(EventStorageEngine storageEngine, MessageMonitor<? super EventMessage<?>> monitor, int cachedEvents, long fetchDelay, long cleanupDelay, TimeUnit timeUnit)
storageEngine - the storage engine to usemonitor - the metrics monitor that tracks how many events are ingested by the event storecachedEvents - the maximum number of events in the cache that is shared between the streams of tracking
event processorsfetchDelay - the time to wait before fetching new events from the backing storage engine while tracking
after a previous stream was fetched and read. Note that this only applies to situations in
which no events from the current application have meanwhile been committed. If the current
application commits events then those events are fetched without delay.cleanupDelay - the delay between two clean ups of lagging event processors. An event processor is lagging
behind and removed from the set of processors that track cached events if the oldest event
in the cache is newer than the last processed event of the event processor. Once removed the
processor will be independently fetching directly from the event storage engine until it has
caught up again. Event processors will not notice this change during tracking (i.e. the
stream is not closed when an event processor falls behind and is removed).timeUnit - time unit for fetch and clean up delaypublic EmbeddedEventStore(EventStorageEngine storageEngine, MessageMonitor<? super EventMessage<?>> monitor, int cachedEvents, long fetchDelay, long cleanupDelay, TimeUnit timeUnit, ThreadFactory threadFactory)
storageEngine - the storage engine to usemonitor - the metrics monitor that tracks how many events are ingested by the event storecachedEvents - the maximum number of events in the cache that is shared between the streams of tracking
event processorsfetchDelay - the time to wait before fetching new events from the backing storage engine while tracking
after a previous stream was fetched and read. Note that this only applies to situations in
which no events from the current application have meanwhile been committed. If the current
application commits events then those events are fetched without delay.cleanupDelay - the delay between two clean ups of lagging event processors. An event processor is lagging
behind and removed from the set of processors that track cached events if the oldest event
in the cache is newer than the last processed event of the event processor. Once removed the
processor will be independently fetching directly from the event storage engine until it has
caught up again. Event processors will not notice this change during tracking (i.e. the
stream is not closed when an event processor falls behind and is removed).timeUnit - time unit for fetch and clean up delaythreadFactory - the factory to create threads withpublic EmbeddedEventStore(EventStorageEngine storageEngine, MessageMonitor<? super EventMessage<?>> monitor, int cachedEvents, long fetchDelay, long cleanupDelay, TimeUnit timeUnit, ThreadFactory threadFactory, boolean optimizeEventConsumption)
storageEngine - the storage engine to usemonitor - the metrics monitor that tracks how many events are ingested by the event storecachedEvents - the maximum number of events in the cache that is shared between the streams of
tracking event processorsfetchDelay - the time to wait before fetching new events from the backing storage engine while
tracking after a previous stream was fetched and read. Note that this only
applies to situations in which no events from the current application have
meanwhile been committed. If the current application commits events then those
events are fetched without delay.cleanupDelay - the delay between two clean ups of lagging event processors. An event processor
is lagging behind and removed from the set of processors that track cached events
if the oldest event in the cache is newer than the last processed event of the
event processor. Once removed the processor will be independently fetching
directly from the event storage engine until it has caught up again. Event
processors will not notice this change during tracking (i.e. the stream is not '
closed when an event processor falls behind and is removed).timeUnit - time unit for fetch and clean up delaythreadFactory - the factory to create threads withoptimizeEventConsumption - toggle whether event consumption should be optimized. If set to true,
distinct Event Consumers will read events from the same stream as soon as they
reach the head of the stream. If false, they will stay on a private
stream. The latter means more database resources will be used. This can be
configured by either adjusting the property through the constructor or by
setting a system property with key optimize-event-consumption and value
true/false. If nothing is provided, this will be defaulted to
true@PreDestroy public void shutDown()
protected void afterCommit(List<? extends EventMessage<?>> events)
AbstractEventBusevents after the Unit of Work has been committed. The default implementation does
nothing.afterCommit in class AbstractEventBusevents - Events to be published by this Event Buspublic TrackingEventStream openStream(TrackingToken trackingToken)
EventBustrackingToken of null to open a
stream containing all available events. Note that the returned stream is infinite, so beware of applying
terminal operations to the returned stream.
In case the event bus cannot open a stream for a given tracking token, for instance because the event bus does
not persist or cache events, the event bus will throw an UnsupportedOperationException.
trackingToken - object describing the previous position in the stream or null to create a
stream of all eventsTrackingEventStream with events since the given trackingTokenCopyright © 2010–2019. All rights reserved.