分享我的“艺术品”：公共建筑能耗监测平台的GPRS通讯服务器的开发方法分享

<?xml version="1.0" encoding="utf-8" ?>
<root>
<!-- 通用部分 -->
<!--
building_id:楼栋编号
gateway_id:采集器编号
type:配置信息数据包的类型
-->
<common>
<building_id>XXXXXX</building_id >
<gateway_id>XXX</gateway_id >
<type>以2种操作类型之一</type>
</common>
<!-- 配置信息 -->
<!--操作有2种类型
period:表示服务器对采集器采集周期的配置,period子元素有效
period_ack:表示采集器对服务器采集周期配置信息的应答
-->
<config operation="period/period_ack">
<period>15</period>
</config>
</root>

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Xml.Serialization;namespace UIShell.EcmCommServerService.Protocol.Message
{
[XmlRoot("config", Namespace = "", IsNullable = false)]
public class ConfigActiveMessageBody : MessagePart
{
[XmlAttribute("operation")]
public string Operation { get; set; } [XmlElement("period")]
public int Period { get; set; } public ConfigActiveMessageBody()
{
Operation = "period";
}
}
}

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Xml.Serialization;
using UIShell.EcmCommServerService.Utility;namespace UIShell.EcmCommServerService.Protocol.Message
{
[XmlRoot("root", Namespace = "", IsNullable = false)]
public class ConfigActiveMessage : MessageBase
{
public static ConfigActiveMessage New(string buildingId, string gatewayId, int period)
{
var message = new ConfigActiveMessage();
message.Header.BuildingId = buildingId;
message.Header.GatewayId = gatewayId;
message.Body.Period = period; return message;
} [XmlElement("config")]
public ConfigActiveMessageBody Body { get; set; } public ConfigActiveMessage()
: base(StringEnum.GetStringValue(MessageType.Config_Period))
{
Body = new ConfigActiveMessageBody();
}
}
}

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Xml.Serialization;namespace UIShell.EcmCommServerService.Protocol.Message
{
[XmlRoot("config", Namespace = "", IsNullable = false)]
public class ConfigAckPassiveMessageBody : MessagePart
{
[XmlAttribute("operation")]
public string Operation { get; set; } public ConfigAckPassiveMessageBody()
{
Operation = "period_ack";
}
}
}

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Xml.Serialization;
using UIShell.EcmCommServerService.Utility;namespace UIShell.EcmCommServerService.Protocol.Message
{
[XmlRoot("root", Namespace = "", IsNullable = false)]
public class ConfigAckPassiveMessage : MessageBase
{
public static ConfigAckPassiveMessage New(string buildingId, string gatewayId)
{
var message = new ConfigAckPassiveMessage();
message.Header.BuildingId = buildingId;
message.Header.GatewayId = gatewayId; return message;
} [XmlElement("config")]
public ConfigAckPassiveMessageBody Body { get; set; } public ConfigAckPassiveMessage()
: base(StringEnum.GetStringValue(MessageType.Config_Period_Ack))
{
Body = new ConfigAckPassiveMessageBody();
}
}
}

public override void Start()
{
Trace(string.Format("开始与集中器{0}会话。", ToBeSentMessage.Header.GatewayId)); // 如果发送失败，则重试。
// 尝试次数为： 1 +失败时重复次数
for (int i = ; i <= MessageConstants.RetryTimesOnTimeout; i++)
{
try
{
Send();
OnRoundTripMessageSend(new RoundTripEventArgs() { RoundTrip = this }); try
{
Trace(string.Format("开始第{0}次消息接收。", i + ));
ReceiveResponseMessages();
OnRoundTripCompleted(new RoundTripEventArgs() { RoundTrip = this });
break;
}
catch (Exception ex)
{
Trace(string.Format("第{0}次接收消息失败，因为：{1}，继续尝试。", i + , CommProtocol.GetErrorMessage(ex)));
throw;
}
}
catch (Exception ex)
{
Trace(string.Format("第{0}次发送命令失败，因为：{1}，继续尝试。", i + , CommProtocol.GetErrorMessage(ex)));
if (i == MessageConstants.RetryTimesOnTimeout)
{
Trace(string.Format("第{0}次发送命令失败，因为：{1}，停止尝试。", i + , CommProtocol.GetErrorMessage(ex)));
OnRoundTripError(new RoundTripEventArgs() { RoundTrip = this, Exception = ex });
throw;
}
}
} Completed = true; if (ReceivedResponseMessages != null)
{
Trace(string.Format("当前会话'{0}'接收了{1}个响应消息，详细如下：", RoundTripDescription, ReceivedResponseMessages.Length));
foreach (var message in ReceivedResponseMessages)
{
//Trace("响应命令内容:" + ProtocolUtility.BytesToHexString(message.ToContent()));
Trace("响应消息:" + message.ToString());
}
}
else
{
Trace("当前会话接收的响应消息为空。");
} Trace(string.Format("与集中器{0}会话成功。", ToBeSentMessage.Header.GatewayId));
}

public override void Send()
{
StreamAdapter.SendRawMessage(ToBeSentMessage.ToContent(), ToBeSentMessage.ToXmlContent());
}

