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

Since:
3.0
Author:
Rene de Waele

  • Constructor Details

  • Method Details

    • builder

      public static JpaEventStorageEngine.Builder builder()
      Instantiate a Builder to be able to create a JpaEventStorageEngine.

      The following configurable fields have defaults:

      The event and snapshot Serializer, the EntityManagerProvider and TransactionManager are hard requirements and as such should be provided.

      Returns:
      a Builder to be able to create a JpaEventStorageEngine

    • asDomainEventMessage

      protected static <T> DomainEventMessage<T> asDomainEventMessage(EventMessage<T> event)
      Converts an EventMessage to a DomainEventMessage. If the message already is a DomainEventMessage it will be returned as is. Otherwise, a new GenericDomainEventMessage is made with null type, aggregateIdentifier equal to messageIdentifier and sequence number of 0L.

      Doing so allows using the DomainEventEntry to store both a GenericEventMessage and a GenericDomainEventMessage.

      Type Parameters:
      T - the type of payload in the message
      Parameters:
      event - the input event message
      Returns:
      the message converted to a domain event message

    • fetchEvents

      protected List<Object[]> fetchEvents(GapAwareTrackingToken token)
      Returns a batch of event data as object entries in the event storage with a greater than the given token. Size of event is decided by BatchingEventStorageEngine.batchSize().
      Parameters:
      token - Object describing the global index of the last processed event.
      Returns:
      A batch of event messages as object stored since the given tracking token.

    • 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. Implementations for which this is not always the case should override BatchingEventStorageEngine.fetchForAggregateUntilEmpty() to return true and preferably configure a better BatchingEventStorageEngine.Builder.finalAggregateBatchPredicate(Predicate) to provide a better heuristic for detecting the last batch in a stream.

      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 Stream<? extends DomainEventData<?>> readSnapshotData(String aggregateIdentifier)
      Description copied from class: AbstractEventStorageEngine
      Returns a stream of serialized event entries for given aggregateIdentifier if the backing database contains a snapshot of the aggregate.

      It is required that specific event storage engines return snapshots in descending order of their sequence number.

      Specified by:
      readSnapshotData in class AbstractEventStorageEngine
      Parameters:
      aggregateIdentifier - The aggregate identifier to fetch a snapshot for
      Returns:
      A stream of serialized snapshots 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(@Nonnull 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(@Nonnull 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 Snapshot event entity. Defaults to 'SnapshotEventEntry'.
      Returns:
      the name of the Jpa snapshot entity

    • entityManager

      protected jakarta.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.