com.github.ddth.queue.impl.universal

Class UniversalJdbcQueue

java.lang.Object

com.github.ddth.dao.BaseDao

com.github.ddth.dao.jdbc.BaseJdbcDao

com.github.ddth.queue.impl.JdbcQueue

com.github.ddth.queue.impl.universal.UniversalJdbcQueue

All Implemented Interfaces:

IQueue, Closeable, AutoCloseable

public class UniversalJdbcQueue
extends JdbcQueue

Universal JDBC implementation of IQueue.

Queue and Take UniversalQueueMessages.

ephemeralDisabled: when set to true, ephemeral storage is disabled. Default value is false

fifo: when set to true (which is default) messages are taken in FIFO manner.

Queue db table schema:

• queue_id: bigint, auto increment, see IQueueMessage.qId(), COL_QUEUE_ID
• msg_org_timestamp: datetime, see IQueueMessage.qOriginalTimestamp(), COL_ORG_TIMESTAMP
• msg_timestamp: datetime, see IQueueMessage.qTimestamp(), COL_TIMESTAMP
• msg_num_requeues: int, see IQueueMessage.qNumRequeues(), COL_NUM_REQUEUES
• msg_content: blob, message's content, see COL_CONTENT

Ephemeral db table schema:

• queue_id: bigint, see IQueueMessage.qId(), COL_QUEUE_ID
• msg_org_timestamp: datetime, see IQueueMessage.qOriginalTimestamp(), COL_ORG_TIMESTAMP
• msg_timestamp: datetime, see IQueueMessage.qTimestamp(), COL_TIMESTAMP
• msg_num_requeues: int, see IQueueMessage.qNumRequeues(), COL_NUM_REQUEUES
• msg_content: blob, message's content, see COL_CONTENT

Since:

0.2.3

Author:

Thanh Ba Nguyen

Field Summary

Fields

Modifier and Type	Field and Description
static String	COL_CONTENT Table's column name to store message's content
static String	COL_NUM_REQUEUES Table's column name to store message's number of requeues
static String	COL_ORG_TIMESTAMP Table's column name to store message's original timestamp
static String	COL_QUEUE_ID Table's column name to store queue-id
static String	COL_TIMESTAMP Table's column name to store message's timestamp

Fields inherited from class com.github.ddth.queue.impl.JdbcQueue

DEFAULT_MAX_RETRIES

Fields inherited from class com.github.ddth.dao.BaseDao

CHARSET

Constructor Summary

Constructors

Constructor and Description
UniversalJdbcQueue()

Method Summary

Modifier and Type	Method and Description
boolean	getEphemeralDisabled()
boolean	getFifo() If true, queue message with lower id is ensured to be taken first.
protected Collection<IQueueMessage>	getOrphanFromEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, long thresholdTimestampMs) Gets all orphan messages (messages that were left in ephemeral storage for a long time).
UniversalJdbcQueue	init()
boolean	isEphemeralDisabled()
boolean	isFifo() If true, queue message with lower id is ensured to be taken first.
UniversalJdbcQueue	markEphemeralDisabled(boolean setEphemeralDisabled)
UniversalJdbcQueue	markFifo(boolean fifo) When set to true, queue message with lower id is ensured to be taken first.
protected boolean	putToEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage _msg) Puts a message to the ephemeral storage.
protected boolean	putToQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage _msg) Puts a message to tail of the queue storage.
protected UniversalQueueMessage	readFromEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage msg) Reads a message from the ephemeral storage.
protected UniversalQueueMessage	readFromQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate) Reads a message from head of queue storage.
protected boolean	removeFromEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage _msg) Removes a message from the ephemeral storage.
protected boolean	removeFromQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage _msg) Removes a message from the queue storage.
UniversalJdbcQueue	setEphemeralDisabled(boolean ephemeralDisabled)
UniversalJdbcQueue	setFifo(boolean fifo) When set to true, queue message with lower id is ensured to be taken first.
UniversalQueueMessage	take() Takes a message out of queue.

Methods inherited from class com.github.ddth.queue.impl.JdbcQueue

_finishWithRetries, _getOrphanMessagesWithRetries, _moveFromEphemeralToQueueStorageWithRetries, _queueWithRetries, _requeueSilentWithRetries, _requeueWithRetries, _takeWithRetries, close, destroy, ephemeralSize, finish, getMaxRetries, getOrphanMessages, getTableName, getTableNameEphemeral, getTransactionIsolationLevel, moveFromEphemeralToQueueStorage, queue, queueSize, requeue, requeueSilent, setMaxRetries, setTableName, setTableNameEphemeral, setTransactionIsolationLevel

Methods inherited from class com.github.ddth.dao.jdbc.BaseJdbcDao

