I wrote this little piece of code, a while back. The intention behind it was to create a system to send messages between consumer and producer threads. I have no idea for what i wanted to use it. I just found it in an old snippet folder of mine. The type of the queue is FIFO.
The usage
Create a queue with new MessageQueue<MessageType>()
.
let's assume we will store an instance of the message queue in an private field inside the classes that will be using it.
This field is called messageQueue
- Use
messageQueue.push(new MessageType([...]))
in a producer thread to add a Message. - Take it out of the queue from a consumer thread with
messageQueue.pull()
. - Check if there is a message in the queue with
messageQueue.hasMessage()
.
I know that I thought about implementing a messageQueue.peek()
but I never implementet it.
The code
/**
* Holds a queue of messages for inter-thread communication
* Producer and Consumer should <b>not</b> be the same thread!
*
* @param <T> type of the messages
*/
public class MessageQueue<T> {
private static final int MAX_SIZE = 255;
private T[] messages;
private int lastReadIndex;
private int lastInsertIndex;
private int adjustIndex(int index) {
if (index >= MAX_SIZE) {
return 0;
}
return index;
}
/**
* Constructor of the {@link MessageQueue}
*/
@SuppressWarnings("unchecked") // Java can't create an Array of a generic type...
public MessageQueue() {
messages = (T[]) new Object[MAX_SIZE];
}
/**
* Pushes a new entry into the queue.
* This function will block when the queue is full.
*
* @param msg The entry to push on the stack
* @throws InterruptedException
*/
public synchronized void push(T msg) throws InterruptedException {
int newIndex = adjustIndex(lastInsertIndex + 1);
// Check if we can push the message
while (messages[newIndex] != null) {
wait();
}
// Push the message
messages[newIndex] = msg;
lastInsertIndex = newIndex;
notifyAll();
}
/**
* Pulls an entry from the queue.
* The entry will be returned and removed from the queue.
*
* @return The entry from the queue
*/
public synchronized T pull() {
int newIndex = adjustIndex(lastReadIndex + 1);
T shouldReturn = messages[newIndex];
messages[newIndex] = null;
lastReadIndex = newIndex;
notifyAll();
return shouldReturn;
}
/**
* Checks if there is an entry in the queue.
*
* @return <code>true</code>, when there is an entry in the queue
*/
public synchronized Boolean hasMessage() {
if (lastInsertIndex > -1) {
int newIndex = adjustIndex(lastReadIndex + 1);
if (messages[newIndex] != null) {
return true;
}
}
return false;
}
}