org.axonframework.eventsourcing.eventstore.jpa

Class JpaEventStorageEngine

java.lang.Object

org.axonframework.eventsourcing.eventstore.AbstractEventStorageEngine

org.axonframework.eventsourcing.eventstore.BatchingEventStorageEngine

org.axonframework.eventsourcing.eventstore.jpa.JpaEventStorageEngine

All Implemented Interfaces:

EventStorageEngine

public class JpaEventStorageEngine
extends BatchingEventStorageEngine

EventStorageEngine implementation that uses JPA to store and fetch events.

By default the payload of events is stored as a serialized blob of bytes. Other columns are used to store meta-data that allow quick finding of DomainEvents for a specific aggregate in the correct order.

Author:

Rene de Waele

Constructor Summary

Constructors

Constructor and Description
JpaEventStorageEngine(EntityManagerProvider entityManagerProvider, TransactionManager transactionManager) Initializes an EventStorageEngine that uses JPA to store and load events.
JpaEventStorageEngine(Serializer serializer, EventUpcaster upcasterChain, DataSource dataSource, EntityManagerProvider entityManagerProvider, TransactionManager transactionManager) Initializes an EventStorageEngine that uses JPA to store and load events.
JpaEventStorageEngine(Serializer snapshotSerializer, EventUpcaster upcasterChain, DataSource dataSource, Serializer eventSerializer, EntityManagerProvider entityManagerProvider, TransactionManager transactionManager) Initializes an EventStorageEngine that uses JPA to store and load events.
JpaEventStorageEngine(Serializer serializer, EventUpcaster upcasterChain, PersistenceExceptionResolver persistenceExceptionResolver, EntityManagerProvider entityManagerProvider, TransactionManager transactionManager) Initializes an EventStorageEngine that uses JPA to store and load events.
JpaEventStorageEngine(Serializer snapshotSerializer, EventUpcaster upcasterChain, PersistenceExceptionResolver persistenceExceptionResolver, Serializer eventSerializer, EntityManagerProvider entityManagerProvider, TransactionManager transactionManager) Initializes an EventStorageEngine that uses JPA to store and load events.
JpaEventStorageEngine(Serializer snapshotSerializer, EventUpcaster upcasterChain, PersistenceExceptionResolver persistenceExceptionResolver, Serializer eventSerializer, Integer batchSize, EntityManagerProvider entityManagerProvider, TransactionManager transactionManager, Long lowestGlobalSequence, Integer maxGapOffset, boolean explicitFlush) Initializes an EventStorageEngine that uses JPA to store and load events.

Method Summary

Modifier and Type	Method and Description
protected void	appendEvents(List<? extends EventMessage<?>> events, Serializer serializer) Append given events to the backing database.
protected Object	createEventEntity(EventMessage<?> eventMessage, Serializer serializer) Returns a Jpa event entity for given eventMessage.
TrackingToken	createHeadToken() Creates a token that is at the head of an event stream - that tracks all new events.
protected Object	createSnapshotEntity(DomainEventMessage<?> snapshot, Serializer serializer) Returns a Jpa snapshot entity for given snapshot of an aggregate.
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.
protected void	deleteSnapshots(String aggregateIdentifier, long sequenceNumber) Deletes all snapshots from the underlying storage with given aggregateIdentifier.
protected String	domainEventEntryEntityName() Returns the name of the Jpa event entity.
protected javax.persistence.EntityManager	entityManager() Provides an EntityManager instance for storing and fetching event data.
protected List<? extends DomainEventData<?>>	fetchDomainEvents(String aggregateIdentifier, long firstSequenceNumber, int batchSize) Returns a batch of events published by an aggregate with given aggregateIdentifier.
protected List<? extends TrackedEventData<?>>	fetchTrackedEvents(TrackingToken lastToken, int batchSize) Returns a batch of serialized event data entries in the event storage that have a TrackingToken greater than the given lastToken.
Optional<Long>	lastSequenceNumberFor(String aggregateIdentifier) Returns the last known sequence number for the given aggregateIdentifier.
protected 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	setGapCleaningThreshold(int gapCleaningThreshold) Sets the threshold of number of gaps in a token before an attempt to clean gaps up is taken.
void	setGapTimeout(int gapTimeout) Sets the amount of time until a 'gap' in a TrackingToken may be considered timed out.
protected String	snapshotEventEntryEntityName() Returns the name of the Snaphot event entity.
protected void	storeSnapshot(DomainEventMessage<?> snapshot, Serializer serializer) Store the given snapshot of an Aggregate.