commitTransaction, connection, connection, execute, execute, execute, executeSelect, executeSelect, executeSelect, executeSelect, executeSelect, executeSelect, getDataSource, jdbcTemplate, jdbcTemplate, returnConnection, rollbackTransaction, setDataSource, startTransaction

Methods inherited from class com.github.ddth.dao.BaseDao

addProfiling, clearProfiling, getCache, getCacheFactory, getFromCache, getFromCache, getProfiling, isCacheEnabled, isCacheItemsExpireAfterWrite, putToCache, putToCache, putToCache, removeFromCache, setCacheFactory, setCacheItemsExpireAfterWrite, startProfiling

Methods inherited from class java.lang.Object

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

Field Detail

COL_QUEUE_ID

public static final String COL_QUEUE_ID

Table's column name to store queue-id

See Also:

Constant Field Values

COL_ORG_TIMESTAMP

public static final String COL_ORG_TIMESTAMP

Table's column name to store message's original timestamp

See Also:

Constant Field Values

COL_TIMESTAMP

public static final String COL_TIMESTAMP

Table's column name to store message's timestamp

See Also:

Constant Field Values

COL_NUM_REQUEUES

public static final String COL_NUM_REQUEUES

Table's column name to store message's number of requeues

See Also:

Constant Field Values

COL_CONTENT

public static final String COL_CONTENT

Table's column name to store message's content

See Also:

Constant Field Values

Constructor Detail

UniversalJdbcQueue

public UniversalJdbcQueue()

Method Detail

setFifo

public UniversalJdbcQueue setFifo(boolean fifo)

When set to true, queue message with lower id is ensured to be taken first. When set to false, order of taken queue messages depends on the DBMS (usually FIFO in most cases).

Parameters:

fifo -

Returns:

markFifo

public UniversalJdbcQueue markFifo(boolean fifo)

When set to true, queue message with lower id is ensured to be taken first. When set to false, order of taken queue messages depends on the DBMS (usually FIFO in most cases).

Parameters:

fifo -

Returns:

isFifo

public boolean isFifo()

If true, queue message with lower id is ensured to be taken first. Otherwise, order of taken queue messages depends on the DBMS (usually FIFO in most cases).

Returns:

getFifo

public boolean getFifo()

If true, queue message with lower id is ensured to be taken first. Otherwise, order of taken queue messages depends on the DBMS (usually FIFO in most cases).

Returns:

setEphemeralDisabled

public UniversalJdbcQueue setEphemeralDisabled(boolean ephemeralDisabled)

markEphemeralDisabled

public UniversalJdbcQueue markEphemeralDisabled(boolean setEphemeralDisabled)

isEphemeralDisabled

public boolean isEphemeralDisabled()

getEphemeralDisabled

public boolean getEphemeralDisabled()

init

public UniversalJdbcQueue init()

Description copied from class: JdbcQueue

Overrides:

init in class JdbcQueue

readFromQueueStorage

protected UniversalQueueMessage readFromQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate)

Reads a message from head of queue storage.

Specified by:

readFromQueueStorage in class JdbcQueue

Returns:

readFromEphemeralStorage

protected UniversalQueueMessage readFromEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate,
                                                         IQueueMessage msg)

Reads a message from the ephemeral storage.

Specified by:

readFromEphemeralStorage in class JdbcQueue

Returns:

getOrphanFromEphemeralStorage

protected Collection<IQueueMessage> getOrphanFromEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate,
                                                                  long thresholdTimestampMs)

Gets all orphan messages (messages that were left in ephemeral storage for a long time).

Specified by:

getOrphanFromEphemeralStorage in class JdbcQueue

thresholdTimestampMs - get all orphan messages that were queued before this timestamp

Returns:

putToQueueStorage

protected boolean putToQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate,
                                    IQueueMessage _msg)

Puts a message to tail of the queue storage.

Specified by:

putToQueueStorage in class JdbcQueue

Returns:

putToEphemeralStorage

protected boolean putToEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate,
                                        IQueueMessage _msg)

Puts a message to the ephemeral storage.

Specified by:

putToEphemeralStorage in class JdbcQueue

Returns:

removeFromQueueStorage

protected boolean removeFromQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate,
                                         IQueueMessage _msg)

Removes a message from the queue storage.

Specified by:

removeFromQueueStorage in class JdbcQueue

Returns:

removeFromEphemeralStorage

protected boolean removeFromEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate,
                                             IQueueMessage _msg)

Removes a message from the ephemeral storage.

Specified by:

removeFromEphemeralStorage in class JdbcQueue

Returns:

take

public UniversalQueueMessage take()

Takes a message out of queue.

Implementation flow:

• Read message from head of queue storage.
• Write message to ephemeral storage.
• Remove message from queue storage.

Note: ephemeral storage implementation is optional, depends on implementation.

Specified by:

take in interface IQueue

Overrides:

take in class JdbcQueue

Returns: