org.axonframework.eventhandling.tokenstore.jpa

Class JpaTokenStore

java.lang.Object

org.axonframework.eventhandling.tokenstore.jpa.JpaTokenStore

All Implemented Interfaces:

TokenStore

public class JpaTokenStore
extends Object
implements TokenStore

Implementation of a token store that uses JPA to save and load tokens. This implementation uses TokenEntry entities.

Since:

3.0

Author:

Rene de Waele

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	JpaTokenStore.Builder Builder class to instantiate a JpaTokenStore.

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	JpaTokenStore(JpaTokenStore.Builder builder) Instantiate a JpaTokenStore based on the fields contained in the JpaTokenStore.Builder.

Method Summary

Modifier and Type	Method and Description
static JpaTokenStore.Builder	builder() Instantiate a Builder to be able to create a JpaTokenStore.
void	deleteToken(String processorName, int segment) Deletes the token for the processor with given processorName and segment.
void	extendClaim(String processorName, int segment) Extends the claim on the current token held by the this node for the given processorName and segment.
List<Segment>	fetchAvailableSegments(String processorName) Returns a List of known available segments for a given processorName.
int[]	fetchSegments(String processorName) Returns an array of known segments for a given processorName.
TrackingToken	fetchToken(String processorName, int segment) Returns the last stored token for the given processorName and segment.
TrackingToken	fetchToken(String processorName, Segment segment) Returns the last stored token for the given processorName and segment.
void	initializeSegment(TrackingToken token, String processorName, int segment) Initializes a segment with given segment for the processor with given processorName to contain the given token.
void	initializeTokenSegments(String processorName, int segmentCount) Initializes the given segmentCount number of segments for the given processorName to track its tokens.
void	initializeTokenSegments(String processorName, int segmentCount, TrackingToken initialToken) Initializes the given segmentCount number of segments for the given processorName to track its tokens.
protected TokenEntry	loadToken(String processorName, int segment, javax.persistence.EntityManager entityManager) Loads an existing TokenEntry or creates a new one using the given entityManager for given processorName and segment.
protected TokenEntry	loadToken(String processorName, Segment segment, javax.persistence.EntityManager entityManager) Tries loading an existing token owned by a processor with given processorName and segment.
void	releaseClaim(String processorName, int segment) Release a claim of the token for given processorName and segment.
boolean	requiresExplicitSegmentInitialization() Indicates whether this TokenStore instance requires segments to be explicitly initialized, before any tokens can be claimed for that segment.
Optional<String>	retrieveStorageIdentifier() Returns a unique identifier that uniquely identifies the storage location of the tokens in this store.
Serializer	serializer() Returns the serializer used by the Token Store to serialize tokens.
void	storeToken(TrackingToken token, String processorName, int segment) Stores the given token in the store.

Methods inherited from class java.lang.Object

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

Constructor Detail

JpaTokenStore

protected JpaTokenStore(JpaTokenStore.Builder builder)

Instantiate a JpaTokenStore based on the fields contained in the JpaTokenStore.Builder.

Will assert that the EntityManager, Serializer, claimTimeout and nodeId are not null, and will throw an AxonConfigurationException if any of them is null.

Parameters:

builder - the JpaTokenStore.Builder used to instantiate a JpaTokenStore instance

Method Detail

builder

public static JpaTokenStore.Builder builder()

Instantiate a Builder to be able to create a JpaTokenStore.

The claimTimeout to a 10 seconds duration, and nodeId is defaulted to the name of the managed bean for the runtime system of the Java virtual machine. The EntityManagerProvider and Serializer are a hard requirements and as such should be provided.

Returns:

a Builder to be able to create a JpaTokenStore

initializeTokenSegments

public void initializeTokenSegments(@Nonnull
                                    String processorName,
                                    int segmentCount)
                             throws UnableToClaimTokenException

Description copied from interface: TokenStore

Initializes the given segmentCount number of segments for the given processorName to track its tokens. This method should only be invoked when no tokens have been stored for the given processor, yet.

This method will initialize the tokens, but not claim them. It will create the segments ranging from 0 until segmentCount - 1.

The exact behavior when this method is called while tokens were already present, is undefined in case the token already present is not owned by the initializing process.

Specified by:

initializeTokenSegments in interface TokenStore

Parameters:

processorName - The name of the processor to initialize segments for

segmentCount - The number of segments to initialize

Throws:

UnableToClaimTokenException - when a segment has already been created

initializeTokenSegments

public void initializeTokenSegments(@Nonnull
                                    String processorName,
                                    int segmentCount,
                                    @Nullable
                                    TrackingToken initialToken)
                             throws UnableToClaimTokenException

Description copied from interface: TokenStore

Initializes the given segmentCount number of segments for the given processorName to track its tokens. This method should only be invoked when no tokens have been stored for the given processor, yet.

This method will store initialToken for all segments as starting point for processor, but not claim them. It will create the segments ranging from 0 until segmentCount - 1.

The exact behavior when this method is called while tokens were already present, is undefined in case the token already present is not owned by the initializing process.

Specified by:

initializeTokenSegments in interface TokenStore

Parameters:

processorName - The name of the processor to initialize segments for

segmentCount - The number of segments to initialize

initialToken - The initial token which is used as a starting point for processor

Throws:

UnableToClaimTokenException - when a segment has already been created

storeToken

public void storeToken(@Nullable
                       TrackingToken token,
                       @Nonnull
                       String processorName,
                       int segment)

Description copied from interface: TokenStore

Stores the given token in the store. The token marks the current position of the process with given processorName and segment. The given token may be null.

Any claims made by the current process have their timestamp updated.

This method should throw an UnableToClaimTokenException when the given segment has not been initialized with a Token (albeit null) yet. In that case, a segment must have been explicitly initialized. A TokenStore implementation's ability to do so is exposed by the TokenStore.requiresExplicitSegmentInitialization() method. If that method returns false, this method may implicitly initialize a token and return that token upon invocation.

Specified by:

storeToken in interface TokenStore

Parameters:

token - The token to store for a given process and segment. May be null.

processorName - The name of the process for which to store the token

segment - The index of the segment for which to store the token

releaseClaim

public void releaseClaim(@Nonnull
                         String processorName,
                         int segment)

Description copied from interface: TokenStore

Release a claim of the token for given processorName and segment. If no such claim existed, nothing happens.

The caller must ensure not to use any streams opened based on the token for which the claim is released.

Specified by:

releaseClaim in interface TokenStore

Parameters:

processorName - The name of the process owning the token (e.g. a TrackingEventProcessor name)

segment - the segment for which a token was obtained

initializeSegment

public void initializeSegment(@Nullable
                              TrackingToken token,
                              @Nonnull
                              String processorName,
                              int segment)
                       throws UnableToInitializeTokenException

Description copied from interface: TokenStore

Initializes a segment with given segment for the processor with given processorName to contain the given token.

This method fails if a Token already exists for the given processor and segment, even if that token has been claimed by the active instance.

This method will not claim the initialized segment. Use TokenStore.fetchToken(String, int) to retrieve and claim the token.

Specified by:

initializeSegment in interface TokenStore

Parameters:

token - The token to initialize the segment with

processorName - The name of the processor to create the segment for

segment - The identifier of the segment to initialize

Throws:

UnableToInitializeTokenException - if a Token already exists

requiresExplicitSegmentInitialization

public boolean requiresExplicitSegmentInitialization()

Description copied from interface: TokenStore

Indicates whether this TokenStore instance requires segments to be explicitly initialized, before any tokens can be claimed for that segment.

Specified by:

requiresExplicitSegmentInitialization in interface TokenStore

Returns:

true if this instance requires tokens to be explicitly initialized, otherwise false.

See Also:

TokenStore.initializeTokenSegments(String, int), TokenStore.initializeTokenSegments(String, int, TrackingToken), TokenStore.initializeSegment(TrackingToken, String, int)

deleteToken

public void deleteToken(@Nonnull
                        String processorName,
                        int segment)
                 throws UnableToClaimTokenException

Description copied from interface: TokenStore

Deletes the token for the processor with given processorName and segment. The token must be owned by the current node, to be able to delete it.

Implementations should implement this method only when TokenStore.requiresExplicitSegmentInitialization() is overridden to return true. Deleting tokens using implementations that do not require explicit token initialization is unsafe, as a claim will automatically recreate the deleted token instance, which may result in concurrency issues.

Specified by:

deleteToken in interface TokenStore

Parameters:

processorName - The name of the processor to remove the token for

segment - The segment to delete

Throws:

UnableToClaimTokenException - if the token is not currently claimed by this node

fetchToken

public TrackingToken fetchToken(@Nonnull
                                String processorName,
                                int segment)

Description copied from interface: TokenStore

Returns the last stored token for the given processorName and segment. Returns null if the stored token for the given process and segment is null.

This method should throw an UnableToClaimTokenException when the given segment has not been initialized with a Token (albeit null) yet. In that case, a segment must have been explicitly initialized. A TokenStore implementation's ability to do so is exposed by the TokenStore.requiresExplicitSegmentInitialization() method. If that method returns false, this method may implicitly initialize a token and return that token upon invocation.

The token will be claimed by the current process (JVM instance), preventing access by other instances. To release the claim, use TokenStore.releaseClaim(String, int)

Specified by:

fetchToken in interface TokenStore

Parameters:

processorName - The process name for which to fetch the token

segment - The segment index for which to fetch the token

Returns:

The last stored TrackingToken or null if the store holds no token for given process and segment

fetchToken

public TrackingToken fetchToken(@Nonnull
                                String processorName,
                                @Nonnull
                                Segment segment)
                         throws UnableToClaimTokenException