public abstract void ReceiveResponseMessages();

public abstract TResponseMessage CreateResponseMessage();public override void Receive()
{
if (ReceivedMessage == null)
{
MessageHeader header;
var receivedMessageContent = StreamAdapter.ReceiveRawMessage(BuildingId, GatewayId, ReceivedMessageType, out header); try
{
ReceivedMessage = MessageSerialiser.DeserializeRaw<TReceivedMessage>(receivedMessageContent);
}
catch (Exception ex)
{
throw new ReceiveMessageException("Parse the received message failed.", ex) { ErrorStatus = ReceiveMessageStatus.Failed };
}
}
}public void SendResponseMessage()
{
ResponsedMessage = CreateResponseMessage();
if (ResponsedMessage != null)
{
//Trace("开始发送的响应消息内容:" + ProtocolUtility.BytesToHexString(ResponsedMessage.ToContent()));
Trace("开始发送的响应消息:" + ResponsedMessage.ToXmlContent()); StreamAdapter.SendRawMessage(ResponsedMessage.ToContent(), ResponsedMessage.ToString());
}
else
{
Trace("不发送响应消息。");
}
}public override void Start()
{
Trace(string.Format("开始尝试与集中器{0}进行被动式会话。", GatewayId)); try
{
Receive();
OnRoundTripMessageReceived(new RoundTripEventArgs() { RoundTrip = this }); try
{
//Trace("接收到消息内容:" + ProtocolUtility.BytesToHexString(ReceivedMessage.ToContent()));
Trace("接收到消息:" + ReceivedMessage.ToXmlContent());
SendResponseMessage();
Completed = true;
OnRoundTripCompleted(new RoundTripEventArgs() { RoundTrip = this });
}
catch (Exception ex)
{
Trace(string.Format("尝试发送响应消息到集中器{0}失败，因为：{1}。", GatewayId, CommProtocol.GetErrorMessage(ex)));
throw;
}
}
catch (Exception ex)
{
Trace(string.Format("尝试从集中器{0}接收消息失败，因为：{1}。", GatewayId, CommProtocol.GetErrorMessage(ex)));
OnRoundTripError(new RoundTripEventArgs() { RoundTrip = this, Exception = ex });
throw;
} Trace(string.Format("与集中器{0}进行被动式会话成功。", GatewayId));
}

public override void Start()
{
int i = ;
int roundTripsCount = RoundTrips.Count;
Trace(string.Format("开始组合会话，由{0}个子会话组成。", roundTripsCount));
RoundTripBase roundTrip;
while (RoundTrips.Count > )
{
roundTrip = RoundTrips.Dequeue();
try
{
Trace(string.Format("开始执行第{0}个子会话。", i));
roundTrip.Start();
Trace(string.Format("第{0}个子会话执行完成。", i));
}
catch (Exception ex)
{
Trace(string.Format("组合会话失败，第{0}个子会话执行失败。", i));
OnRoundTripError(new RoundTripEventArgs() { RoundTrip = roundTrip, Exception = ex });
throw;
}
finally
{
roundTrip.Dispose();
}
i++;
}
Trace(string.Format("组合会话完成，由{0}个子会话组成。", roundTripsCount));
OnRoundTripCompleted(new RoundTripEventArgs() { RoundTrip = this });
}

using System;
using System.Collections.Generic;
using System.Linq;
using System.Net.Sockets;
using System.Text;
using UIShell.EcmCommServerService.Protocol.Message;
using UIShell.EcmCommServerService.Utility;namespace UIShell.EcmCommServerService.Protocol.RoundTrip.Active
{
public class ConfigActiveRoundTrip : ActiveRoundTrip<ConfigActiveMessage, ConfigAckPassiveMessage>
{
public ConfigActiveRoundTrip(
string buildingId,
string gatewayId,
int period,
MessageConstants messageConstants,
TcpClient client)
: base(buildingId, gatewayId, StringEnum.GetStringValue(MessageType.Config_Period_Ack),
ConfigActiveMessage.New(buildingId, gatewayId, period),
messageConstants, client)
{
} public override void ReceiveResponseMessages()
{
MessageHeader header;
var messageContent = ReceiveRawMessage(out header);
var message = MessageSerialiser.DeserializeRaw<ConfigAckPassiveMessage>(messageContent);
ReceivedResponseMessages = new ConfigAckPassiveMessage[] { message };
}
}
}

using System;
using System.Collections.Generic;
using System.Linq;
using System.Net.Sockets;
using System.Text;
using UIShell.EcmCommServerService.Protocol.Message;
using UIShell.EcmCommServerService.Utility;namespace UIShell.EcmCommServerService.Protocol.RoundTrip.Passive
{
public class HeartBeatPassiveRoundTrip : PassiveRoundTrip<HeartBeatNotifyPassiveMessage, HeartBeatTimeActiveMessage>
{
public HeartBeatPassiveRoundTrip(
string buildingId,
string gatewayId,
MessageConstants messageConstants,
TcpClient client)
: base(buildingId, gatewayId, StringEnum.GetStringValue(MessageType.HeartBeat_Notify), messageConstants, client)
{
IsKeepAliveRoundTrip = true;
} public HeartBeatPassiveRoundTrip(
HeartBeatNotifyPassiveMessage receiveMessage,
MessageConstants messageConstants,
TcpClient client)
: this(receiveMessage.Header.BuildingId, receiveMessage.Header.GatewayId, messageConstants, client)
{
ReceivedMessage = receiveMessage;
} public override HeartBeatTimeActiveMessage CreateResponseMessage()
{
return HeartBeatTimeActiveMessage.New(BuildingId, GatewayId, DateTime.Now);
}
}
}

