com.github.ddth.queue.impl

Class UniversalRedisQueue

java.lang.Object

com.github.ddth.queue.impl.UniversalRedisQueue

All Implemented Interfaces:

IQueue

public class UniversalRedisQueue
extends Object
implements IQueue

Universal Redis implementation of IQueue.

Queue and Take UniversalQueueMessages.

Implementation:

• A hash to store message's content, format {queue_id => message}. See setRedisHashName(String).
• A list to act as a queue of message's queue_id. See setRedisListName(String).
• A sorted set to act as ephemeral storage of message's queue_id, store is message's timestamp. See setRedisSortedSetName(String).

Since:

0.3.0

Author:

Thanh Ba Nguyen

Constructor Summary

Constructors

Constructor and Description
UniversalRedisQueue()

Method Summary

Methods

Modifier and Type	Method and Description
void	destroy() Destroy method.
int	ephemeralSize() Gets ephemeral-storage's number of items.
void	finish(IQueueMessage _msg) Called when finish processing the message to cleanup ephemeral storage.
Collection<IQueueMessage>	getOrphanMessages(long thresholdTimestampMs) Gets all orphan messages (messages that were left in ephemeral storage for a long time).
String	getRedisHashName()
String	getRedisHost()
String	getRedisListName()
int	getRedisPort()
String	getRedisSortedSetName()
UniversalRedisQueue	init() Init method.
boolean	moveFromEphemeralToQueueStorage(IQueueMessage _msg) Moves a message from ephemeral back to queue storage.
boolean	queue(IQueueMessage _msg) Queues a message.
int	queueSize() Gets queue's number of items.
boolean	requeue(IQueueMessage _msg) Re-queues a message.
boolean	requeueSilent(IQueueMessage _msg) Silently re-queues a message.
UniversalRedisQueue	setRedisHashName(String redisHashName)
UniversalRedisQueue	setRedisHost(String redisHost)
UniversalRedisQueue	setRedisListName(String redisListName)
UniversalRedisQueue	setRedisPort(int redisPort)
UniversalRedisQueue	setRedisSortedSetName(String redisSortedSetName)
UniversalQueueMessage	take() Takes a message out of queue.

Methods inherited from class java.lang.Object

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

Constructor Detail

UniversalRedisQueue

public UniversalRedisQueue()

Method Detail

getRedisHost

public String getRedisHost()

setRedisHost

public UniversalRedisQueue setRedisHost(String redisHost)

getRedisPort

public int getRedisPort()

setRedisPort

public UniversalRedisQueue setRedisPort(int redisPort)

getRedisHashName

public String getRedisHashName()

setRedisHashName

public UniversalRedisQueue setRedisHashName(String redisHashName)

getRedisListName

public String getRedisListName()

setRedisListName

public UniversalRedisQueue setRedisListName(String redisListName)

getRedisSortedSetName

public String getRedisSortedSetName()

setRedisSortedSetName

public UniversalRedisQueue setRedisSortedSetName(String redisSortedSetName)

init

public UniversalRedisQueue init()

Init method.

Returns:

destroy

public void destroy()

Destroy method.

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:

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 count & update message's queue timestamp.
• Remove message from ephemeral storage.

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

Specified by:

requeue in interface IQueue

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 and do NOT 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:

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

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

Returns:

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 -

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.
• Put the message back to queue.
• Remove the message 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: