org.ak.trafficController.messaging.mem

Class GenericInMemoryQueue<T>

java.lang.Object

org.ak.trafficController.messaging.mem.GenericInMemoryQueue<T>

Type Parameters:

T - This defines the type of object the queue will work on.

public class GenericInMemoryQueue<T>
extends java.lang.Object

Each object of this class act as in memory queue. One can attach listener to it. The listener can be of direct retrieval or list retrieval. One can consider them as event handler as well. Whenever any data added to it will be processed by available listener. This works on observer pattern where all listeners are waiting for notification which will be sent when any item is added.

Constructor Summary

Constructors

Constructor and Description
GenericInMemoryQueue(java.lang.String name) Constructor of Generic in memory queue.
GenericInMemoryQueue(java.lang.String name, int batchSize) Constructor which also configures batch size which is used by batch consumers.

Method Summary

Modifier and Type	Method and Description
void	add(java.util.Collection<T> items) This method adds list of items to be processed in queue.
void	add(T item) This method add an item to the queue.
protected void	batchProcess(java.util.function.Consumer<java.util.List<T>> consumer, java.lang.String consumerName) This method gets from the queue and process.
protected java.lang.StringBuilder	buildErrorMessage(java.lang.String consumerName) Build error message using consumer name and queue name.
protected void	clear(java.util.function.Consumer<java.util.Queue<T>>... preCleanHandlers) This method clears the current queue and pass the data to the consumer.
protected void	doNotify() This method Notify all consumers.
int	getBatchConsumerCount() Get batch consumer count.
protected java.util.List<T>	getBatchData() Get batch data which will be used to pass to batch consumer.
int	getBatchSize() Batch size which will be used by batch consumers.
int	getDirectConsumerCount() Get number of direct consumers.
protected java.lang.String	getExceptionMessage(java.lang.String consumerName, java.util.List<T> items) This method builds the exception message for batch consumer.
protected java.lang.String	getExceptionMessage(java.lang.String consumerName, T item) This method prepares the exception message containing name of consumer and queue.
java.lang.Long	getNumberOfItemsInQueue() Get number of items in queue.
java.lang.String	getQueueName() Get queue name.
java.lang.Long	getSleepPostConsumingAll()
java.lang.Long	getSleepPostConsumingEvery()
int	getWaitTimeout() This is the wait time for which different consumers wait for some message.
boolean	isHasShutdown() Have this queue shutdown?
boolean	isShouldSleepAfterEveryConsumption()
boolean	isShouldSleepConsumingAll()
protected void	notifyConsumers() This method notify on addition of an item in the queue.
void	register(java.util.function.Consumer<T> consumer, java.lang.String consumerName) This method register the direct consumer.
void	registerBatchConsumer(java.util.function.Consumer<java.util.List<T>> consumer, java.lang.String consumerName) Register batch consumer
void	removeAllBatchConsumers() Remove all batch consumers.
void	removeAllDirectConsumers() Removes all direct consumers.
GenericInMemoryQueue<T>	setBatchSize(int batchSize) Setter Batch size which will be used by batch consumers.
GenericInMemoryQueue<T>	setNumberOfItemsInQueue(java.util.concurrent.atomic.AtomicLong numberOfItemsInQueue) Set number of items in queue.
GenericInMemoryQueue<T>	setQueueName(java.lang.String queueName) Set name of the queue
GenericInMemoryQueue<T>	setSleepPostConsumingAll(java.lang.Long sleepPostConsumingAll) This method sets sleepPostConsumingAll with the value passed.
GenericInMemoryQueue<T>	setSleepPostConsumingEach(java.lang.Long sleepPostConsumingEvery) This method sets sleepPostConsumingEvery with the value passed.
GenericInMemoryQueue<T>	setWaitTimeout(int waitTimeout) This is the wait time for which different consumers wait for some message.
void	shutdown() To cleanup, shutdown method should be called.
protected void	singleItemProcess(java.util.function.Consumer<T> consumer, java.lang.String consumerName) This method polls from the queue and process the consumer with data found.
protected void	sleepIfConfigured() Sleep post consuming all messages if configured.
protected void	sleepIfConfiguredPostEveryConsumption()
protected void	startConsumer(java.lang.Runnable executionLogic, java.lang.String name) This method starts the consumer.
protected void	startListening(java.lang.Runnable consumerLogic, java.lang.String consumerName, java.util.Collection<java.lang.String> consumerList) This method start listener.
void	unregister(java.lang.String consumerName) This method unregister the consumer.

