zeroflag

/**
* Serves as an unidirectional, double-buffered message queue between 2 modules.
* @remark Only one writer and one reader are allowed.
*/
template<int BufferSize = 2048>
class NonLockingMessageQueue : public MessageQueue
{
	typedef queue<Message> MessageQueue;
	typedef vector<shared_ptr<Message>> MessageStorage;
	typedef vector<shared_ptr<Message>*> MessageExchange;

public:
	NonLockingMessageQueue()
		: mReadIndex(0), mWriteIndex(0), mExchange(BufferSize, NULL), mStorage(BufferSize)
	{
		// make sure the exchange field is initialized with NULL...
		for(MessageExchange::iterator i = mExchange.begin();
			i != mExchange.end(); i++)
		{
			(*i) = NULL;
		}
	}

	~NonLockingMessageQueue()
	{
	}

	/**
	 * Pushes a new message to the queue.
	 * Writer interface.
	 * @return void
	 * @param Message message
	 * @remark Also tries to fill the Buffer.
	 */
	virtual void push(Message message)
	{
		this->mWriteQueue.push(message);

		while(this->pushToExchange());
	}

	/**
	 * Pops a message from the queue/buffer.
	 * Reader interface.
	 * @return shared_ptr<Message>
	 */
	virtual shared_ptr<Message> pop()
	{
		return this->popFromExchange();
	}

	/**
	 * Updates the buffer.
	 * Writer interface.
	 * @return void
	 */
	virtual void update()
	{
		while(this->pushToExchange());
	}

	/**
	 * Whether the queue and buffer are empty.
	 * Reader interface.
	 * @return bool
	 */
	virtual bool isEmpty()
	{
		if (this->mExchange[this->mReadIndex] != NULL)
			return false;
		else
			return true;
	}

protected:
	inline bool pushToExchange()
	{
		if (this->mExchange[this->mWriteIndex] == NULL && !this->mWriteQueue.empty())
		{
			// get the message...
			Message value = this->mWriteQueue.front();
			int index = this->mWriteIndex++;
			this->mWriteQueue.pop();

			if (this->mWriteIndex >= BufferSize)
			{
				this->mWriteIndex = 0;
			}

			// store the message...
			shared_ptr<Message>* address;
			this->mStorage[index] = shared_ptr<Message>(new Message(value));
			address = &this->mStorage[index];
			
			// exchange the message's address...
			this->mExchange[index] = address;

			// yes, we pushed a new message from the write-queue to the exchange table...
			return true;
		}
		else
		{
			// no, the exchange table is full...
			return false;
		}
	}

	inline shared_ptr<Message> popFromExchange()
	{
		if (this->mExchange[this->mReadIndex] != NULL)
		{
			// read the shared message...
			shared_ptr<Message> value = *this->mExchange[this->mReadIndex];
			// remove the pointer from the exchange table...
			this->mExchange[this->mReadIndex] = NULL;

			// increase the read index...
			this->mReadIndex++;
			if (this->mReadIndex >= BufferSize)
			{
				this->mReadIndex = 0;
			}

			// return the shared message...
			return value;
		}
		else
		{
			// nope, exchange table is empty...
			return shared_ptr<Message>((Message*)NULL);
		}
	}

	int mReadIndex;
	int mWriteIndex;
	MessageQueue mWriteQueue;
	MessageStorage mStorage;
	MessageExchange mExchange;
};