com.github.ddth.queue.impl

Class JdbcQueue

java.lang.Object

com.github.ddth.dao.BaseDao

com.github.ddth.dao.jdbc.BaseJdbcDao

com.github.ddth.queue.impl.JdbcQueue

All Implemented Interfaces:

IQueue

public abstract class JdbcQueue
extends com.github.ddth.dao.jdbc.BaseJdbcDao
implements IQueue

Abstract JDBC implementation of IQueue.

Implementation:

• Queue storage & Ephemeral storage are 2 database tables, same structure!

Since:

0.1.0

Author:

Thanh Ba Nguyen

Field Summary

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

CHARSET

Constructor Summary

Constructors

Constructor and Description
JdbcQueue()

Method Summary

Methods

Modifier and Type	Method and Description
protected void	_finishWithRetries(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage msg, int numRetries, int maxRetries) Performs "finish" action, retry if deadlock.
protected Collection<IQueueMessage>	_getOrphanMessagesWithRetries(long thresholdTimestampMs, org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, int numRetries, int maxRetries) Gets all orphan messages (messages that were left in ephemeral storage for a long time), retry if deadlock.
protected boolean	_moveFromEphemeralToQueueStorageWithRetries(IQueueMessage msg, org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, int numRetries, int maxRetries) Moves a message from ephemeral back to queue storage, retry if deadlock.
protected boolean	_queueWithRetries(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage msg, int numRetries, int maxRetries) Queues a message, retry if deadlock.
protected boolean	_requeueSilentWithRetries(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage msg, int numRetries, int maxRetries) Re-queues a message silently, retry if deadlock.
protected boolean	_requeueWithRetries(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage msg, int numRetries, int maxRetries) Re-queues a message, retry if deadlock.
protected IQueueMessage	_takeWithRetries(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, int numRetries, int maxRetries) Takes a message from queue, retry if deadlock.
void	destroy()
int	ephemeralSize() Gets ephemeral-storage's number of items.
void	finish(IQueueMessage msg) Called when finish processing the message to cleanup ephemeral storage.
protected abstract 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).
Collection<IQueueMessage>	getOrphanMessages(long thresholdTimestampMs) Gets all orphan messages (messages that were left in ephemeral storage for a long time).
protected String	getTableName()
protected String	getTableNameEphemeral()
JdbcQueue	init()
boolean	moveFromEphemeralToQueueStorage(IQueueMessage msg) Moves a message from ephemeral back to queue storage.
protected abstract boolean	putToEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage msg) Puts a message to the ephemeral storage.
protected abstract boolean	putToQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage msg) Puts a message to tail of the queue storage.
boolean	queue(IQueueMessage msg) Queues a message.
int	queueSize() Gets queue's number of items.
protected abstract IQueueMessage	readFromEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage msg) Reads a message from the ephemeral storage.
protected abstract IQueueMessage	readFromQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate) Reads a message from head of queue storage.
protected abstract boolean	removeFromEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage msg) Removes a message from the ephemeral storage.
protected abstract boolean	removeFromQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage msg) Removes a message from the queue storage.
boolean	requeue(IQueueMessage msg) Re-queues a message.
boolean	requeueSilent(IQueueMessage msg) Silently re-queues a message.
JdbcQueue	setTableName(String tableName)
JdbcQueue	setTableNameEphemeral(String tableNameEphemeral)
IQueueMessage	take() Takes a message out of queue.

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

Constructor Detail

JdbcQueue

public JdbcQueue()

Method Detail

setTableName

public JdbcQueue setTableName(String tableName)

getTableName

protected String getTableName()

setTableNameEphemeral

public JdbcQueue setTableNameEphemeral(String tableNameEphemeral)

getTableNameEphemeral

protected String getTableNameEphemeral()

init

public JdbcQueue init()

Overrides:

init in class com.github.ddth.dao.jdbc.BaseJdbcDao

destroy

public void destroy()

Overrides:

destroy in class com.github.ddth.dao.jdbc.BaseJdbcDao

readFromQueueStorage

protected abstract IQueueMessage readFromQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate)

Reads a message from head of queue storage.

Parameters:

jdbcTemplate -

Returns:

readFromEphemeralStorage

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

Reads a message from the ephemeral storage.

Parameters:

jdbcTemplate -

msg -

Returns:

Since:

0.2.1

getOrphanFromEphemeralStorage

protected abstract 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).

Parameters:

jdbcTemplate -

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

Returns:

Since:

0.2.0

putToQueueStorage

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

Puts a message to tail of the queue storage.

Parameters:

jdbcTemplate -

msg -

Returns:

putToEphemeralStorage

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

Puts a message to the ephemeral storage.

Parameters:

jdbcTemplate -

msg -

Returns:

removeFromQueueStorage

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

Removes a message from the queue storage.

Parameters:

jdbcTemplate -

msg -

Returns:

removeFromEphemeralStorage

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

Removes a message from the ephemeral storage.

Parameters:

jdbcTemplate -

msg -

Returns:

_queueWithRetries

protected boolean _queueWithRetries(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate,
                        IQueueMessage msg,
                        int numRetries,
                        int maxRetries)

Queues a message, retry if deadlock.

Note: http://dev.mysql.com/doc/refman/5.0/en/innodb-deadlocks.html

InnoDB uses automatic row-level locking. You can get deadlocks even in the case of transactions that just insert or delete a single row. That is because these operations are not really "atomic"; they automatically set locks on the (possibly several) index records of the row inserted or deleted.

Parameters:

jdbcTemplate -

msg -

numRetries -

maxRetries -

Returns:

queue

public boolean queue(IQueueMessage msg)

Queues a message.

Implementation flow:

• Put message to tail of queue storage.

Specified by:

queue in interface IQueue

Returns:

