使用WCF实现消息推送
2024-09-26 19:14:16
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);
}
}
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; } }
}
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>();
}
}
}
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);
}
}
}
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 = ;
public static int C_HeartbeatInterval = * ;//10秒一次心跳检测
IEventService _Proxy = null;
private int ErrCounter = ; 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 = ;
}
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 = ; _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";
}
}
}
}
最新文章
- 序言<EntityFramework6.0>
- jquery插件treetable使用
- ZeroMQ研究与应用分析
- Codeforces Round #313 (Div. 2) C. Gerald's Hexagon 数学
- json-c-0.9 的简单用法
- 关于vmware下复制linux系统虚拟机后eth0变成eth1问题解决
- SVG之初识
- 密码加密md5和sha
- IIS本地部署项目出错
- [jQuery编程挑战]005 使用最短的代码生成元素的闪烁效果
- JavaScript基础（一）
- Chapter 7 代理模式
- [Android阅读代码]圆形旋转菜单CircleMenu
- NET那点不为人知的事
- ASP.NET MVC View向Controller提交数据
- Linux下的Shell编程（1）最简单的例子
- c语言相关概念
- grid的简单使用
- 斯诺克台球比赛规则 (Snooker)
- Ubuntu 14.04 将一个sh文件制作成类似于windows下的可以双击执行的快捷方式
热门文章
- LeetCode OJ - Best Time to Buy and Sell Stock
- posix对线程的调整
- laravel 整合 swoole ，并简单 ab 测试对比性能以及在 PHPstorm 中利用debug调试配置swoole服务中的PHP代码
- There is no Action mapped for namespace / and action name login. - [unknown location]
- word文档批量合并工具
- Java LinkedList 和 ArrayList
- [转]jQuery 读取 xml
- 用shp制作geoJson格式地图数据(shp convert to geoJson)
- 为Web页中的Table对象创建一个映射表
- samba性能调优