org.axonframework.eventhandling.tokenstore.jdbc

Class JdbcTokenStore

java.lang.Object

org.axonframework.eventhandling.tokenstore.jdbc.JdbcTokenStore

All Implemented Interfaces:

TokenStore

public class JdbcTokenStore
extends Object
implements TokenStore

Implementation of a token store that uses JDBC to save and load tokens. Before using this store make sure the database contains a table named TokenSchema.tokenTable() in which to store the tokens.

Author:

Rene de Waele

Constructor Summary

Constructors

Constructor and Description
JdbcTokenStore(ConnectionProvider connectionProvider, Serializer serializer) Initialize the JpaTokenStore with given resources.
JdbcTokenStore(ConnectionProvider connectionProvider, Serializer serializer, TokenSchema schema, TemporalAmount claimTimeout, String nodeId, Class<?> contentType) Initialize the JpaTokenStore with given resources.

Method Summary

Modifier and Type	Method and Description
protected TrackingToken	claimToken(ResultSet resultSet, AbstractTokenEntry<?> entry) Tries to claim the given token entry.
void	createSchema(TokenTableFactory schemaFactory) Performs the DDL queries to create the schema necessary for this token store implementation.
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.
protected Connection	getConnection() Returns a Connection to the database.
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 void	insertOrUpdateToken(ResultSet resultSet, TrackingToken token, String processorName, int segment) If the given resultSet has no items this method should insert a new token entry.
protected TrackingToken	insertTokenEntry(ResultSet resultSet, TrackingToken token, String processorName, int segment) Inserts a new token entry via the given updatable resultSet.
protected TrackingToken	loadOrInsertToken(ResultSet resultSet, String processorName, int segment) Tries loading an existing token owned by a processor with given processorName and segment.
protected <T> T	readSerializedData(ResultSet resultSet, String columnName) Returns the serialized token data from the given resultSet at given columnName.
protected AbstractTokenEntry<?>	readTokenEntry(ResultSet resultSet) Convert given resultSet to an AbstractTokenEntry.
protected PreparedStatement	releaseClaim(Connection connection, String processorName, int segment) Creates a new PreparedStatement to release the current claim this node has on a token belonging to a processor with given processorName and segment.
void	releaseClaim(String processorName, int segment) Release a claim of the token for given processorName and segment.
protected PreparedStatement	selectForSegments(Connection connection, String processorName) Returns a PreparedStatement to select all segments ids for a given processorName from the underlying storage.
protected PreparedStatement	selectForUpdate(Connection connection, String processorName, int segment) Returns a PreparedStatement to select a token entry from the underlying storage.
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

Methods inherited from interface org.axonframework.eventhandling.tokenstore.TokenStore

extendClaim

Constructor Detail

JdbcTokenStore

public JdbcTokenStore(ConnectionProvider connectionProvider,
                      Serializer serializer)

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. The token is serialized to a byte array.

Parameters:

connectionProvider - The provider of connections to the underlying database

serializer - The serializer to serialize tokens with

JdbcTokenStore

public JdbcTokenStore(ConnectionProvider connectionProvider,
                      Serializer serializer,
                      TokenSchema schema,
                      TemporalAmount claimTimeout,
                      String nodeId,
                      Class<?> contentType)

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:

connectionProvider - The provider of connections to the underlying database

serializer - The serializer to serialize tokens with

schema - The schema that describes a Jdbc token entry

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

nodeId - The identifier to identify ownership of the tokens

contentType - The data type of the serialized token

Method Detail

createSchema

public void createSchema(TokenTableFactory schemaFactory)

Performs the DDL queries to create the schema necessary for this token store implementation.

Parameters:

schemaFactory - factory of the token entry schema

Throws:

EventStoreException - when an error occurs executing SQL statements

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)
                throws UnableToClaimTokenException

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

Throws:

UnableToClaimTokenException - when the token being updated has been claimed by another process.

fetchToken

public TrackingToken fetchToken(String processorName,
                                int segment)
                         throws UnableToClaimTokenException

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

Throws:

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

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

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.

selectForSegments

protected PreparedStatement selectForSegments(Connection connection,
                                              String processorName)
                                       throws SQLException

Returns a PreparedStatement to select all segments ids for a given processorName from the underlying storage.

Parameters:

connection - the connection to the underlying database

