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);}
}

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; }}
}

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>();}}
}

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);}}
}

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 MessageReceivepublic 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{#regionErrCounter = 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}#endregionpublic 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";}}}
}