public override void Start()
{
// 检查离线存储区是否存在未传输完成的文件
var item = ContinuousFileStorage.GetNotCompletedFile(BuildingId, GatewayId); if (item == null && Content == null) // 说明被调用的是protected的构造函数，用于检测是否需要进行断点续传。
{
Trace(string.Format("集中器{0}不需要进行文件断点续传。", GatewayId));
return;
} try
{
if (item == null) // 从头开始传输文件
{
Trace(string.Format("集中器{0}开始进行文件传输。", GatewayId));
Trace(string.Format("文件名称：{0}，文件长度：{1}，包大小：{2}，包数：{3}。", FileName, Content.Length, PackageSize, PackageCount));
// 创建离线存储区
ContinuousFileStorage.StartFileTransfer(BuildingId, GatewayId, FileType, FileName, Content, PackageSize); List<int> indexes = new List<int>(); for (int index = ; index <= PackageCount; index++)
{
indexes.Add(index);
} SendFilePackage(indexes);
}
else // 开始断点续传
{
Trace(string.Format("集中器{0}上次文件传输未完成，继续进行断点续传。", GatewayId));
Trace(string.Format("断点续传的文件名称：{0}，文件长度：{1}，包大小：{2}，包数：{3}。", FileName, Content.Length, PackageSize, PackageCount));
} // 查询丢失的包并重传
if (QueryLostPackageAndResend())
{
// 删除离线存储区
ContinuousFileStorage.EndFileTransfer(BuildingId, GatewayId, FileType, FileName, PackageSize);
Trace(string.Format("集中器{0}文件传输成功。", GatewayId));
OnRoundTripCompleted(new RoundTripEventArgs() { RoundTrip = this });
}
else
{
Trace(string.Format("集中器{0}文件传输未完成。", GatewayId));
OnRoundTripError(new RoundTripEventArgs() { RoundTrip = this, Exception = new Exception(string.Format("集中器{0}文件传输未完成。", GatewayId)) });
ContinuousFileStorage.IncrementFileTransferFailedCount(BuildingId, GatewayId);
}
}
catch(Exception ex)
{
OnRoundTripError(new RoundTripEventArgs() { RoundTrip = this, Exception = ex });
ContinuousFileStorage.IncrementFileTransferFailedCount(BuildingId, GatewayId);
}
}

public bool Start()
{
if (_started)
{
return true;
} FireOnStarting(); _commThread = new Thread(() =>
{ RoundTripBase roundTrip;
while (!_exited)
{
Monitor.Enter(_queue.SyncRoot); roundTrip = Dequeue();
if (roundTrip != null)
{
try
{
try
{
Monitor.Exit(_queue.SyncRoot);
OnRoundTripStartingHandler(this, new RoundTripEventArgs() { RoundTrip = roundTrip });
roundTrip.Start();
}
catch (ThreadAbortException)
{
Trace("通讯线程被终止。");
throw;
}
catch (CommStreamException ex) // 无法获取Stream的时候，直接退出？？需要加一个标志位
// 需要抛出事件，通知后续处理，如将RoundTrip另存
{
_exited = true;
roundTrip.Trace("会话失败，因为：链路已经关闭。");
_log.Error(string.Format("Start the round trip '{0}' error.", roundTrip), ex);
}
catch (Exception ex)
{
string error = GetErrorMessage(ex);
roundTrip.Trace(string.Format("会话失败，因为：{0}。", error));
_log.Error(string.Format("Start the round trip '{0}' error.", roundTrip), ex);
} if (!_exited)
{
roundTrip.Trace(Environment.NewLine);
OnRoundTripStartedHandler(this, new RoundTripEventArgs() { RoundTrip = roundTrip });
}
else
{
// 1 将当前失败的RoundTrip保存入队
FailedRoundTrips.Enqueue(roundTrip);
// 2 保存其它没有处理的RoundTrip
do
{
roundTrip = _queue.Dequeue();
if (roundTrip != null)
{
FailedRoundTrips.Enqueue(roundTrip);
}
} while (roundTrip != null);
// 3 停止当前协议
Stop();
}
// 执行完RoundTrip后，开始清理资源
roundTrip.Dispose();
}
catch (ThreadAbortException)
{
Trace("通讯线程被终止。");
throw;
}
catch (Exception ex)
{
_log.Error("Unhandled exception in CommProtocol.", ex);
}
}
else
{
Monitor.Exit(_queue.SyncRoot);
OnIdleHandler(this, new RoundTripEventArgs());
ContinuousFileTransfer();
try
{
CheckPassiveRoundTripAvailableAndEnqueue();
if (_queue.Count == )
{
Thread.Sleep((int)MessageConstants.PassiveRoundTripCheckInterval.TotalMilliseconds);
}
}
catch (ThreadAbortException)
{
Trace("通讯线程被终止。");
throw;
}
catch (CommStreamException ex) // 无法获取Stream的时候，直接退出？？需要加一个标志位
// 需要抛出事件，通知后续处理，如将RoundTrip另存
{
_exited = true;
Trace("检查被动消息失败，因为：链路已经关闭。");
_log.Error("Check the passive message error.", ex);
}
catch (Exception ex)
{
string error = GetErrorMessage(ex);
Trace(string.Format("检查被动消息失败，因为：{0}。", error));
_log.Error("Check the passive message error.", ex);
}
//_autoResetEvent.WaitOne();
}
}
}); _commThread.Start(); _started = true; FireOnStarted();
return true;
}

