com.viaoa.util

Class OACircularQueue<TYPE>

java.lang.Object

com.viaoa.util.OACircularQueue<TYPE>

public abstract class OACircularQueue<TYPE>
extends Object

Thread safe Circular Queue. There is a single headPostion, and each consumer can have a tailPosition. Client access (getMessageXxx methods) will check overruns, and will throw an exception if a queue overrun occurs. A session id can be used so that entries can be removed from the queue once all sessions have received it.

Author:

vvia Note: this is made abstract to be able to get the Generic class that is used.

Field Summary

Fields

Modifier and Type	Field and Description
int	msWaitUntilNotified

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	OACircularQueue()
	OACircularQueue(Class clazz, int queueSize)
	OACircularQueue(int queueSize) Create a new circular queue.

Method Summary

Modifier and Type	Method and Description
int	addMessage(TYPE msg)
int	addMessageToQueue(TYPE msg) Add a new message to the queue.
int	addMessageToQueue(TYPE msg, int throttleAmount) param bThrottle if true, then make sure that headPos is not too far ahead of readers
int	addMessageToQueue(TYPE msg, int throttleAmount, int throttleSessionToIgnore)
protected void	cleanupQueue()
int	getAmountAvailable(long posTail)
long	getHeadPostion() current position where the next message will be added.
TYPE	getMessage(long posTail) will block until a message is available.
TYPE	getMessage(long posTail, int msMaxWait) Get next message, with a timeout
TYPE[]	getMessages(int sessionId, long posTail, int maxReturnAmount, int msMaxWait)
TYPE[]	getMessages(long posTail) will block until at least one message is available.
TYPE[]	getMessages(long posTail, int maxReturnAmount) will block until at least one message is available.
TYPE[]	getMessages(long posTail, int maxReturnAmount, int msMaxWait)
TYPE	getMessagesAtPos(int pos) Get message at actual position in queue
String	getName()
int	getSize()
void	keepAlive(int sessionId)
long	registerSession(int sessionId) This is used to let the queue know who the consumers are, so that queue slots can be set to null once they are not needed.
void	setName(String s)
void	setSize(int queueSize) set the size of the array behind the circular queue
protected boolean	shouldWaitOnSlowSession(int sessionId, int msSinceLastRead) This is call whenever a session is getting close to a queue overrun and it has not called get for 1+ seconds.
void	unregisterSession(int sessionId)

Methods inherited from class java.lang.Object

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

Field Detail

msWaitUntilNotified

public final int msWaitUntilNotified

See Also:

Constant Field Values

Constructor Detail

OACircularQueue

public OACircularQueue(int queueSize)

Create a new circular queue.

Parameters:

queueSize - actual size of the array that backs the queue.

OACircularQueue

public OACircularQueue(Class clazz,
                       int queueSize)

OACircularQueue

protected OACircularQueue()

Method Detail

setSize

public void setSize(int queueSize)

set the size of the array behind the circular queue

getSize

public int getSize()

registerSession

public long registerSession(int sessionId)

This is used to let the queue know who the consumers are, so that queue slots can be set to null once they are not needed. It is also used so that queue writers dont go too fast and overrun the readers.

Parameters:

sessionId - identifier for the session param maxFallBehind max amount that it can fall behind the head, else an addMessage will wait for up to 1 second.

Returns:

current queueHeadPosition

unregisterSession

public void unregisterSession(int sessionId)

cleanupQueue

protected void cleanupQueue()

getHeadPostion

public long getHeadPostion()

current position where the next message will be added.

addMessageToQueue

public int addMessageToQueue(TYPE msg)

Add a new message to the queue.

Returns:

position of message in the queue

addMessage

public int addMessage(TYPE msg)

addMessageToQueue

public int addMessageToQueue(TYPE msg,
                             int throttleAmount)

param bThrottle if true, then make sure that headPos is not too far ahead of readers

addMessageToQueue

public int addMessageToQueue(TYPE msg,
                             int throttleAmount,
                             int throttleSessionToIgnore)

shouldWaitOnSlowSession

protected boolean shouldWaitOnSlowSession(int sessionId,
                                          int msSinceLastRead)

This is call whenever a session is getting close to a queue overrun and it has not called get for 1+ seconds.

Parameters:

sessionId -

Returns:

default is false

getMessage

public TYPE getMessage(long posTail)
                throws Exception

will block until a message is available.

Throws:

Exception

getMessage

public TYPE getMessage(long posTail,
                       int msMaxWait)
                throws Exception

Get next message, with a timeout

Parameters:

posTail - current position to pull messages from

msMaxWait - max number of milliseconds to wait

Throws:

Exception

getAmountAvailable

public int getAmountAvailable(long posTail)
                       throws Exception

Throws:

Exception

getMessages

public TYPE[] getMessages(long posTail)
                   throws Exception

will block until at least one message is available.

Throws:

Exception

getMessages

public TYPE[] getMessages(long posTail,
                          int maxReturnAmount)
                   throws Exception

will block until at least one message is available.

Throws:

Exception

getMessages

public TYPE[] getMessages(long posTail,
                          int maxReturnAmount,
                          int msMaxWait)
                   throws Exception

Parameters:

posTail - current position to use to get next message

maxReturnAmount -

msMaxWait - if no messages are available, wait this amount of miliseconds for an available message.

Throws:

Exception

getMessages

public TYPE[] getMessages(int sessionId,
                          long posTail,
                          int maxReturnAmount,
                          int msMaxWait)
                   throws Exception

Throws:

Exception

keepAlive

public void keepAlive(int sessionId)

getMessagesAtPos

public TYPE getMessagesAtPos(int pos)

Get message at actual position in queue

Parameters:

pos - is actual array position, must be less then queSize, else null is returned.

Returns:

setName

public void setName(String s)

getName

public String getName()