Methods inherited from class org.axonframework.eventsourcing.eventstore.BatchingEventStorageEngine

batchSize, readEventData, readEventData

Methods inherited from class org.axonframework.eventsourcing.eventstore.AbstractEventStorageEngine

appendEvents, getEventSerializer, getSerializer, handlePersistenceException, readEvents, readEvents, readSnapshot, storeSnapshot

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, readEvents

Constructor Detail

JpaEventStorageEngine

public JpaEventStorageEngine(EntityManagerProvider entityManagerProvider,
                             TransactionManager transactionManager)

Initializes an EventStorageEngine that uses JPA to store and load events. The payload and metadata of events is stored as a serialized blob of bytes using a new XStreamSerializer.

Events are read in batches of 100. No upcasting is performed after the events have been fetched.

Parameters:

entityManagerProvider - Provider for the EntityManager used by this EventStorageEngine.

transactionManager - The instance managing transactions around fetching event data. Required by certain databases for reading blob data.

JpaEventStorageEngine

public JpaEventStorageEngine(Serializer serializer,
                             EventUpcaster upcasterChain,
                             DataSource dataSource,
                             EntityManagerProvider entityManagerProvider,
                             TransactionManager transactionManager)
                      throws SQLException

Initializes an EventStorageEngine that uses JPA to store and load events. Events are fetched in batches of 100. The same Serializer is used for both snapshots and events.

Parameters:

serializer - Used to serialize and deserialize event payload and metadata, and snapshots.

upcasterChain - Allows older revisions of serialized objects to be deserialized.

dataSource - Allows the EventStore to detect the database type and define the error codes that represent concurrent access failures for most database types.

entityManagerProvider - Provider for the EntityManager used by this EventStorageEngine.

transactionManager - The instance managing transactions around fetching event data. Required by certain databases for reading blob data.

Throws:

SQLException - If the database product name can not be determined from the given dataSource

JpaEventStorageEngine

public JpaEventStorageEngine(Serializer snapshotSerializer,
                             EventUpcaster upcasterChain,
                             DataSource dataSource,
                             Serializer eventSerializer,
                             EntityManagerProvider entityManagerProvider,
                             TransactionManager transactionManager)
                      throws SQLException

Initializes an EventStorageEngine that uses JPA to store and load events. Events are fetched in batches of 100.

Parameters:

snapshotSerializer - Used to serialize and deserialize snapshots.

upcasterChain - Allows older revisions of serialized objects to be deserialized.

dataSource - Allows the EventStore to detect the database type and define the error codes that represent concurrent access failures for most database types.

eventSerializer - Used to serialize and deserialize event payload and metadata.

entityManagerProvider - Provider for the EntityManager used by this EventStorageEngine.

transactionManager - The instance managing transactions around fetching event data. Required by certain databases for reading blob data.

Throws:

SQLException - If the database product name can not be determined from the given dataSource

JpaEventStorageEngine

public JpaEventStorageEngine(Serializer serializer,
                             EventUpcaster upcasterChain,
                             PersistenceExceptionResolver persistenceExceptionResolver,
                             EntityManagerProvider entityManagerProvider,
                             TransactionManager transactionManager)

Initializes an EventStorageEngine that uses JPA to store and load events. Events are fetched in batches of 100. The same Serializer is used for both snapshots and events.

Parameters:

serializer - Used to serialize and deserialize event payload and metadata, and snapshots.

upcasterChain - Allows older revisions of serialized objects to be deserialized.

persistenceExceptionResolver - Detects concurrency exceptions from the backing database. If null persistence exceptions are not explicitly resolved. that represent concurrent access failures for most database types.

entityManagerProvider - Provider for the EntityManager used by this EventStorageEngine.

transactionManager - The instance managing transactions around fetching event data. Required by certain databases for reading blob data.

JpaEventStorageEngine

