SequentialMessageQueues in Java

/**
 * Interface for message queues.
 * 
 * @author Benjamin Erb
 * 
 * @param 
 *            Element type
 */
public interface MessageQueue
{
	/**
	 * Adds element to queue. Blocks until element can be added.
	 * 
	 * @param e
	 *            new element
	 * @throws InterruptedException
	 */
	void push(E e) throws InterruptedException;

	/**
	 * Takes an element from the queue. Blocks until element becomes available.
	 * 
	 * @return taken element
	 * @throws InterruptedException
	 */
	E pop() throws InterruptedException;

	/**
	 * Gets the size of the queue
	 * 
	 * @return
	 */
	int size();

	/**
	 * Checks whether queue is empty or not.
	 * 
	 * @return
	 */
	boolean empty();

}

import java.util.Comparator;

/**
 * An enhanced comparator interface for retrieving preceding and subsequent identifiers.  
 * 
 * @author Benjamin Erb
 *
 * @param  Element type
 * @param  Element identifier
 */
/**
 */
public interface SequentialComparator> extends Comparator
{
	/**
	 * Returns the subsequent identifier of a given entity.
	 * 
	 * @param t
	 *            entity
	 * @return subsequent identifier
	 */
	T getNext(E e);

	/**
	 * Returns the preceding identifier of a given entity.
	 * 
	 * @param t
	 *            entity
	 * @return preceding identifier
	 */
	T getPrevious(E e);

	/**
	 * Compares two entities.
	 * 
	 * @param t1
	 *            entity 1
	 * @param t2
	 *            entity 2
	 * @return a negative integer, zero, or a positive integer as the first
	 *         argument is less than, equal to, or greater than the second.
	 */
	int compare(T t1, T t2);

	/**
	 * Compares an entity and an identifier.
	 * 
	 * @param t1
	 *            entity
	 * @param e2
	 *            identifier
	 * @return a negative integer, zero, or a positive integer as the first
	 *         argument is less than, equal to, or greater than the second.
	 */
	int compare(T t1, E e2);

	/**
	 * Compares an identifier and an entity.
	 * 
	 * @param e1
	 *            identifier
	 * @param t2
	 *            entity
	 * @return a negative integer, zero, or a positive integer as the first
	 *         argument is less than, equal to, or greater than the second.
	 */
	int compare(E e1, T t2);
}

import java.util.concurrent.PriorityBlockingQueue;

/**
 * A message queue for sequential message processing. This queue orders incoming
 * messages entities by their identifiers and outputs entites in-order. This
 * queue supports more than one writing thread. However, it is designed for one
 * reading/consuming thread in order to prevent out-of-order processing by
 * multiple threads.
 * 
 * @author Benjamin Erb
 * 
 * @param 
 *            Message entity
 * @param 
 *            Message identifier
 */
public class SequentialMessageQueue> implements MessageQueue
{
	/**
	 * Lock object for queueing
	 */
	private final Object lock = new Object();

	/**
	 * internal queue
	 */
	private final PriorityBlockingQueue internalQueue;
	/**
	 * comparator
	 */
	private final SequentialComparator comparator;

	/**
	 * Represents the identifier of the next expected entity
	 */
	private T expectedIdentifier;

	public SequentialMessageQueue(SequentialComparator comparator, T initialIdentifier)
	{
		this.internalQueue = new PriorityBlockingQueue(16, comparator);
		this.comparator = comparator;
		this.expectedIdentifier = initialIdentifier;
	}

	@Override
	public boolean empty()
	{
		return internalQueue.isEmpty();
	}

	@Override
	public E pop() throws InterruptedException
	{
		synchronized (lock)
		{
			E firstElement;
			while ((firstElement = internalQueue.peek()) == null || comparator.compare(firstElement, expectedIdentifier) != 0)
			{
				lock.wait();
			}
			internalQueue.remove();
			expectedIdentifier = comparator.getNext(firstElement);
			return firstElement;
		}
	}

	@Override
	public void push(E e) throws InterruptedException
	{
		synchronized (lock)
		{
			internalQueue.put(e);
			lock.notifyAll();
		}

	}

	@Override
	public int size()
	{
		return internalQueue.size();
	}

	/**
	 * Returns the used comparator
	 * 
	 * @return
	 */
	protected SequentialComparator getSequentialComparator()
	{
		return comparator;
	}

}