Windows Services Made Simple

using System;
using System.Messaging;
using System.Threading;

using Pegasus.Diagnostics;
using Pegasus.Log4Net;
using Pegasus.Threading;

namespace Pegasus.Messaging
{
	/// <summary>
	/// Summary description for MessageQueueWorker.
	/// </summary>
	public abstract class MessageQueueReceiver
	{
		// Local Instance Values
		private ILog m_log = LogManager.GetLogger( typeof( MessageQueueReceiver ) );

		private MessageQueue m_queue;
		private TimeSpan m_timeout = TimeSpan.FromSeconds( 0.25 );
		private TimeSpan m_joinTimeout = TimeSpan.FromSeconds( 10.0 );

		private Thread m_workingThread = null;
		private bool m_done = false;

		private int m_messagesReceived = 0;
		private int m_messagesDropped = 0;
		
		/// <summary>
		/// Default Constructor
		/// </summary>
		/// <param name="queue">Queue that messages will be received on.</param>
		public MessageQueueReceiver( MessageQueue queue )
		{
			ParamCode.AssertNotNull( queue, "queue" );

			m_queue = queue;

			// Launch this object on it's own thread
			m_workingThread = new Thread( new ThreadStart( ThreadProc ) );
			m_workingThread.Name = "MessageQueueReceiver Thread: " + GetHashCode();
			m_workingThread.Start();
		}

		/// <summary>
		/// Get the number of messages received.  You can set the value only to 0.
		/// </summary>
		public int MessagesReceived
		{
			get
			{
				return m_messagesReceived;
			}

			set
			{
				ParamCode.Assert( value == 0, "MessagesReceived can only be set to 0" );
				m_messagesReceived = 0;
			}
		}

		/// <summary>
		/// Get the number of messages dropped.  You can set the value only to 0.
		/// </summary>
		public int MessagesDropped
		{
			get
			{
				return m_messagesDropped;
			}

			set
			{
				ParamCode.Assert( value == 0, "MessagesDropped can only be set to 0" );
				m_messagesDropped = 0;
			}
		}

		/// <summary>
		/// Called to stop the object and close the processing thread
		/// </summary>
		public virtual void Stop()
		{
			m_log.Debug( "MessageQueueReceiver:Stop()" );

			m_done = true;
			
			if( m_workingThread != null )
			{
				m_log.Debug( "MessageQueueReceiver:Stop: Waiting for thread to stop." );
				if( !m_workingThread.Join( m_joinTimeout ) )
				{
					m_log.Debug( "MessageQueueReceiver:Stop: Stop thread timed out, aborting thread." );
					m_workingThread.Abort();
				}
			}
		}

		
		/// <summary>
		/// Call to have the dirived class process a message.
		/// </summary>
		/// <param name="msg">The message object that was received from the queue.</param>
		/// <returns>True if the message was processed, false if the mesasge was not processed</returns>
		protected abstract bool ProcessMessage( Message msg );

		/// <summary>
		/// Background thread for receiving message from a queue.
		/// </summary>
		private void ThreadProc()
		{
			m_log.Debug( "MessageQueueReceiver:ThreadProc()" );

			while( !m_done )
			{
				try
				{
					Message msg = m_queue.Receive( m_timeout );
					if( msg != null )
					{
						try
						{
							if( ProcessMessage( msg ) )
							{
								m_messagesReceived++;
							}
							else
							{
								m_messagesDropped++;
							}
						}
						catch( Exception e )
						{
							m_log.Error( "Exception procesing message", e );
							m_messagesDropped++;
						}
					}
				}
				catch( ThreadAbortException )
				{
					m_log.DebugFormat( "MessageQueueReceiver:ThreadProc: Thread abort exception receiving message on {0}", m_queue.QueueName );

					Thread.ResetAbort();
					m_done = true;
				}
				catch( MessageQueueException e )
				{
					if( e.MessageQueueErrorCode != MessageQueueErrorCode.IOTimeout )
					{
						m_log.Error( string.Format( "MessageQueueReceiver:ThreadProc: Exception receiving message on {0}", m_queue.QueueName ), e );
					}
					else
					{
						m_log.DebugFormat( "MessageQueueReceiver:ThreadProc: Timeout exception receiving message on {0}", m_queue.QueueName );
					}
				}
				catch( Exception e )
				{
					m_log.Error( string.Format( "MessageQueueReceiver:ThreadProc: Exception receiving message on {0}", m_queue.QueueName ), e );
				}
			}
		}
	}
}