/** * Serves as an unidirectional, double-buffered message queue between 2 modules. * @remark Only one writer and one reader are allowed. */ template class NonLockingMessageQueue : public MessageQueue { typedef queue MessageQueue; typedef vector> MessageStorage; typedef vector*> 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 */ virtual shared_ptr 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* address; this->mStorage[index] = shared_ptr(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 popFromExchange() { if (this->mExchange[this->mReadIndex] != NULL) { // read the shared message... shared_ptr 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*)NULL); } } int mReadIndex; int mWriteIndex; MessageQueue mWriteQueue; MessageStorage mStorage; MessageExchange mExchange; };