public void CheckPassiveRoundTripAvailableAndEnqueue()
{
Trace("开始检查被动通讯。"); if (!NetworkUtility.IsConnected(Client))
{
Trace("被动通讯检测时，链路已经关闭，关闭会话。");
Stop();
return;
} // 1 Check ShardInputBuffer
UIShell.OSGi.Utility.Tuple<MessageHeader, byte[]> tuple;
foreach (var pair in _passiveRoundTripFactoryRegistry)
{
while ((tuple = SharedInputBuffer.FindAndThenRemove(BuildingId, GatewayId, pair.Key)) != null)
{
Trace(string.Format("从共享缓冲区获取到一条被动消息，消息头为：'{0}'。", pair.Key));
Enqueue(pair.Value(tuple.Item1, tuple.Item2));
}
}
// 2 Check StreamAdapter
if (Client.Available == )
{
Trace("通讯链路没有可用数据。");
if (!NetworkUtility.IsConnected(Client))
{
Trace("被动通讯检测时，链路已经关闭，关闭会话。");
Stop();
return;
}
}
else
{
while (Client.Available > )
{
MessageHeader header;
var content = CommStreamAdapter.ReceiveOneRawMessage(out header);
if (content != null && content.Length > && header != null)
{
// 1 如果当前消息在被动检测时收到，但不属于注册的被动消息，则放弃该消息。
if (header.BuildingId.Equals(BuildingId) && header.GatewayId.Equals(GatewayId) && !_passiveRoundTripFactoryRegistry.ContainsKey(header.MessageType))
{
Trace(string.Format("从通讯链路获取到一条被动消息，该消息不是注册的被动消息，忽略它，被忽略的消息头为：'{0}'。", header));
continue;
}
// 2 如果当前消息在被动检测时收到，并不属于当前通讯线程处理的范围，则添加到共享缓冲区。
// TODO: 这可能会产生一个Bug，如果接收到其它线程的消息时怎么办？
else if (!header.BuildingId.Equals(BuildingId) || !header.GatewayId.Equals(GatewayId) || !_passiveRoundTripFactoryRegistry.ContainsKey(header.MessageType))
{
Trace(string.Format("从通讯链路获取到一条被动消息，添加到共享缓冲区，消息头为：'{0}'。", header));
SharedInputBuffer.AddSharedBufferItem(new OSGi.Utility.Tuple<MessageHeader, byte[]> { Item1 = header, Item2 = content });
}
else // 3 如果是当前可以处理的被动消息，则创建一个被动RoundTrip
{
CreateRoundTripDelegate createRoundTrip;
if (_passiveRoundTripFactoryRegistry.TryGetValue(header.MessageType, out createRoundTrip))
{
Trace(string.Format("从通讯链路获取到一条被动消息，添加到通讯队列，消息头为：'{0}'。", header));
Enqueue(createRoundTrip(header, content));
}
else
{
Trace(string.Format("从通讯链路获取到一条被动消息，添加到共享缓冲区，消息头为：'{0}'。", header));
SharedInputBuffer.AddSharedBufferItem(new OSGi.Utility.Tuple<MessageHeader, byte[]> { Item1 = header, Item2 = content });
}
}
}
} if (!NetworkUtility.IsConnected(Client))
{
Trace("被动通讯检测时，链路已经关闭，关闭会话。");
Stop();
return;
}
} Trace("检查被动通讯完成。");
}

