使用WCF实现消息推送

1.协议

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.ServiceModel;

namespace WCFHub.IService
{
    [ServiceContract(CallbackContract=typeof(IEventCallback))]
    public interface IEventService
    {
        [OperationContract(IsOneWay = true)]
        void Subscribe(SubscribeArg a);

        [OperationContract(IsOneWay = true)]
        void Unsubscribe(ArgumentBase<String> a);

        [OperationContract]
        DateTime Ping();

    }

    public interface IEventCallback
    {
         [OperationContract(IsOneWay = true)]
        void OnMessageReceived(ArgumentBase<String> a);
    }
}

View Code

2.实现

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Collections.Concurrent;
using System.ServiceModel;
using System.ServiceModel.Channels;

namespace WCFHub.IService
{
    public class EventServiceImpl:IEventService
    {
        public static readonly ConcurrentDictionary<String, SubscribeContext> _Subscribers = new ConcurrentDictionary<String, SubscribeContext>();

        public string ClientIpAndPort()
        {
            OperationContext context = OperationContext.Current;
            MessageProperties properties = context.IncomingMessageProperties;
            RemoteEndpointMessageProperty endpoint = properties[RemoteEndpointMessageProperty.Name] as RemoteEndpointMessageProperty;
            return endpoint.Address + ":" + endpoint.Port.ToString();
        }

        public void Subscribe(SubscribeArg a)
        {
            Console.WriteLine(ClientIpAndPort());

           var callback  = OperationContext.Current.GetCallbackChannel<IEventCallback>();
            a.Username=a.Username.ToLower();
           _Subscribers[a.Username]=new SubscribeContext(){Arg=a,Callback=callback};

           #region 事件处理
           ICommunicationObject obj = (ICommunicationObject)callback;
           obj.Closed += (s,e) =>
           {
               
               Console.WriteLine("Closed");
           };

           obj.Faulted += (s, e) => {

               Console.WriteLine("Faulted");
           };

           obj.Closing += (s,e) => 
           {

               Console.WriteLine("Closeing" + OperationContext.Current);

               var callback2=(IEventCallback)s;

               _Subscribers.ToList().ForEach(ent => {

                   if (ent.Value.Callback == callback2)
                   {
                       RemoveSubscriber(ent.Value.Arg.Username);
                   }
               });
           };
           #endregion

        }

        public void Unsubscribe(ArgumentBase<string> a)
        {
            RemoveSubscriber(a.Model);
   
        }
        private static void RemoveSubscriber(string username)
        {
            username = username.ToLower();
            if (_Subscribers.ContainsKey(username))
            {
                SubscribeContext outObj = null;
                _Subscribers.TryRemove(username, out outObj);
            }
        }

        public static void PostData(ArgumentBase<string> a)
        {
            Console.WriteLine("收到待发消息:" + a.Model);

            _Subscribers.ToList().ForEach(subscriber =>
            {


                ICommunicationObject callback = (ICommunicationObject)subscriber.Value.Callback;
                if (((ICommunicationObject)callback).State == CommunicationState.Opened)
                {
                    try
                    {
                        //此处需要加上权限判断、订阅判断等
                       subscriber.Value.Callback.OnMessageReceived(a);
                    }
                    catch (Exception ex)
                    {
                        RemoveSubscriber(subscriber.Value.Arg.Username);
                        Console.WriteLine("PostData:" + ex.Message);
                    }
                }
                else
                {
                    RemoveSubscriber(subscriber.Value.Arg.Username);
                    Console.WriteLine("PostData,用户链接已经关闭");
                }
            
            
            
            });
        }


        #region IEventService 成员


        public DateTime Ping()
        {
            Console.WriteLine("Ping:" + ClientIpAndPort() +"," +DateTime.Now);
            return DateTime.Now;
        }

        #endregion
    }

    public class SubscribeContext
    {
        public SubscribeArg Arg { get; set; }
        public IEventCallback Callback { get; set; }

    }
}

View Code

3.实体类

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

namespace WCFHub.IService
{
    [Serializable]
    public class ArgumentBase<T>
    {
        private int code;
        private string msg;
        private T model;

        public int Code
        {
            get { return code; }
            set { code = value; }
        }
        public string Msg
        {
            get { return msg; }
            set { msg = value; }
        }
        public T Model
        {
            get { return model; }
            set { model = value; }

        }
    }

    public class SubscribeArg : ArgumentBase<int>
    {
        public String Username { get; set; }
        public List<int> Alarms { get; set; }
        public SubscribeArg()
        {
            Alarms = new List<int>();
        }
    }
}

View Code

4.服务托管

using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Windows.Forms;
using System.ServiceModel;
using WCFHub.IService;

namespace WCFHub.Win
{
    public partial class Form1 : Form
    {
        public Form1()
        {
            InitializeComponent();
        }
        private ServiceHost _Host = new ServiceHost(typeof(EventServiceImpl));

        private void Form1_Load(object sender, EventArgs e)
        {
            _Host.AddServiceEndpoint(typeof(IEventService), new NetTcpBinding(SecurityMode.None),
                                     "net.tcp://192.168.30.30:9999/EventService"
                                      );
            _Host.Open();
            Console.WriteLine("服务开启...");
        }
        protected override void OnClosed(EventArgs e)
        {
            _Host.Close();
            base.OnClosed(e);
            Console.WriteLine("服务关闭!");
        }