Methods inherited from class java.lang.Object

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

Constructor Detail

GenericInMemoryQueue

public GenericInMemoryQueue(java.lang.String name)

Constructor of Generic in memory queue.

Parameters:

name - Name of queue

GenericInMemoryQueue

public GenericInMemoryQueue(java.lang.String name,
                            int batchSize)

Constructor which also configures batch size which is used by batch consumers. Batch size set here is the maximum items in list a consumer can expect.

Parameters:

name - Name of queue

batchSize - Batch size

Method Detail

getSleepPostConsumingEvery

public java.lang.Long getSleepPostConsumingEvery()

Returns:

the sleepPostConsumingEvery

setSleepPostConsumingEach

public GenericInMemoryQueue<T> setSleepPostConsumingEach(java.lang.Long sleepPostConsumingEvery)

This method sets sleepPostConsumingEvery with the value passed.

Parameters:

sleepPostConsumingEvery - the sleepPostConsumingEvery to set

Returns:

Self to use further

isShouldSleepConsumingAll

public boolean isShouldSleepConsumingAll()

Returns:

the shouldSleepConsumingAll

isShouldSleepAfterEveryConsumption

public boolean isShouldSleepAfterEveryConsumption()

Returns:

the shouldSleepAfterEveryConsumption

getSleepPostConsumingAll

public java.lang.Long getSleepPostConsumingAll()

Returns:

the sleepPostConsumingAll

setSleepPostConsumingAll

public GenericInMemoryQueue<T> setSleepPostConsumingAll(java.lang.Long sleepPostConsumingAll)

This method sets sleepPostConsumingAll with the value passed.

Parameters:

sleepPostConsumingAll - the sleepPostConsumingAll to set

Returns:

Self to use further

register

public void register(java.util.function.Consumer<T> consumer,
                     java.lang.String consumerName)

This method register the direct consumer.

Parameters:

consumer - The instance of Consumer to set

consumerName - The consumer name.

singleItemProcess

protected void singleItemProcess(java.util.function.Consumer<T> consumer,
                                 java.lang.String consumerName)

This method polls from the queue and process the consumer with data found. The method keeps control till there is any data in queue.

Parameters:

consumer - The Consumer instance to set

consumerName - The String consumer name to set

sleepIfConfiguredPostEveryConsumption

protected void sleepIfConfiguredPostEveryConsumption()

sleepIfConfigured

protected void sleepIfConfigured()

Sleep post consuming all messages if configured.

getExceptionMessage

protected java.lang.String getExceptionMessage(java.lang.String consumerName,
                                               T item)

This method prepares the exception message containing name of consumer and queue.

Parameters:

consumerName - The String consumer name

item - The item of T type to set

Returns:

The String exception message

getExceptionMessage

protected java.lang.String getExceptionMessage(java.lang.String consumerName,
                                               java.util.List<T> items)

This method builds the exception message for batch consumer.

Parameters:

consumerName - the consumerName

items - the list of items

Returns:

String append message

buildErrorMessage

protected java.lang.StringBuilder buildErrorMessage(java.lang.String consumerName)

Build error message using consumer name and queue name.

Parameters:

consumerName - Consumer name

Returns:

Message built in string builder

registerBatchConsumer

public void registerBatchConsumer(java.util.function.Consumer<java.util.List<T>> consumer,
                                  java.lang.String consumerName)

Register batch consumer

Parameters:

consumer - the consumer

consumerName - the consumer name

startListening

protected void startListening(java.lang.Runnable consumerLogic,
                              java.lang.String consumerName,
                              java.util.Collection<java.lang.String> consumerList)

This method start listener. It creates a thread which will perform the operation required on the message in queue.

Parameters:

consumerLogic - The instance of Runnable to set

consumerName - The String consumer name to set

consumerList - consumersList direct or batch

startConsumer

protected void startConsumer(java.lang.Runnable executionLogic,
                             java.lang.String name)

This method starts the consumer. This method will decide whether to use any executor or create new thread. Currently it looks like creating thread is better than managing executors as these threads will keep on running.

Parameters:

executionLogic - Execution logic

name - Name of consumer which will be set as name.

batchProcess

