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.

Author:

Rene de Waele

Constructor Summary

Constructors

Constructor and Description
JpaTokenStore(EntityManagerProvider entityManagerProvider, Serializer serializer) Initializes a token store with given entityManagerProvider and serializer.
JpaTokenStore(EntityManagerProvider entityManagerProvider, Serializer serializer, TemporalAmount claimTimeout, String nodeId) Initialize the JpaTokenStore with given resources.

Method Summary

Modifier and Type	Method and Description
void	extendClaim(String processorName, int segment) Extends the claim on the current token held by the this node for the given processorName and segment.
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.
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	loadOrCreateToken(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.
void	releaseClaim(String processorName, int segment) Release a claim of the token for given processorName and segment.
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

public JpaTokenStore(EntityManagerProvider entityManagerProvider,
                     Serializer serializer)

Initializes a token store with given entityManagerProvider and serializer.

Parameters:

entityManagerProvider - The provider of the entity manager

serializer - The serializer used to serialize tokens

JpaTokenStore

public JpaTokenStore(EntityManagerProvider entityManagerProvider,
                     Serializer serializer,
                     TemporalAmount claimTimeout,
                     String nodeId)

Initialize the JpaTokenStore with given resources. The given claimTimeout is used to 'steal' any claim that has not been updated since that amount of time.

Parameters:

entityManagerProvider - provides the EntityManager to access the underlying database

serializer - The serializer to serialize tokens with

claimTimeout - The timeout after which this process will force a claim

nodeId - The identifier to identify ownership of the tokens

Method Detail

initializeTokenSegments

public void initializeTokenSegments(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(String processorName,
                                    int segmentCount,
                                    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(TrackingToken token,
                       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.

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(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

fetchToken

public TrackingToken fetchToken(String processorName,
                                int segment)

Description copied from interface: TokenStore

Returns the last stored token for the given processorName and segment. Returns null if the store holds no token or if the stored token for the given process and segment is null.

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

extendClaim

public void extendClaim(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(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.

loadOrCreateToken

protected TokenEntry loadOrCreateToken(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.