public RoundTripQueue FailedRoundTrips = new RoundTripQueue();public void Enqueue(RoundTripBase roundTrip)
{
if (!_started)
{
throw new Exception("The protocol is not started yet or exited.");
}
_queue.Enqueue(roundTrip);
OnRoundTripEnquedHandler(this, new RoundTripEventArgs() { RoundTrip = roundTrip }); if (!(roundTrip.IsKeepAliveRoundTrip) || MessageConstants.ShowKeepAliveMessage)
{
roundTrip.OnTraceMessageAdded += DispatchAsyncTraceMessageAddedEvent;
}
try
{
_autoResetEvent.Set();
}
catch
{
}
}

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using UIShell.EcmCommServerService.Protocol.RoundTrip.Active;namespace UIShell.EcmCommServerService.Protocol
{
public partial class CommProtocol
{
public void Config(
int period,
EventHandler<RoundTripEventArgs> onMessageSend,
EventHandler<RoundTripEventArgs> onCompleted,
EventHandler<RoundTripEventArgs> onError)
{
ConfigActiveRoundTrip roundTrip;
Config(period, onMessageSend, onCompleted, onError, out roundTrip);
} internal void Config(
int period,
EventHandler<RoundTripEventArgs> onMessageSend,
EventHandler<RoundTripEventArgs> onCompleted,
EventHandler<RoundTripEventArgs> onError, out ConfigActiveRoundTrip roundTrip)
{
var configRoundTrip = new ConfigActiveRoundTrip(BuildingId, GatewayId, period, MessageConstants, Client);
if (onMessageSend != null)
{
configRoundTrip.OnMessageSend += onMessageSend;
}
if (onCompleted != null)
{
configRoundTrip.OnCompleted += onCompleted;
}
if (onError != null)
{
configRoundTrip.OnError += onError;
}
Enqueue(configRoundTrip);
roundTrip = configRoundTrip;
}
}
}