public JpaEventStorageEngine(Serializer snapshotSerializer,
                             EventUpcaster upcasterChain,
                             PersistenceExceptionResolver persistenceExceptionResolver,
                             Serializer eventSerializer,
                             EntityManagerProvider entityManagerProvider,
                             TransactionManager transactionManager)

Initializes an EventStorageEngine that uses JPA to store and load events. Events are fetched in batches of 100.

Parameters:

snapshotSerializer - Used to serialize and deserialize snapshots.

upcasterChain - Allows older revisions of serialized objects to be deserialized.

persistenceExceptionResolver - Detects concurrency exceptions from the backing database. If null persistence exceptions are not explicitly resolved. that represent concurrent access failures for most database types.

eventSerializer - Used to serialize and deserialize event payload and metadata.

entityManagerProvider - Provider for the EntityManager used by this EventStorageEngine.

transactionManager - The instance managing transactions around fetching event data. Required by certain databases for reading blob data.

JpaEventStorageEngine

public JpaEventStorageEngine(Serializer snapshotSerializer,
                             EventUpcaster upcasterChain,
                             PersistenceExceptionResolver persistenceExceptionResolver,
                             Serializer eventSerializer,
                             Integer batchSize,
                             EntityManagerProvider entityManagerProvider,
                             TransactionManager transactionManager,
                             Long lowestGlobalSequence,
                             Integer maxGapOffset,
                             boolean explicitFlush)

Initializes an EventStorageEngine that uses JPA to store and load events.

Parameters:

snapshotSerializer - Used to serialize and deserialize snapshots.

upcasterChain - Allows older revisions of serialized objects to be deserialized.

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.

batchSize - The number of events that should be read at each database access. When more than this number of events must be read to rebuild an aggregate's state, the events are read in batches of this size. Tip: if you use a snapshotter, make sure to choose snapshot trigger and batch size such that a single batch will generally retrieve all events required to rebuild an aggregate's state.

entityManagerProvider - Provider for the EntityManager used by this EventStorageEngine.

maxGapOffset - The maximum distance in sequence numbers between a missing event and the event with the highest known index. If the gap is bigger it is assumed that the missing event will not be committed to the store anymore. This event storage engine will no longer look for those events the next time a batch is fetched.

explicitFlush - Whether to explicitly call EntityManager.flush() after inserting the Events published in this Unit of Work. If false, this instance relies on the transaction manager to flush data. Note that the persistenceExceptionResolver may not be able to translate exceptions anymore. false Should only be used to optimize performance for batch operations. In other cases, true is recommended.

transactionManager - The instance managing transactions around fetching event data. Required by certain databases for reading blob data.

lowestGlobalSequence - The first expected auto generated sequence number. For most data stores this is 1 unless the table has contained entries before.

Method Detail

fetchTrackedEvents

protected List<? extends TrackedEventData<?>> fetchTrackedEvents(TrackingToken lastToken,
                                                                 int batchSize)

Description copied from class: BatchingEventStorageEngine

Returns a batch of serialized event data entries in the event storage that have a TrackingToken greater than the given lastToken. Event entries in the stream should be ordered by tracking token. If the lastToken is null a stream containing all events should be returned.

Only if the returned List is empty the event storage assumes that the backing database holds no further applicable entries.

Specified by:

fetchTrackedEvents in class BatchingEventStorageEngine

Parameters:

lastToken - Object describing the global index of the last processed event or null to create a stream of all events in the store

batchSize - The maximum number of events that should be returned

Returns:

A batch of tracked event messages stored since the given tracking token

fetchDomainEvents

protected List<? extends DomainEventData<?>> fetchDomainEvents(String aggregateIdentifier,
                                                               long firstSequenceNumber,
                                                               int batchSize)

Description copied from class: BatchingEventStorageEngine

Returns a batch of events published by an aggregate with given aggregateIdentifier.

The sequence numbers in the returned batch should be ordered by sequence number. The first event in the batch should have a sequence number equal to or larger than given firstSequenceNumber. Implementations should make sure the returned batch does not contain gaps between events due to uncommitted storage transactions.

If the returned number of entries is smaller than the given batchSize it is assumed that the storage holds no further applicable entries.

Specified by:

fetchDomainEvents in class BatchingEventStorageEngine