processorName - the name of the processor to fetch the segments for

Returns:

a PreparedStatement that will fetch segments when executed

Throws:

SQLException - when an exception occurs while creating the prepared statement

selectForUpdate

protected PreparedStatement selectForUpdate(Connection connection,
                                            String processorName,
                                            int segment)
                                     throws SQLException

Returns a PreparedStatement to select a token entry from the underlying storage. The ResultSet that is returned when this statement is executed should be updatable.

Parameters:

connection - the connection to the underlying database

processorName - the name of the processor to fetch the entry for

segment - the segment of the processor to fetch the entry for

Returns:

a PreparedStatement that will fetch an updatable token entry when executed

Throws:

SQLException - when an exception occurs while creating the prepared statement

insertOrUpdateToken

protected void insertOrUpdateToken(ResultSet resultSet,
                                   TrackingToken token,
                                   String processorName,
                                   int segment)
                            throws SQLException

If the given resultSet has no items this method should insert a new token entry. If a token already exists it should be attempted to replace the token in the entry with the given token and claim ownership.

Parameters:

resultSet - the updatable query result set of an executed PreparedStatement

token - the token for the new or updated entry

processorName - the name of the processor owning the token

segment - the segment of the processor owning the token

Throws:

UnableToClaimTokenException - if the token cannot be claimed because another node currently owns the token

SQLException - when an exception occurs while updating the result set

claimToken

protected TrackingToken claimToken(ResultSet resultSet,
                                   AbstractTokenEntry<?> entry)
                            throws SQLException

Tries to claim the given token entry. If the claim fails an UnableToClaimTokenException should be thrown. Otherwise the given resultSet should be updated to reflect the claim.

Parameters:

resultSet - the updatable query result of an executed PreparedStatement

entry - the entry extracted from the given result set

Returns:

the claimed tracking token

Throws:

UnableToClaimTokenException - if the token cannot be claimed because another node currently owns the token

SQLException - when an exception occurs while claiming the token entry

loadOrInsertToken

protected TrackingToken loadOrInsertToken(ResultSet resultSet,
                                          String processorName,
                                          int segment)
                                   throws SQLException

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 null token owned by this node and return null.

Parameters:

resultSet - the updatable result set from a prior select for update query

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

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

SQLException - when an exception occurs while loading or inserting the entry

insertTokenEntry

protected TrackingToken insertTokenEntry(ResultSet resultSet,
                                         TrackingToken token,
                                         String processorName,
                                         int segment)
                                  throws SQLException

Inserts a new token entry via the given updatable resultSet.

Parameters:

resultSet - the updatable result set to add the entry to

token - the token of the entry to insert

processorName - the name of the processor to insert a token for

segment - the segment of the processor to insert a token for

Returns:

the tracking token of the inserted entry

Throws:

SQLException - when an exception occurs while inserting a token entry

readTokenEntry

protected AbstractTokenEntry<?> readTokenEntry(ResultSet resultSet)
                                        throws SQLException

Convert given resultSet to an AbstractTokenEntry. The result set contains a single token entry.

Parameters:

resultSet - the result set of a prior select statement containing a single token entry

Returns:

an token entry with data extracted from the result set

Throws:

SQLException - if the result set cannot be converted to an entry

releaseClaim

protected PreparedStatement releaseClaim(Connection connection,
                                         String processorName,
                                         int segment)
                                  throws SQLException

Creates a new PreparedStatement to release the current claim this node has on a token belonging to a processor with given processorName and segment.

Parameters:

connection - the connection that should be used to create a PreparedStatement

processorName - the name of the processor for which to release this node's claim

segment - the segment of the processor for which to release this node's claim

Returns:

a PreparedStatement that will release the claim this node has on the token entry

Throws:

SQLException - if the statement to release a claim cannot be created

readSerializedData

protected <T> T readSerializedData(ResultSet resultSet,
                                   String columnName)
                            throws SQLException

Returns the serialized token data from the given resultSet at given columnName.

Type Parameters:

T - the type of data to return

Parameters:

resultSet - the result set to get serialized data from

columnName - the name of the column containing the serialized token

Returns:

the serialized data of the token

Throws:

SQLException - if the token cannot be read from the entry

getConnection

protected Connection getConnection()

Returns a Connection to the database.

Returns:

a database Connection