protected void batchProcess(java.util.function.Consumer<java.util.List<T>> consumer,
                            java.lang.String consumerName)

This method gets from the queue and process.

Parameters:

consumer - The Consumer instance to set

consumerName - The String consumer name to set

getBatchData

protected java.util.List<T> getBatchData()

Get batch data which will be used to pass to batch consumer. This method tries to get maximum "batchsize" messages. It will not wait if the number of messages is lesser.

Returns:

List containing data

unregister

public void unregister(java.lang.String consumerName)

This method unregister the consumer.

Parameters:

consumerName - The String consumer name to remove

doNotify

protected void doNotify()

This method Notify all consumers. This is done whenever a new message arrives or a consumer is asked to stop.

shutdown

public void shutdown()

To cleanup, shutdown method should be called. This method shutdown all consumers batch or direct.

add

public void add(T item)

This method add an item to the queue.

Parameters:

item - The item to add

notifyConsumers

protected void notifyConsumers()

This method notify on addition of an item in the queue.

add

public void add(java.util.Collection<T> items)

This method adds list of items to be processed in queue.

Parameters:

items - The List of items to add

getDirectConsumerCount

public int getDirectConsumerCount()

Get number of direct consumers.

Returns:

Direct consumers count

getBatchConsumerCount

public int getBatchConsumerCount()

Get batch consumer count.

Returns:

Batch consumer count

isHasShutdown

public boolean isHasShutdown()

Have this queue shutdown?

Returns:

Have this queue shutdown?

getQueueName

public java.lang.String getQueueName()

Get queue name.

Returns:

the queueName

setQueueName

public GenericInMemoryQueue<T> setQueueName(java.lang.String queueName)

Set name of the queue

Parameters:

queueName - the queueName to set

Returns:

This object for further use

getBatchSize

public int getBatchSize()

Batch size which will be used by batch consumers. Any batch consumer will receive maximum of this many data items in list. The batch consumer might get fewer amount as well as consumers are called with only current data in queue and not waited for this list to be filled

Returns:

the batchSize

setBatchSize

public GenericInMemoryQueue<T> setBatchSize(int batchSize)

Setter Batch size which will be used by batch consumers. Any batch consumer will receive maximum of this many data items in list. The batch consumer might get fewer amount as well as consumers are called with only current data in queue and not waited for this list to be filled

Parameters:

batchSize - the batchSize to set

Returns:

This object for further use

getNumberOfItemsInQueue

public java.lang.Long getNumberOfItemsInQueue()

Get number of items in queue. This method is used for diagnostics or by some tuner which want to resize number of consumers.

Returns:

the numberOfItemsInQueue

setNumberOfItemsInQueue

public GenericInMemoryQueue<T> setNumberOfItemsInQueue(java.util.concurrent.atomic.AtomicLong numberOfItemsInQueue)

Set number of items in queue. This method is used for diagnostics or by some tuner which want to resize number of consumers.

Parameters:

numberOfItemsInQueue - the numberOfItemsInQueue to set

Returns:

This object for further use

clear

protected void clear(java.util.function.Consumer<java.util.Queue<T>>... preCleanHandlers)

This method clears the current queue and pass the data to the consumer. Post this data is consumed by cleaner, queue is cleaned. This method is used by scenarios where the messages are allowed to be wiped off to save the health of the system. This method is called by some tuner.

Parameters:

preCleanHandlers - Cleaners

removeAllDirectConsumers

public void removeAllDirectConsumers()

Removes all direct consumers. This method is called either at shutdown or we want to move to different logic of processing the data.

removeAllBatchConsumers

public void removeAllBatchConsumers()

Remove all batch consumers. This method is called either at shutdown or we want to move to different logic of processing the data.

getWaitTimeout

public int getWaitTimeout()

This is the wait time for which different consumers wait for some message. Post wait, the consumers retry finding data and if not found go to wait again. In rare case if some notification is lost, this mechanism guarantees that messages will be picked if present post this time.

Returns:

Wait time

setWaitTimeout

public GenericInMemoryQueue<T> setWaitTimeout(int waitTimeout)

This is the wait time for which different consumers wait for some message. Post wait, the consumers retry finding data and if not found go to wait again. In rare case if some notification is lost, this mechanism guarantees that messages will be picked if present post this time.

Parameters:

waitTimeout - wait time out to be set

Returns:

This object to further use