Parameters:

aggregateIdentifier - The identifier of the aggregate to open a stream for

firstSequenceNumber - The sequence number of the first excepted event entry

batchSize - The maximum number of events that should be returned

Returns:

a batch of serialized event entries for the given aggregate

readSnapshotData

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

Description copied from class: AbstractEventStorageEngine

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

Specified by:

readSnapshotData in class AbstractEventStorageEngine

Parameters:

aggregateIdentifier - The aggregate identifier to fetch a snapshot for

Returns:

An optional with a serialized snapshot of the aggregate

appendEvents

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

Description copied from class: AbstractEventStorageEngine

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

Specified by:

appendEvents in class AbstractEventStorageEngine

Parameters:

events - Events to append to the database

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

storeSnapshot

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

Description copied from class: AbstractEventStorageEngine

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

Specified by:

storeSnapshot in class AbstractEventStorageEngine

Parameters:

snapshot - Snapshot Event of the aggregate

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

lastSequenceNumberFor

public Optional<Long> lastSequenceNumberFor(String aggregateIdentifier)

Description copied from interface: EventStorageEngine

Returns the last known sequence number for the given 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.

Parameters:

aggregateIdentifier - The identifier to find the last sequence number for

Returns:

an optional with the highest sequence number, or an empty optional if the aggregate identifier wasn't found

createTailToken

public TrackingToken createTailToken()

Description copied from interface: EventStorageEngine

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

Returns:

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

createHeadToken

public TrackingToken createHeadToken()

Description copied from interface: EventStorageEngine

Creates a token that is at the head of an event stream - that tracks all new events.

Returns:

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

createTokenAt

public TrackingToken createTokenAt(Instant dateTime)

Description copied from interface: EventStorageEngine

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.

Parameters:

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.

Returns:

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

deleteSnapshots

protected void deleteSnapshots(String aggregateIdentifier,
                               long sequenceNumber)

Deletes all snapshots from the underlying storage with given aggregateIdentifier.

Parameters:

aggregateIdentifier - the identifier of the aggregate to delete snapshots for

sequenceNumber - The sequence number from which value snapshots should be kept

createEventEntity

protected Object createEventEntity(EventMessage<?> eventMessage,
                                   Serializer serializer)

Returns a Jpa event entity for given eventMessage. Use the given serializer to serialize the payload and metadata of the event.

Parameters:

eventMessage - the event message to store

serializer - the serializer to serialize the payload and metadata

Returns:

the Jpa entity to be inserted

createSnapshotEntity

protected Object createSnapshotEntity(DomainEventMessage<?> snapshot,
                                      Serializer serializer)

Returns a Jpa snapshot entity for given snapshot of an aggregate. Use the given serializer to serialize the payload and metadata of the snapshot event.

Parameters:

snapshot - the domain event message containing a snapshot of the aggregate

serializer - the serializer to serialize the payload and metadata

Returns:

the Jpa entity to be inserted

domainEventEntryEntityName

protected String domainEventEntryEntityName()

Returns the name of the Jpa event entity. Defaults to 'DomainEventEntry'.

Returns:

the name of the Jpa event entity

snapshotEventEntryEntityName

protected String snapshotEventEntryEntityName()

Returns the name of the Snaphot event entity. Defaults to 'SnapshotEventEntry'.

Returns:

the name of the Jpa snapshot entity

entityManager

protected javax.persistence.EntityManager entityManager()

Provides an EntityManager instance for storing and fetching event data.

Returns:

a provided entity manager

setGapTimeout

public void setGapTimeout(int gapTimeout)

Sets the amount of time until a 'gap' in a TrackingToken may be considered timed out. This setting will affect the cleaning process of gaps. Gaps that have timed out will be removed from Tracking Tokens to improve performance of reading events. Defaults to 60000 (1 minute).

Parameters:

gapTimeout - The amount of time, in milliseconds until a gap may be considered timed out.

setGapCleaningThreshold

public void setGapCleaningThreshold(int gapCleaningThreshold)

Sets the threshold of number of gaps in a token before an attempt to clean gaps up is taken. Defaults to 250.

Parameters:

gapCleaningThreshold - The number of gaps before triggering a cleanup.