zeroflag

#region Header
//********************************************************************
//	author:         Thomas "zeroflag" Kraemer
//	author email:   zeroflag@zeroflag.de
//	
//	Copyright (C) 2006-2007  Thomas "zeroflag" Kraemer
//	
//	license:	(LGPL)
//	
//		This library is free software; you can redistribute it and/or
//		modify it under the terms of the GNU Lesser General Public
//		License as published by the Free Software Foundation; either
//		version 2.1 of the License, or (at your option) any later version.
//
//		This library is distributed in the hope that it will be useful,
//		but WITHOUT ANY WARRANTY; without even the implied warranty of
//		MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
//		Lesser General Public License for more details.
//
//		You should have received a copy of the GNU Lesser General Public
//		License along with this library; if not, write to the Free Software
//		Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
//
//		http://www.gnu.org/licenses/lgpl.html#TOC1
//
//*********************************************************************
#endregion Header

using System;
using System.Collections.Generic;

namespace Sage.Threading
{
	/// <summary>
	/// A single reader single writer message queues without locking.
	/// </summary>
	/// <typeparam name="T"></typeparam>
	/// <remarks>Update must be called by the writer when lots (= more than the buffer size, 2048 by default) of messages were sent before but nothing is added.</remarks>
	public class MessageQueue<T> : Sage.Threading.IMessageQueue<T>
		where T : class // <-- must be a reference type or atomic type to be threadsafe
	{
		Queue<T> m_WriteQueue = new Queue<T>();
		/// <summary>
		/// Queue for new values before they're added to the buffer (through Update).
		/// </summary>
		protected Queue<T> WriteQueue
		{
			get { return m_WriteQueue; }
		}

		#region MessageContainer
		/// <summary>
		/// A box arround the stored objects.
		/// With this one can set a value in the array without causing it to synchronize to the other threads.
		/// </summary>
		protected class MessageContainer
		{
			T m_Value;
			/// <summary>
			/// Gets or sets the value.
			/// Getting the value will result in it's deletion (null).
			/// </summary>
			public T Value
			{
				get
				{
					T value = m_Value;
					if (value != null)
					{
						this.m_Value = null;
					}
					return value;
				}
				set
				{
					m_Value = value;
				}
			}
			/// <summary>
			/// Whether the box has a value or not.
			/// </summary>
			public bool HasValue
			{
				get
				{
					if (m_Value != null)
						return true;
					else
						return false;
				}
			}
		}
		#endregion MessageContainer

		public MessageQueue(int bufferSize)
		{
			this.m_Buffer = new MessageContainer[bufferSize];
			this.Clear();
		}

		/// <summary>
		/// Creates a new message queue and initializes it's buffer.
		/// </summary>
		public MessageQueue()
		{
			this.Clear();
		}

		protected void Clear()
		{
			for (int i = 0; i < this.Buffer.Length; i++)
			{
				this.Buffer[i] = new MessageContainer();
			}
		}

		private int ReadIndex = 0;
		private int WriteIndex = 0;

		MessageContainer[] m_Buffer = new MessageContainer[2048];
		/// <summary>
		/// The queue's ring buffer.
		/// Default size is 2048 entries.
		/// </summary>
		protected MessageContainer[] Buffer
		{
			get { return m_Buffer; }
		}

		#region IMessageQueue<T> Members
		/// <summary>
		/// Whether there are values ready to be read.
		/// *** READER THREAD ***
		/// </summary>
		public bool IsEmpty
		{
			get { return !this.Buffer[this.ReadIndex].HasValue; }
		}
		/// <summary>
		/// Pops a value from the queue.
		/// *** READER THREAD ***
		/// </summary>
		/// <returns></returns>
		public T Pop()
		{
			if (this.Buffer[this.ReadIndex].HasValue)
			{
				// grab the current value...
				T value = this.Buffer[this.ReadIndex].Value;

				// continue to the next...
				this.ReadIndex++;
				if (this.ReadIndex >= this.Buffer.Length)
				{
					this.ReadIndex = 0;
				}

				return value;
			}
			else
			{
				// no value waiting...
				return null;
			}
		}
		/// <summary>
		/// Pushes a new value to the queue and updates the buffer.
		/// *** WRITER THREAD ***
		/// </summary>
		/// <param name="msg"></param>
		public void Push(T msg)
		{
			this.WriteQueue.Enqueue(msg);

			this.Update();
		}

		/// <summary>
		/// Fills the buffer from the queue.
		/// *** WRITER THREAD ***
		/// </summary>
		public void Update()
		{
			while (this.PushToBuffer()) ;
		}

		#endregion
		/// <summary>
		/// Pushes a single value from the WriteQueue to the Buffer.
		/// </summary>
		/// <returns>True if a value was pushed, false if it wasn't (like no items in queue or no free cell in buffer).</returns>
		protected bool PushToBuffer()
		{
			if (!this.Buffer[this.WriteIndex].HasValue && this.WriteQueue.Count > 0)
			{
				// get the message...
				T value = this.WriteQueue.Dequeue();

				// store the message...
				this.Buffer[this.WriteIndex].Value = value;

				// move the index...
				this.WriteIndex++;
				if (this.WriteIndex >= this.Buffer.Length)
				{
					this.WriteIndex = 0;
				}
				// yes, we pushed a new message from the write-queue to the buffer...
				return true;

			}
			else
			{
				// no, the exchange table is full...
				return false;
			}
		}
	}
}