org.ak.trafficController.messaging.mem

Class InMemoryQueue<T>

java.lang.Object

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

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

This class can be considered as extension of GenericInMemoryQueue. This class is suitable for majority of use cases. Using this class means we are saying only one consumer (with multiple instance) is consuming (each for direct and batch). GenericInMemoryQueue can have different type of consumers working on same data which means as queue logic, some data will move to one consumer and other data will move to different consumer. This use case is very rare, thus this new class tries to hnadle majority of normal use case. As we have only one consumer of each type, we can think of adding or decreasing number of consumers easily.

Constructor Summary

Constructors

Constructor and Description
InMemoryQueue(java.lang.String queueName) Creates a new in memory queue.

Method Summary

Modifier and Type	Method and Description
void	add(T i) Add item to the queue.
void	addAllFromCollection(java.util.Collection<T> input) This method adds data present in input collection
protected void	clear(java.util.function.Consumer<java.util.Queue<T>>... consumer) This method clears the queue.
void	decrementBatchConsumer() Decrement batch consumer count.
void	decrementDirectConsumer() Decrement direct consumer count.
java.lang.Integer	getBatchConsumerCount() Get current batch consumer count
protected java.lang.String	getBatchConsumerName(java.lang.Integer i) Name of batch consumer.
java.lang.Integer	getDirectConsumerCount() Get direct consumer count.
protected java.lang.String	getDirectConsumerName(java.lang.Integer i) Name of direct consumer.
java.lang.Long	getNumberOfItemsInQueue() Get current number of items in queue.
java.lang.Long	getSleepTimePostConsumingAllMessage()
java.lang.Long	getSleepTimePostEachConsumption()
void	incrementBatchConsumer() Increment batch consumer count.
void	incrementDirectConsumer() Increment direct consumer count.
boolean	isBatchConsumerSet() This method lets us know that for this queue, any batch consumer set.
boolean	isDirectConsumerSet() This method lets us know that for this queue, any direct consumer set.
protected void	processForCounter(int start, int end, java.util.function.Consumer<java.lang.Integer> consumer) Process for add/remove a consumer from given start to given end index.
void	removeBatchConsumer() Remove Batch Consumer.
void	removeDirectConsumer() Remove direct consumer.
protected void	resetConsumer(java.util.function.Consumer<java.lang.Integer> unregisterConsumer, java.util.function.Consumer<java.lang.Integer> registerConsumer, int consumerCount, java.util.concurrent.atomic.AtomicInteger index) Whenever a consumer needs to be changed, this method is called to remove current consumers of queue and then add fresh consumers.
InMemoryQueue<T>	setBatchConsumer(java.util.function.Consumer<java.util.List<T>> batchConsumer) Set batch consumer logic
InMemoryQueue<T>	setBatchConsumerCount(int batchConsumerCount) Reset batch consumer count.
InMemoryQueue<T>	setBatchSize(int batchSize) Set batch size of the queue.
protected void	setConsumerCount(int currentCount, int newCount, java.util.function.Consumer<java.lang.Integer> registerCounsumer, java.util.function.Consumer<java.lang.Integer> unregisterConsumer) This method is responsible for increasing/decreasing consumer count.
InMemoryQueue<T>	setDirectConsumer(java.util.function.Consumer<T> consumer) Set direct consumer which will work on the data present in queue.
InMemoryQueue<T>	setDirectConsumerCount(int directConsumerCount) Set direct consumer count.
InMemoryQueue<T>	setSleepTimePostConsumingAllMessage(java.lang.Long sleepTimePostConsumingAllMessage) This method sets sleepTimePostConsumingAllMessage with the value passed.
InMemoryQueue<T>	setSleepTimePostEachConsumption(java.lang.Long sleepTimePostEachConsumption) This method sets sleepTimePostEachConsumption with the value passed.
void	shutdown() Shutdown the queue.

Methods inherited from class java.lang.Object

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

Constructor Detail

InMemoryQueue

public InMemoryQueue(java.lang.String queueName)

Creates a new in memory queue.

Parameters:

queueName - Name of queue.

Method Detail

getSleepTimePostEachConsumption

public java.lang.Long getSleepTimePostEachConsumption()

Returns:

the sleepTimePostEachConsumption

setSleepTimePostEachConsumption

public InMemoryQueue<T> setSleepTimePostEachConsumption(java.lang.Long sleepTimePostEachConsumption)

This method sets sleepTimePostEachConsumption with the value passed.

Parameters:

sleepTimePostEachConsumption - the sleepTimePostEachConsumption to set

Returns:

Self to use further

getBatchConsumerName

protected java.lang.String getBatchConsumerName(java.lang.Integer i)

Name of batch consumer.

Parameters:

i - Batch number id

Returns:

Name of batch consumer

getDirectConsumerName

protected java.lang.String getDirectConsumerName(java.lang.Integer i)

Name of direct consumer.

Parameters:

i - Direct consumer id

Returns:

Name of direct consumer

setBatchSize

public InMemoryQueue<T> setBatchSize(int batchSize)

Set batch size of the queue.

Parameters:

batchSize - Batch size

Returns:

InMemoryQueue for ease of use

processForCounter

protected void processForCounter(int start,
                                 int end,
                                 java.util.function.Consumer<java.lang.Integer> consumer)

Process for add/remove a consumer from given start to given end index.

Parameters:

start - Start index from which we have to perform operation

end - End index till which we have to perform operation

consumer - Consumer which will perform actual operation

setDirectConsumer

public InMemoryQueue<T> setDirectConsumer(java.util.function.Consumer<T> consumer)

Set direct consumer which will work on the data present in queue.

Parameters:

consumer - Consumer logic

Returns:

This object for further modifications

resetConsumer

protected void resetConsumer(java.util.function.Consumer<java.lang.Integer> unregisterConsumer,
                             java.util.function.Consumer<java.lang.Integer> registerConsumer,
                             int consumerCount,
                             java.util.concurrent.atomic.AtomicInteger index)

Whenever a consumer needs to be changed, this method is called to remove current consumers of queue and then add fresh consumers.

Parameters:

unregisterConsumer - Unregister consumer logic

registerConsumer - Register consumer logic

consumerCount - Count of consumers to maintain

index - Current index

setBatchConsumer

public InMemoryQueue<T> setBatchConsumer(java.util.function.Consumer<java.util.List<T>> batchConsumer)

Set batch consumer logic

Parameters:

batchConsumer - Batch consumer

Returns:

This object for further modification

setConsumerCount

protected void setConsumerCount(int currentCount,
                                int newCount,
                                java.util.function.Consumer<java.lang.Integer> registerCounsumer,
                                java.util.function.Consumer<java.lang.Integer> unregisterConsumer)

This method is responsible for increasing/decreasing consumer count.

Parameters:

currentCount - Current count

newCount - New count which needs to be set

registerCounsumer - Register logic

unregisterConsumer - Unregister logic

setDirectConsumerCount

public InMemoryQueue<T> setDirectConsumerCount(int directConsumerCount)

Set direct consumer count.

Parameters:

directConsumerCount - Direct consumer count

Returns:

This object for further modification

setBatchConsumerCount

public InMemoryQueue<T> setBatchConsumerCount(int batchConsumerCount)

Reset batch consumer count.

Parameters:

batchConsumerCount - batch consumers count

Returns:

Same instance on which we are looking for setting consumers

shutdown

public void shutdown()

Shutdown the queue. All consumers are shutdown. Should be called before last use of queue.

addAllFromCollection

public void addAllFromCollection(java.util.Collection<T> input)

This method adds data present in input collection

Parameters:

input - Collection containing data

add

public void add(T i)

Add item to the queue.

Parameters:

i - Item to be added to queue

getBatchConsumerCount

public java.lang.Integer getBatchConsumerCount()

Get current batch consumer count

Returns:

Batch consumer count

getDirectConsumerCount

public java.lang.Integer getDirectConsumerCount()

Get direct consumer count.

Returns:

Direct consumer count

getNumberOfItemsInQueue

public java.lang.Long getNumberOfItemsInQueue()

Get current number of items in queue.

Returns:

items in queue

incrementDirectConsumer

public void incrementDirectConsumer()

Increment direct consumer count.

decrementDirectConsumer

public void decrementDirectConsumer()

Decrement direct consumer count.

incrementBatchConsumer

public void incrementBatchConsumer()

Increment batch consumer count.

decrementBatchConsumer

public void decrementBatchConsumer()

Decrement batch consumer count.

isDirectConsumerSet

public boolean isDirectConsumerSet()

This method lets us know that for this queue, any direct consumer set.

Returns:

true if already some direct consumer set

isBatchConsumerSet

public boolean isBatchConsumerSet()

This method lets us know that for this queue, any batch consumer set.

Returns:

true if already some batch consumer set

clear

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

This method clears the queue. All data present in queue is removed. The data present in queue is first consumed by passed consumers.

Parameters:

consumer - Consumers consuming data present in queue

removeDirectConsumer

public void removeDirectConsumer()

Remove direct consumer.

removeBatchConsumer

public void removeBatchConsumer()

Remove Batch Consumer.

getSleepTimePostConsumingAllMessage

public java.lang.Long getSleepTimePostConsumingAllMessage()

Returns:

the sleepTimePostConsumingAllMessage

setSleepTimePostConsumingAllMessage

public InMemoryQueue<T> setSleepTimePostConsumingAllMessage(java.lang.Long sleepTimePostConsumingAllMessage)

This method sets sleepTimePostConsumingAllMessage with the value passed.

Parameters:

sleepTimePostConsumingAllMessage - the sleepTimePostConsumingAllMessage to set

Returns:

Self to use further