Description copied from interface: TokenStore

Returns the last stored token for the given processorName and segment. Returns null if the stored token for the given process and segment is null.

This method should throw an UnableToClaimTokenException when the given segment has not been initialized with a Token (albeit null) yet. In that case, a segment must have been explicitly initialized. A TokenStore implementation's ability to do so is exposed by the TokenStore.requiresExplicitSegmentInitialization() method. If that method returns false, this method may implicitly initialize a token and return that token upon invocation.

The token will be claimed by the current process (JVM instance), preventing access by other instances. To release the claim, use TokenStore.releaseClaim(String, int)

Specified by:

fetchToken in interface TokenStore

Parameters:

processorName - The process name for which to fetch the token

segment - The segment for which to fetch the token

Returns:

The last stored TrackingToken or null if the store holds no token for given process and segment

Throws:

UnableToClaimTokenException - if there is a token for given processorName and segment, but they are claimed by another process, or if the segment has been split or merged concurrently

extendClaim

public void extendClaim(@Nonnull
                        String processorName,
                        int segment)
                 throws UnableToClaimTokenException

Description copied from interface: TokenStore

Extends the claim on the current token held by the this node for the given processorName and segment.

Specified by:

extendClaim in interface TokenStore

Parameters:

processorName - The process name for which to fetch the token

segment - The segment index for which to fetch the token

Throws:

UnableToClaimTokenException - if there is no token for given processorName and segment, or if it has been claimed by another process.

fetchSegments

public int[] fetchSegments(@Nonnull
                           String processorName)

Description copied from interface: TokenStore

Returns an array of known segments for a given processorName.

The segments returned are segments for which a token has been stored previously. When the TokenStore is empty, an empty array is returned.

Specified by:

fetchSegments in interface TokenStore

Parameters:

processorName - The process name for which to fetch the segments

Returns:

an array of segment identifiers.

fetchAvailableSegments

public List<Segment> fetchAvailableSegments(@Nonnull
                                            String processorName)

Description copied from interface: TokenStore

Returns a List of known available segments for a given processorName. A segment is considered available if it is not claimed by any other event processor.

The segments returned are segments for which a token has been stored previously and have not been claimed by another processor. When the TokenStore is empty, an empty list is returned. By default, if this method is not implemented, we will return all segments instead, whether they are available or not.

Specified by:

fetchAvailableSegments in interface TokenStore

Parameters:

processorName - the processor's name for which to fetch the segments

Returns:

a List of available segment identifiers for the specified processorName

loadToken

protected TokenEntry loadToken(String processorName,
                               int segment,
                               javax.persistence.EntityManager entityManager)

Loads an existing TokenEntry or creates a new one using the given entityManager for given processorName and segment.

Parameters:

processorName - the name of the event processor

segment - the segment of the event processor

entityManager - the entity manager instance to use for the query

Returns:

the token entry for the given processor name and segment

Throws:

UnableToClaimTokenException - if there is a token for given processorName and segment, but it is claimed by another process.

loadToken

protected TokenEntry loadToken(String processorName,
                               Segment segment,
                               javax.persistence.EntityManager entityManager)

Tries loading an existing token owned by a processor with given processorName and segment. If such a token entry exists an attempt will be made to claim the token. If that succeeds the token will be returned. If the token is already owned by another node an UnableToClaimTokenException will be thrown.

If no such token exists yet, a new token entry will be inserted with a null token, owned by this node, and this method returns null.

If a token has been claimed, the segment will be validated by checking the database for the split and merge candidate segments. If a concurrent split or merge operation has been detected, the calim will be released and an UnableToClaimTokenException will be thrown.}

Parameters:

processorName - the name of the processor to load or insert a token entry for

segment - the segment of the processor to load or insert a token entry for

entityManager - the entity manager instance to use for the query

Returns:

the tracking token of the fetched entry or null if a new entry was inserted

Throws:

UnableToClaimTokenException - if the token cannot be claimed because another node currently owns the token or if the segment has been split or merged concurrently

retrieveStorageIdentifier

public Optional<String> retrieveStorageIdentifier()

Description copied from interface: TokenStore

Returns a unique identifier that uniquely identifies the storage location of the tokens in this store. Two token store implementations that share state, must return the same identifier. Two token store implementations that do not share a location, must return a different identifier (or an empty optional if identifiers are not supported).

Note that this method may require the implementation to consult its underlying storage. Therefore, a Transaction should be active when this method is called, similarly to invocations like TokenStore.fetchToken(String, int), TokenStore.fetchSegments(String), etc. When no Transaction is active, the behavior is undefined.

Specified by:

retrieveStorageIdentifier in interface TokenStore

Returns:

an identifier to uniquely identify the storage location of tokens in this TokenStore.

serializer

public Serializer serializer()

Returns the serializer used by the Token Store to serialize tokens.

Returns:

the serializer used by the Token Store to serialize tokens