Package com.mimecast.robin.queue

Class SQLQueueDatabase<T extends Serializable>

java.lang.Object

com.mimecast.robin.queue.SQLQueueDatabase<T>

Type Parameters:

T - payload type

All Implemented Interfaces:

QueueDatabase<T>, Closeable, AutoCloseable

Direct Known Subclasses:

QueueMariaDB, QueuePgSQL

public abstract class SQLQueueDatabase<T extends Serializable>
extends Object
implements QueueDatabase<T>

SQL-backed scheduled work queue.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
protected static class	SQLQueueDatabase.DBConfig

Field Summary

Fields

Modifier and Type	Field	Description
private com.zaxxer.hikari.HikariDataSource	dataSource
private final String	jdbcUrl
private static final org.apache.logging.log4j.Logger	log
private final int	maxPoolSize
private final String	password
protected final String	tableName
private final String	username

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	SQLQueueDatabase(SQLQueueDatabase.DBConfig config)

Method Summary

Modifier and Type	Method	Description
boolean	acknowledge(String uid)	Acknowledges successful completion and removes the item from the active queue.
void	applyMutations(QueueMutationBatch<T> batch)	Applies a batch of dequeue outcomes and derived enqueues atomically when supported.
private int	bindFilter(PreparedStatement statement, QueueListFilter filter, int startIndex)
private void	bindItem(PreparedStatement statement, QueueItem<T> item)
private String	buildWhereClause(QueueListFilter filter)
List<QueueItem<T>>	claimReady(int limit, long nowEpochSeconds, String consumerId, long claimUntilEpochSeconds)	Claims ready items for the given consumer until the lease expires.
private List<QueueItem<T>>	claimReadyLegacy(int limit, long nowEpochSeconds, String consumerId, long claimUntilEpochSeconds)
void	clear()	Clears the full queue state, including dead items.
void	close()
private long	countByState(QueueItemState state)
private void	deleteBatch(Connection connection, List<String> ackUids)
boolean	deleteByUID(String uid)	Removes the item by queue UID regardless of state.
int	deleteByUIDs(List<String> uids)	Removes multiple items by queue UID regardless of state.
QueueItem<T>	enqueue(QueueItem<T> item)	Inserts a ready queue item.
private List<QueueItem<T>>	fetchByUIDs(Connection connection, List<String> uids)
QueueItem<T>	getByUID(String uid)	Returns the item by queue UID.
protected List<String>	getCreateIndexSQL()
protected abstract String	getCreateTableSQL()
protected abstract String	getDatabaseType()
void	initialize()	Initialize external resources.
private void	insertNewItems(Connection connection, List<T> newItems)
private String	insertSql()
QueuePage<T>	list(int offset, int limit, QueueListFilter filter)	Returns a paged queue listing ordered by creation time.
boolean	markDead(String uid, String lastError)	Marks an item as dead.
private long	minEpochForState(String column, QueueItemState state)
protected QueueItem<T>	readItem(ResultSet rs)
int	releaseExpiredClaims(long nowEpochSeconds)	Releases expired claims back to READY.
boolean	reschedule(QueueItem<T> item, long nextAttemptAtEpochSeconds, String lastError)	Reschedules a claimed item for a future attempt.
private static void	setNullableString(PreparedStatement statement, int index, String value)
long	size()	Active queue size.
QueueStats	stats()	Aggregate queue statistics.
private void	updateDead(Connection connection, List<QueueMutation<T>> mutations)
private void	updateReschedules(Connection connection, List<QueueMutation<T>> mutations)
private String	validateTableName(String tableName)

Methods inherited from class java.lang.Object

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

Field Details

log

private static final org.apache.logging.log4j.Logger log

tableName

protected final String tableName

jdbcUrl

private final String jdbcUrl

username

private final String username

password

private final String password

maxPoolSize

private final int maxPoolSize

dataSource

private com.zaxxer.hikari.HikariDataSource dataSource

Constructor Details

SQLQueueDatabase

protected SQLQueueDatabase(SQLQueueDatabase.DBConfig config)

Method Details

getDatabaseType

protected abstract String getDatabaseType()

getCreateTableSQL

protected abstract String getCreateTableSQL()

getCreateIndexSQL

protected List<String> getCreateIndexSQL()

initialize

public void initialize()

Description copied from interface: QueueDatabase

Initialize external resources.

Specified by:

initialize in interface QueueDatabase<T extends Serializable>

enqueue

public QueueItem<T> enqueue(QueueItem<T> item)

Description copied from interface: QueueDatabase

Inserts a ready queue item.

Specified by:

enqueue in interface QueueDatabase<T extends Serializable>

applyMutations

public void applyMutations(QueueMutationBatch<T> batch)