        private void button1_Click(object sender, EventArgs e)
        {
            var data = new ArgumentBase<string>() { Model = textBox1.Text + "," + DateTime.Now.ToString() };
            EventServiceImpl.PostData(data);
        }
    }
}

View Code

5.客户端

using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Windows.Forms;
using WCFHub.IService;
using System.ServiceModel;
using System.Threading;

namespace WCFHub.WinClient
{
    #region MessageReceive
    public delegate void MessageReceivedHandle(ArgumentBase<string> a);

    public class NotifyManager : IEventCallback
    {

        public event MessageReceivedHandle MessageReceived;

        public static int C_MaxErrCount = 5;
        public static int C_HeartbeatInterval = 1000 * 10;//10秒一次心跳检测
        IEventService _Proxy = null;
        private int ErrCounter = 0;

        public bool Enabled { get; set; }

        public NotifyManager()
        {
            Enabled = false;
        }

        private void Close()
        {
            if (_Proxy != null)
            {
                try
                {
                    var comObj = _Proxy as ICommunicationObject;
                    comObj.Abort();
                }
                catch { }
            }
        }
        public void Start()
        {
            Enabled = true;
            StartInternal();
            #region 心跳检测
            var timer = new System.Timers.Timer();
            timer.Enabled = false;
            timer.Interval = C_HeartbeatInterval;
            timer.Elapsed += (s, ie) =>
            {
                try
                {
                    WriteLine("心跳检测...");
                    timer.Enabled = false;
                    _Proxy.Ping();
                    ErrCounter = 0;
                }
                catch (Exception ex)
                {
                    WriteLine(ex.Message);

                    ErrCounter++;
                    if (ErrCounter >= C_MaxErrCount)
                    {
                        Close();
                        StartInternal();
                    }
                }
                finally
                {
                    timer.Enabled = true;
                }
            };
            timer.Start();
            #endregion
        }

        private void StartInternal()
        {
            if (!Enabled) return;

            lock (this)
            {

                try
                {
                    #region
                    ErrCounter = 0;

                    _Proxy = WCFHelper.Factory.CreateChannel(new InstanceContext(this));



                    var comObj = _Proxy as ICommunicationObject;

                    comObj.Faulted += (s, ie) =>
                    {
                        WriteLine("Faulted");

                    };
                    comObj.Closed += (s, ie) =>
                    {
                        WriteLine("Closed!");
                    };
                    comObj.Closing += (s, ie) =>
                    {
                        WriteLine("Closing!");
                    };


                    WriteLine("加载并配置完成!");

                    _Proxy.Subscribe(new SubscribeArg() { Username = Guid.NewGuid().ToString("N") });
                    WriteLine("注册成功!");
                    #endregion

                }
                catch (Exception ex)
                {

                    WriteLine(ex.Message);

                }
            }
        }

        public void Stop()
        {
            Enabled = false;
            Close();
        }

        public void WriteLine(string msg)
        {
            Console.WriteLine(msg + "," + DateTime.Now);
        }

        #region IEventCallback 成员

        public void OnMessageReceived(ArgumentBase<string> a)
        {
            if (MessageReceived != null)
            {
                MessageReceived(a);
            }
        }

        #endregion
    }
    #endregion

    public partial class Form1 : Form
    {
        private SynchronizationContext SyncContext = null;
        public Form1()
        {
            InitializeComponent();
            SyncContext = SynchronizationContext.Current;
        }

        NotifyManager _NotifyManager = null;


        private void Form1_Load(object sender, EventArgs e)
        {

            _NotifyManager = new NotifyManager();
            _NotifyManager.MessageReceived += OnMessageReceived;
            _NotifyManager.Start();
        }
        protected override void OnClosed(EventArgs e)
        {
            if (_NotifyManager != null)
            {
                _NotifyManager.MessageReceived -= this.OnMessageReceived;
                _NotifyManager.Stop();
            }
            base.OnClosed(e);
        }

        public void OnMessageReceived(ArgumentBase<string> a)
        {
            Console.WriteLine("收到消息:" + a.Model +",InvokeRequired:" + this.InvokeRequired);

            if (this.InvokeRequired)
            {
                SyncContext.Post((d) =>
                {
                    textBox1.Text += a.Model + Environment.NewLine;
                }, null);
            }
            else
            {
                textBox1.Text += a.Model + Environment.NewLine;
            }
        }

        private void button1_Click(object sender, EventArgs e)
        {
            if (_NotifyManager != null)
            {
                Console.WriteLine((_NotifyManager as ICommunicationObject).State);
            }
        }
    }

    public class WCFHelper
    {

        private static DuplexChannelFactory<IEventService> _channelFac;

        public static DuplexChannelFactory<IEventService> Factory
        {
            get
            {
                if (_channelFac == null)
                {
                    _channelFac =
                        new DuplexChannelFactory<IEventService>(typeof(NotifyManager), new NetTcpBinding(SecurityMode.None),
                        EndpointStr);

                }
                return _channelFac;
            }
        }

        private static string EndpointStr
        {
            get
            {
                return "net.tcp://192.168.30.30:9999/EventService";
            }
        }
    }
}

View Code