public partial class HttpCommServer : TrackableBase
{
public MessageConstants MessageConstants { get; private set; }
public string IPAddressString { get; private set; }
public int Port { get; private set; }
public ThreadSafeList<CommProtocol> Sessions { get; private set; } private Thread _listenerThread;
private TcpListener _listener;
private volatile bool _exited;
private ILog _log;
private object _syncRoot = new object(); public HttpCommServer(string ipaddress, int port, MessageConstants messageConstants)
{
IPAddressString = ipaddress;
Port = port;
Sessions = new ThreadSafeList<CommProtocol>(); MessageConstants = messageConstants; _log = BundleActivator.LogService.CreateLog(BundleActivator.Bundle, GetType()); RegisterDomainHandlerCreationDelegates();
} public CommProtocol GetSession(string buildingId, string gatewayId)
{
return Sessions.Find(s => s.BuildingId.Equals(buildingId) && s.GatewayId.Equals(gatewayId));
} public void Start()
{
lock (_syncRoot)
{
IPEndPoint local = new IPEndPoint(IPAddress.Parse(IPAddressString), Port);
_listener = new TcpListener(local);
_listener.Start();
_listenerThread = new Thread(new ThreadStart(ListenGprsRequest));
_listenerThread.Start();
}
OnSessionChanged += OnSessionChangedForDomain;
} private void ListenGprsRequest()
{
while (!_exited)
{
// 接受一次连接
if (!_exited)
{
try
{
TcpClient tcpClient = null;
try
{
tcpClient = _listener.AcceptTcpClient(); Trace(string.Format("接收到来自IP地址'{0}'的连接。", (tcpClient.Client.RemoteEndPoint as IPEndPoint).Address));
_log.Info(string.Format("Accept new connection from ip '{0}'.", (tcpClient.Client.RemoteEndPoint as IPEndPoint).Address));
lock (_syncRoot)
{
if (!_exited)
{
var loginRoundTrip = new LoginCompositeRoundTrip(MessageConstants, tcpClient);
loginRoundTrip.ParentTracker = this;
loginRoundTrip.Start();
loginRoundTrip.Dispose();
var session = new CommProtocol(loginRoundTrip.BuildingId, loginRoundTrip.GatewayId, MessageConstants, tcpClient);
session.ParentTracker = this;
session.Start();
AddSession(session);
// 清空离线存储区
ContinuousDataStorage.Reset(loginRoundTrip.GatewayId);
_log.Info(string.Format("Start the session for gateway '{0}' of the building '{1}'.", session.GatewayId, session.BuildingId));
}
}
}
catch (ThreadAbortException)
{
throw;
}
catch (Exception ex)
{
try
{
if (tcpClient != null) // 登录失败，断开连接
{
tcpClient.Close();
}
}
catch { }
// Trace(string.Format("登录失败，失败IP地址为'{0}'。", tcpClient != null ? (tcpClient.Client.RemoteEndPoint as IPEndPoint).Address.ToString() : "N/A"));
_log.Error("The connection login failed.", ex);
}
}
catch (ThreadAbortException)
{
throw;
}
catch (Exception ex)
{
_log.Error("Can not listen any more.", ex);
break;
}
}
}
} public void Stop()
{
if (_exited)
{
return;
} _log.Info("The server is stopping.");
lock (_syncRoot)
{
_log.Info("The sessions are stopping."); // 不能使用 Sessions.ForEach(s => s.Stop())，这是因为s.Stop将会删除Sessions
// 从而改变ForEach的行为，造成Session泄露。
var sesions = Sessions.ToArray();
foreach (var session in sesions)
{
session.Stop();
}
OnSessionChanged -= OnSessionChangedForDomain;
_log.Info("The sessions are stopped and cleared.");
_listener.Stop(); _log.Info("The listener is stopped."); Thread.Sleep();
try
{
_listenerThread.Abort();
}
catch
{
}
_log.Info("The listener thread is stopped.");
_exited = true;
}
_log.Info("The server is stopped.");
SharedInputBuffer.ClearSharedBuffer();
} public event EventHandler<SessionChangedEventArgs> OnSessionChanged; public void AddSession(CommProtocol session)
{
var oldSession = Sessions.Find(p => p.BuildingId.Equals(session.BuildingId) && p.GatewayId.Equals(session.GatewayId));
if (oldSession != null)
{
RemoveSession(oldSession);
_log.Info(string.Format("The session for gateway '{0}' of building '{1}' already existed, it will be deleted first.", session.GatewayId, session.BuildingId));
} Sessions.Add(session);
session.OnStopped += OnSessionStopped;
_log.Info(string.Format("Add the session for gateway '{0}' of building '{1}'.", session.GatewayId, session.BuildingId));
Trace(string.Format("为采集器'{0}'创建通讯会话，目前会话数目为'{1}'。", session.GatewayId, SessionNumber)); if (OnSessionChanged != null)
{
OnSessionChanged(this, new SessionChangedEventArgs() { ChangedAction = CollectionChangedAction.Add, BuildingId = session.BuildingId, GatewayId = session.GatewayId, Session = session });
}
} private void OnSessionStopped(object sender, EventArgs e)
{
RemoveSession(sender as CommProtocol);
} public void RemoveSession(CommProtocol session)
{
session.OnStopped -= OnSessionStopped;
Sessions.Remove(session);
_log.Info(string.Format("Remove the session for gateway '{0}' of building '{1}'.", session.GatewayId, session.BuildingId));
Trace(string.Format("集中器'{0}'通讯会话已经断开，目前会话数目为'{1}'。", session.GatewayId, SessionNumber));
if (OnSessionChanged != null)
{
OnSessionChanged(this, new SessionChangedEventArgs() { ChangedAction = CollectionChangedAction.Remove, BuildingId = session.BuildingId, GatewayId = session.GatewayId, Session = session });
}
}
}

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using UIShell.EcmCommServerService.Domain;
using UIShell.EcmCommServerService.Protocol;
using UIShell.EcmCommServerService.Protocol.RoundTrip.Passive;
using UIShell.OSGi.Utility;namespace UIShell.EcmCommServerService.Server
{
public partial class HttpCommServer
{
/// <summary>
/// 跟踪每一个Session的RoundTripEnqueued事件，当有RoundTrip注册时，便注册事件，处理领域知识。
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
private void OnSessionChangedForDomain(object sender, SessionChangedEventArgs e)
{
if (e.ChangedAction == OSGi.CollectionChangedAction.Add)
{
e.Session.OnRoundTripEnqued += OnSessionRoundTripEnqued;
}
else
{
e.Session.OnRoundTripEnqued -= OnSessionRoundTripEnqued;
}
} private Dictionary<Type, CreateDomainHandlerDelegate> _handlers = new Dictionary<Type, CreateDomainHandlerDelegate>(); private void RegisterDomainHandlerCreationDelegates()
{
_handlers.Add(typeof(DataReportPassiveRoundTrip), roundTrip => new DataReportDomainHandler() { RoundTrip = roundTrip });
} private void OnSessionRoundTripEnqued(object sender, RoundTripEventArgs e)
{
CreateDomainHandlerDelegate del;
if (_handlers.TryGetValue(e.RoundTrip.GetType(), out del))
{
del(e.RoundTrip);
_log.Info(string.Format("Create handler for RoundTrip '{0}' completed.", e.RoundTrip.GetType().FullName));
}
else
{
_log.Info(string.Format("The handler for RoundTrip '{0}' not found.", e.RoundTrip.GetType().FullName));
}
}
}
}

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using UIShell.EcmCommServerService.Protocol;
using UIShell.OSGi.Utility;namespace UIShell.EcmCommServerService.Domain
{
/// <summary>
/// RoundTrip领域处理器，当RoundTrip操作成功时，将相应结果保存到数据库；
/// 相反，如果操作失败，则需要做异常处理。
/// </summary>
/// <typeparam name="TRoundTrip">RoundTrip类型</typeparam>
public abstract class RoundTripDomainHandler
{
private RoundTripBase _roundTrip; public RoundTripBase RoundTrip
{
get
{
return _roundTrip;
}
set
{
if (_roundTrip == null)
{
AssertUtility.NotNull(value);
_roundTrip = value;
_roundTrip.OnCompleted += OnCompleted;
_roundTrip.OnError += OnError;
}
}
}
public abstract void OnCompleted(object sender, RoundTripEventArgs e);
public abstract void OnError(object sender, RoundTripEventArgs e);
} public delegate RoundTripDomainHandler CreateDomainHandlerDelegate(RoundTripBase roundTrip);
}

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using UIShell.EcmCommServerService.Protocol;
using UIShell.EcmCommServerService.Protocol.RoundTrip.Passive;
using UIShell.EcmCommServerService.Server;namespace UIShell.EcmCommServerService.Domain
{
public class DataReportDomainHandler : RoundTripDomainHandler
{
public override void OnCompleted(object sender, RoundTripEventArgs e)
{
var dataReportRoundTrip = RoundTrip as DataReportPassiveRoundTrip;
e.RoundTrip.Trace("开始将会话结果持久化到数据存储。");
// ...
e.RoundTrip.Trace("将会话结果持久化到数据存储成功。");
} public override void OnError(object sender, RoundTripEventArgs e)
{
// 为采集器关联的Building创建一条失败记录
// ...
}
}
}

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using UIShell.OSGi.Collection;namespace UIShell.EcmCommServerService.Protocol
{
/// <summary>
/// 添加共享输入缓冲区的原因如下：
/// 1 对于GPRS服务器，所有的会话都将从同一个网络数据缓冲区中读取数据；
/// 2 每一个集中器对应一个通讯会话；
/// 3 这样集中器A从网络数据缓冲区读取数据时，可能读取到来自集中器B的数据，
/// 因此，我们需要使用缓冲区将集中器B的数据缓存起来，并继续读取直到读取到
/// A的数据或者读取失败；
/// 4 此外，每一个集中器读取数据时，都先尝试从共享缓冲区读取数据，然后再
/// 尝试从网络数据缓冲区读取。
/// </summary>
public static class SharedInputBuffer
{
private static ThreadSafeList<UIShell.OSGi.Utility.Tuple<MessageHeader, byte[]>> _sharedBuffer = new ThreadSafeList<OSGi.Utility.Tuple<MessageHeader, byte[]>>(); public static ThreadSafeList<UIShell.OSGi.Utility.Tuple<MessageHeader, byte[]>> SharedBuffer
{
get
{
return _sharedBuffer;
}
} public static int Count
{
get
{
using (var locker = SharedBuffer.Lock())
{
return locker.Count;
}
}
} public static UIShell.OSGi.Utility.Tuple<MessageHeader, byte[]> FindAndThenRemoveByMessageType(string type)
{
return FindAndThenRemove(p => p.Item1.MessageType.Equals(type));
} public static UIShell.OSGi.Utility.Tuple<MessageHeader, byte[]> FindAndThenRemove(string buildingId, string gatewayId, string type)
{
return FindAndThenRemove(p => p.Item1.BuildingId.Equals(buildingId) && p.Item1.GatewayId.Equals(gatewayId) && p.Item1.MessageType.Equals(type));
} public static UIShell.OSGi.Utility.Tuple<MessageHeader, byte[]> FindAndThenRemove(Predicate<UIShell.OSGi.Utility.Tuple<MessageHeader, byte[]>> predicate)
{
using (var locker = SharedBuffer.Lock())
{
if (locker.Count > )
{
// 查找缓冲项
var item = SharedBuffer.Find(predicate);
if (item != null)
{
// 删除并返回
RemoveSharedBufferItem(item);
return item;
}
}
return null;
}
} public static void AddSharedBufferItem(UIShell.OSGi.Utility.Tuple<MessageHeader, byte[]> item)
{
SharedBuffer.Add(item);
} public static void RemoveSharedBufferItem(UIShell.OSGi.Utility.Tuple<MessageHeader, byte[]> item)
{
SharedBuffer.Remove(item);
} public static void ClearSharedBuffer()
{
SharedBuffer.Clear();
}
}
}

private List<EventHandler<RoundTripEventArgs>> _onCompletedEventHandlers = new List<EventHandler<RoundTripEventArgs>>();
/// <summary>
/// 这是一个异步事件，避免在处理事件时，阻塞其它RoundTrip的运行。
/// </summary>
public event EventHandler<RoundTripEventArgs> OnCompleted
{
add
{
_onCompletedEventHandlers.Add(value);
}
remove
{
_onCompletedEventHandlers.Remove(value);
}
}public override void Dispose()
{
_onCompletedEventHandlers.Clear();
_onErrorEventHandlers.Clear();
base.Dispose();
}

try
{
try
{
Monitor.Exit(_queue.SyncRoot);
OnRoundTripStartingHandler(this, new RoundTripEventArgs() { RoundTrip = roundTrip });
roundTrip.Start();
}
catch (ThreadAbortException)
{
Trace("通讯线程被终止。");
throw;
}
catch (CommStreamException ex) // 无法获取Stream的时候，直接退出？？需要加一个标志位
// 需要抛出事件，通知后续处理，如将RoundTrip另存
{
_exited = true;
roundTrip.Trace("会话失败，因为：链路已经关闭。");
_log.Error(string.Format("Start the round trip '{0}' error.", roundTrip), ex);
}
catch (Exception ex)
{
string error = GetErrorMessage(ex);
roundTrip.Trace(string.Format("会话失败，因为：{0}。", error));
_log.Error(string.Format("Start the round trip '{0}' error.", roundTrip), ex);
}
// ......
// 执行完RoundTrip后，开始清理资源
roundTrip.Dispose();
}
catch (ThreadAbortException)
{
Trace("通讯线程被终止。");
throw;
}
catch (Exception ex)
{
_log.Error("Unhandled exception in CommProtocol.", ex);
}

[TestFixture]
public partial class ProtocolTest
{
private HttpCommServer _commServer;
private TcpClient _tcpClient;
private CommProtocol _currentSession;
private CommStreamAdapter _clientStremAdapter;
public AutoResetEvent AutoResetEvent { get; set; } public const string BuildingId = "b001";
public const string GatewayId = "g001";
static ProtocolTest()
{
// 加载插件运行时，准备运行环境
if (BundleRuntime.Instance == null)
{
BundleRuntime bundleRuntime = new BundleRuntime("../../../");
bundleRuntime.Start();
}
}
}

[SetUp]
public void Setup()
{
AutoResetEvent = new AutoResetEvent(false);
MessageConstants.GprsMessageConstants.Timeout = new TimeSpan(, , );
MessageConstants.GprsMessageConstants.RetryTimesOnTimeout = ;
_commServer = new HttpCommServer("127.0.0.1", , MessageConstants.GprsMessageConstants);
_commServer.OnTraceMessageAdded += (sender, e) => { Debug.WriteLine(e.Message); };
_commServer.Start(); _tcpClient = new TcpClient();
_tcpClient.Connect("127.0.0.1", );
_clientStremAdapter = new CommStreamAdapter(_commServer.MessageConstants, _tcpClient);
_clientStremAdapter.ParentTracker = _commServer; var request = ValidateRequestPassiveMessage.New(BuildingId, GatewayId);
_clientStremAdapter.SendRawMessage(request.ToContent(), request.ToXmlContent()); var prefix = ProtocolUtility.BytesToHexString(MessageConstants.XmlMessagePrefixBytes);
var root = ProtocolUtility.BytesToHexString(MessageConstants.XmlMessageRootStartBytes);
var common = ProtocolUtility.BytesToHexString(MessageConstants.XmlMessageCommonStartBytes);
var commonend = ProtocolUtility.BytesToHexString(MessageConstants.XmlMessageCommonEndBytes);
var rootend = ProtocolUtility.BytesToHexString(MessageConstants.XmlMessageRootEndBytes); MessageHeader header;
var messageContent = _clientStremAdapter.ReceiveOneRawMessage(out header); Assert.AreEqual(header.BuildingId, BuildingId);
Assert.AreEqual(header.GatewayId, GatewayId);
Assert.AreEqual(header.MessageType, StringEnum.GetStringValue(MessageType.Validate_Sequence)); var sequenceMessage = MessageSerialiser.DeserializeRaw<ValidateSequenceActiveMessage>(messageContent);
string md5 = CreateSequenceAndHash(sequenceMessage.Body.Sequence);
var md5Message = ValidateMd5PassiveMessage.New(BuildingId, GatewayId, md5); _clientStremAdapter.SendRawMessage(md5Message.ToContent(), md5Message.ToXmlContent()); messageContent = _clientStremAdapter.ReceiveOneRawMessage(out header); Assert.AreEqual(header.BuildingId, BuildingId);
Assert.AreEqual(header.GatewayId, GatewayId);
Assert.AreEqual(header.MessageType, StringEnum.GetStringValue(MessageType.Validate_Result)); var resultMessage = MessageSerialiser.DeserializeRaw<ValidateResultActiveMessage>(messageContent);
Assert.AreEqual(resultMessage.Body.Result, "pass"); while (_currentSession == null)
{
_currentSession = _commServer.Sessions.Find(s => s.BuildingId.Equals(BuildingId) && s.GatewayId.Equals(GatewayId));
Thread.Sleep();
}
}

[Test]
public void ConfigRoundTrip()
{
bool completed = false;
Exception ex = null;
MessageHeader receivedMessageHeader = null;
ConfigActiveMessage receivedMessage = null; _currentSession.Config(,
(sender, e) => {
var configMessage = _clientStremAdapter.ReceiveOneRawMessage(out receivedMessageHeader);
receivedMessage = MessageSerialiser.DeserializeRaw<ConfigActiveMessage>(configMessage); var configAckMessage = ConfigAckPassiveMessage.New(BuildingId, GatewayId);
_clientStremAdapter.SendRawMessage(configAckMessage.ToContent(), configAckMessage.ToXmlContent());
},
(sender, e) => {
ex = e.Exception;
completed = true;
AutoResetEvent.Set();
},
(sender, e) => {
ex = e.Exception;
completed = false;
AutoResetEvent.Set();
}); AutoResetEvent.WaitOne(); Assert.IsTrue(completed);
Assert.AreEqual(receivedMessageHeader.MessageType, StringEnum.GetStringValue(MessageType.Config_Period));
Assert.AreEqual(receivedMessage.Body.Period, );
}

[Test]
public void HeartBeatPassiveRoundTrip()
{
bool completed = false;
Exception ex = null;
HeartBeatTimeActiveMessage responseMessage = null; _currentSession.OnRoundTripStarted += (sender, e) => {
if (e.RoundTrip is HeartBeatPassiveRoundTrip)
{
var roundTrip = e.RoundTrip as HeartBeatPassiveRoundTrip;
responseMessage = roundTrip.ResponsedMessage;
ex = e.Exception;
completed = true;
AutoResetEvent.Set();
}
}; var heartbeatMessage = HeartBeatNotifyPassiveMessage.New(BuildingId, GatewayId);
_clientStremAdapter.SendRawMessage(heartbeatMessage.ToContent(), heartbeatMessage.ToXmlContent()); AutoResetEvent.WaitOne(); Assert.IsTrue(completed);
Assert.NotNull(responseMessage);
Assert.AreEqual(responseMessage.Header.MessageType, StringEnum.GetStringValue(MessageType.HeartBeat_Time));
}