Description copied from interface: QueueDatabase

Applies a batch of dequeue outcomes and derived enqueues atomically when supported.

Specified by:

applyMutations in interface QueueDatabase<T extends Serializable>

claimReady

public List<QueueItem<T>> claimReady(int limit,
 long nowEpochSeconds,
 String consumerId,
 long claimUntilEpochSeconds)

Description copied from interface: QueueDatabase

Claims ready items for the given consumer until the lease expires.

Specified by:

claimReady in interface QueueDatabase<T extends Serializable>

acknowledge

public boolean acknowledge(String uid)

Description copied from interface: QueueDatabase

Acknowledges successful completion and removes the item from the active queue.

Specified by:

acknowledge in interface QueueDatabase<T extends Serializable>

reschedule

public boolean reschedule(QueueItem<T> item,
 long nextAttemptAtEpochSeconds,
 String lastError)

Description copied from interface: QueueDatabase

Reschedules a claimed item for a future attempt.

Specified by:

reschedule in interface QueueDatabase<T extends Serializable>

releaseExpiredClaims

public int releaseExpiredClaims(long nowEpochSeconds)

Description copied from interface: QueueDatabase

Releases expired claims back to READY.

Specified by:

releaseExpiredClaims in interface QueueDatabase<T extends Serializable>

markDead

public boolean markDead(String uid,
 String lastError)

Description copied from interface: QueueDatabase

Marks an item as dead.

Specified by:

markDead in interface QueueDatabase<T extends Serializable>

size

public long size()

Description copied from interface: QueueDatabase

Active queue size.

Specified by:

size in interface QueueDatabase<T extends Serializable>

stats

public QueueStats stats()

Description copied from interface: QueueDatabase

Aggregate queue statistics.

Specified by:

stats in interface QueueDatabase<T extends Serializable>

list

public QueuePage<T> list(int offset,
 int limit,
 QueueListFilter filter)

Description copied from interface: QueueDatabase

Returns a paged queue listing ordered by creation time.

Specified by:

list in interface QueueDatabase<T extends Serializable>

getByUID

public QueueItem<T> getByUID(String uid)

Description copied from interface: QueueDatabase

Returns the item by queue UID.

Specified by:

getByUID in interface QueueDatabase<T extends Serializable>

deleteByUID

public boolean deleteByUID(String uid)

Description copied from interface: QueueDatabase

Removes the item by queue UID regardless of state.

Specified by:

deleteByUID in interface QueueDatabase<T extends Serializable>

deleteByUIDs

public int deleteByUIDs(List<String> uids)

Description copied from interface: QueueDatabase

Removes multiple items by queue UID regardless of state.

Specified by:

deleteByUIDs in interface QueueDatabase<T extends Serializable>

clear

public void clear()

Description copied from interface: QueueDatabase

Clears the full queue state, including dead items.

Specified by:

clear in interface QueueDatabase<T extends Serializable>

close

public void close()

Specified by:

close in interface AutoCloseable

Specified by:

close in interface Closeable

claimReadyLegacy

private List<QueueItem<T>> claimReadyLegacy(int limit,
 long nowEpochSeconds,
 String consumerId,
 long claimUntilEpochSeconds)

insertSql

private String insertSql()

deleteBatch

private void deleteBatch(Connection connection,
 List<String> ackUids)
                  throws SQLException

Throws:

SQLException

updateReschedules

private void updateReschedules(Connection connection,
 List<QueueMutation<T>> mutations)
                        throws SQLException

Throws:

SQLException

updateDead

private void updateDead(Connection connection,
 List<QueueMutation<T>> mutations)
                 throws SQLException

Throws:

SQLException

insertNewItems

private void insertNewItems(Connection connection,
 List<T> newItems)
                     throws SQLException

Throws:

SQLException

bindItem

private void bindItem(PreparedStatement statement,
 QueueItem<T> item)
               throws SQLException

Throws:

SQLException

readItem

protected QueueItem<T> readItem(ResultSet rs)
                         throws SQLException

Throws:

SQLException

fetchByUIDs

private List<QueueItem<T>> fetchByUIDs(Connection connection,
 List<String> uids)
                                throws SQLException

Throws:

SQLException

countByState

private long countByState(QueueItemState state)

minEpochForState

private long minEpochForState(String column,
 QueueItemState state)

buildWhereClause

private String buildWhereClause(QueueListFilter filter)

bindFilter

private int bindFilter(PreparedStatement statement,
 QueueListFilter filter,
 int startIndex)
                throws SQLException

Throws:

SQLException

setNullableString

private static void setNullableString(PreparedStatement statement,
 int index,
 String value)
                               throws SQLException

Throws:

SQLException

validateTableName

private String validateTableName(String tableName)