notificationagent.cs

微软的行业应用解决方案示例

using System;
using System.Linq;
using System.Collections.Generic;
using System.Text;
using StoreAndFwdTransport;
using System.Runtime.Serialization;

using System.ServiceModel.Channels;
using System.Threading;
using HardwareDistributor.Business;

namespace HardwareDistributor.Synchronization
{
    delegate void ProcessMessageDelegate(string action, Order order);
    
    /// <summary>
    /// singleton class used for sending and receiving peer to peer messages
    /// </summary>
    /// 
    sealed class NotificationAgent
    {
        #region Constants

        private const int SHUTDOWN_TIMEOUT = 2000;
        private const string CHANNEL_NAME = "HardwareDistributorChannel";

        #endregion

        #region Fields
        private static NotificationAgent _instance = new NotificationAgent();
        
        private WcfTransport _transport;
        private IInputChannel _inputChannel;
        private CFMessagingSerializer _serializer;

        //helpers for the receiving loop 
        private volatile bool _receiveMessages;
        private volatile bool _processMessages = true; // be default we always process the messages

        /// <summary>
        /// controls whether we process the messages or not
        /// </summary>
        public bool ProcessMessages
        {
            get { return _processMessages; }
            set { _processMessages = value; }
        }
        private Thread _receiveThread;

        #endregion 

        #region Properties

        public static NotificationAgent Instance
        {
            get { return _instance; }
        }

        //call back method for processing received message
        public ProcessMessageDelegate ProcessMessage;

        #endregion

        #region Constructors

        private NotificationAgent()
        {
            _transport = new WcfTransport();

            //single input channel
            _inputChannel = _transport.AcceptInputChannel(CHANNEL_NAME);

            _serializer = new CFMessagingSerializer( typeof(Order) );
        }

        #endregion 
                
        #region Methods

        /// <summary>
        /// start listening for messages
        /// </summary>
        /// <param name="processMessages">instruction to process message or simply discard them</param>
        public void StartListening()
        {
            if (!_receiveMessages)
            {
                //spawn off a back ground thread to wait for messages
                _receiveMessages = true;
                _receiveThread = new Thread(new ThreadStart(DoReceive));
                _receiveThread.IsBackground = true;
                _receiveThread.Start();
            }

        }


        /// <summary>
        /// continously receives and dispatched messages that are received.
        /// </summary>
        private void DoReceive()
        {
            while (_receiveMessages)
            {
                //wait until message arrives
                Message message = _inputChannel.Receive();

                //send off for processing
                if (_processMessages)
                {
                    DispatchMessage(message);
                }
            }

        }

        /// <summary>
        /// Desserializes the order and invokes the call back method
        /// </summary>
        /// <param name="message">received message</param>
        private void DispatchMessage(Message message)
        {
            //if the callback function has been hooked up 
            if (ProcessMessage != null)
            {
                //extract order object from the message body
                Order order = message.GetBody<Order>(_serializer);
                
                string action = message.Headers.Action;

                ProcessMessage(action, order);
            }
        }

        /// <summary>
        /// stop listening for messages
        /// </summary>
        public void StopListening()
        {
            
            //indicate that we are to stop receiving messages
            _receiveMessages = false;

            //break out of the blocking call to Receive()
            _inputChannel.Close();

            //ensure the thread is shut down
            bool shutDownSuccessful = _receiveThread.Join(SHUTDOWN_TIMEOUT);
            if (shutDownSuccessful)
                _receiveThread.Abort();

            _receiveThread = null;
        }

        /// <summary>
        /// sends message using the output channel
        /// </summary>
        /// <param name="emailAddress">email address to send message to </param>
        /// 
        /// <param name="action">action string</param>
        public void SendMessage(string emailAddress, string action, Order body)
        {
            //get the open output channel
            IOutputChannel outputChannel = _transport.CreateOutputChannel(
                                                CHANNEL_NAME, emailAddress);

            //create message
            Message message = Message.CreateMessage(
                                MessageVersion.Soap12WSAddressing10, action,
                                body, _serializer); 
            //send message
            outputChannel.Send(message);
        }

        /// <summary>
        /// reads and discards any pending notifications in the input channel
        /// </summary>
        public void ClearPendingMesasages()
        {
            bool moreMessagesAvailable = true;

            while (moreMessagesAvailable)
            {
                Message dummyMessage;
                //wipe up any messages available in 1 sec
                moreMessagesAvailable = _inputChannel.TryReceive(System.TimeSpan.FromMilliseconds(1000), out dummyMessage);
                if (moreMessagesAvailable)
                {
                    //receive and discard
                    _inputChannel.Receive();
                }
            }
        }

        #endregion


    }
}