_requeueWithRetries

protected boolean _requeueWithRetries(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate,
                          IQueueMessage msg,
                          int numRetries,
                          int maxRetries)

Re-queues a message, retry if deadlock.

Note: http://dev.mysql.com/doc/refman/5.0/en/innodb-deadlocks.html

InnoDB uses automatic row-level locking. You can get deadlocks even in the case of transactions that just insert or delete a single row. That is because these operations are not really "atomic"; they automatically set locks on the (possibly several) index records of the row inserted or deleted.

Parameters:

jdbcTemplate -

msg -

numRetries -

maxRetries -

Returns:

requeue

public boolean requeue(IQueueMessage msg)

Re-queues a message.

Implementation flow:

• Put message to tail of queue storage; and increase message's re-queue & update message's queue timestamp. count.
• Remove message from ephemeral storage.

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

Specified by:

requeue in interface IQueue

Returns:

_requeueSilentWithRetries

protected boolean _requeueSilentWithRetries(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate,
                                IQueueMessage msg,
                                int numRetries,
                                int maxRetries)

Re-queues a message silently, retry if deadlock.

Note: http://dev.mysql.com/doc/refman/5.0/en/innodb-deadlocks.html

InnoDB uses automatic row-level locking. You can get deadlocks even in the case of transactions that just insert or delete a single row. That is because these operations are not really "atomic"; they automatically set locks on the (possibly several) index records of the row inserted or deleted.

Parameters:

jdbcTemplate -

msg -

numRetries -

maxRetries -

Returns:

requeueSilent

public boolean requeueSilent(IQueueMessage msg)

Silently re-queues a message.

Implementation flow:

• Put message to tail of queue storage; do NOT increase message's re-queue count or update message's queue timestamp.
• Remove message from ephemeral storage.

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

Specified by:

requeueSilent in interface IQueue

Returns:

_finishWithRetries

protected void _finishWithRetries(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate,
                      IQueueMessage msg,
                      int numRetries,
                      int maxRetries)

Performs "finish" action, retry if deadlock.

Note: http://dev.mysql.com/doc/refman/5.0/en/innodb-deadlocks.html

InnoDB uses automatic row-level locking. You can get deadlocks even in the case of transactions that just insert or delete a single row. That is because these operations are not really "atomic"; they automatically set locks on the (possibly several) index records of the row inserted or deleted.

Parameters:

jdbcTemplate -

msg -

numRetries -

maxRetries -

finish

public void finish(IQueueMessage msg)

Called when finish processing the message to cleanup ephemeral storage.

Implementation flow:

• Remove message from ephemeral storage.

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

Specified by:

finish in interface IQueue

_takeWithRetries

protected IQueueMessage _takeWithRetries(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate,
                             int numRetries,
                             int maxRetries)

Takes a message from queue, retry if deadlock.

Note: http://dev.mysql.com/doc/refman/5.0/en/innodb-deadlocks.html

InnoDB uses automatic row-level locking. You can get deadlocks even in the case of transactions that just insert or delete a single row. That is because these operations are not really "atomic"; they automatically set locks on the (possibly several) index records of the row inserted or deleted.

Parameters:

jdbcTemplate -

numRetries -

maxRetries -

Returns:

take

public IQueueMessage 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

Returns:

_getOrphanMessagesWithRetries

protected Collection<IQueueMessage> _getOrphanMessagesWithRetries(long thresholdTimestampMs,
                                                      org.springframework.jdbc.core.JdbcTemplate jdbcTemplate,
                                                      int numRetries,
                                                      int maxRetries)

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

Note: http://dev.mysql.com/doc/refman/5.0/en/innodb-deadlocks.html

InnoDB uses automatic row-level locking. You can get deadlocks even in the case of transactions that just insert or delete a single row. That is because these operations are not really "atomic"; they automatically set locks on the (possibly several) index records of the row inserted or deleted.

Parameters:

thresholdTimestampMs -

jdbcTemplate -

numRetries -

maxRetries -

Returns:

Since:

0.2.0

getOrphanMessages

public Collection<IQueueMessage> getOrphanMessages(long thresholdTimestampMs)

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

Specified by:

getOrphanMessages in interface IQueue

Parameters:

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

Returns:

_moveFromEphemeralToQueueStorageWithRetries

protected boolean _moveFromEphemeralToQueueStorageWithRetries(IQueueMessage msg,
                                                  org.springframework.jdbc.core.JdbcTemplate jdbcTemplate,
                                                  int numRetries,
                                                  int maxRetries)

Moves a message from ephemeral back to queue storage, retry if deadlock.

Note: http://dev.mysql.com/doc/refman/5.0/en/innodb-deadlocks.html

InnoDB uses automatic row-level locking. You can get deadlocks even in the case of transactions that just insert or delete a single row. That is because these operations are not really "atomic"; they automatically set locks on the (possibly several) index records of the row inserted or deleted.

Parameters:

msg -

jdbcTemplate -

numRetries -

maxRetries -

Returns:

moveFromEphemeralToQueueStorage

public boolean moveFromEphemeralToQueueStorage(IQueueMessage msg)

Moves a message from ephemeral back to queue storage. Useful when dealing with orphan messages.

Implementation flow:

• Read message from the ephemeral storage.
• Re-queue the message.
• (message is removed from the ephemeral storage).

Specified by:

moveFromEphemeralToQueueStorage in interface IQueue

Returns:

true if a move has been made, false otherwise (e.g. the message didn't exist in ephemeral storage)

queueSize

public int queueSize()

Gets queue's number of items.

Specified by:

queueSize in interface IQueue

Returns:

ephemeralSize

public int ephemeralSize()

Gets ephemeral-storage's number of items.

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

Specified by:

ephemeralSize in interface IQueue

Returns: