c#网络通信框架networkcomms内核解析之五 数据监听

本文基于networkcomms2.3.1开源版本 gplv3协议

在networkcomms中，服务器端可以同步监听数据，也可以异步监听数据。

以开源的networkcomms.2.31为例

服务器端监听代码：

protected override void StartIncomingDataListen() { if (!NetworkComms.ConnectionExists(ConnectionInfo.RemoteEndPoint, ConnectionType.TCP)) { CloseConnection(true, 18); throw new ConnectionSetupException("A connection reference by endPoint should exist before starting an incoming data listener."); } #if WINDOWS_PHONE
var stream = socket.InputStream.AsStreamForRead(); stream.BeginRead(dataBuffer, 0, dataBuffer.Length, new AsyncCallback(IncomingTCPPacketHandler), stream); #else
lock (delegateLocker) { //同步监听模式
if (NetworkComms.ConnectionListenModeUseSync) { if (incomingDataListenThread == null) { incomingDataListenThread = new Thread(IncomingTCPDataSyncWorker); //Incoming data always gets handled in a time critical fashion
incomingDataListenThread.Priority = NetworkComms.timeCriticalThreadPriority; incomingDataListenThread.Name = "IncomingDataListener"; incomingDataListenThread.Start(); } } //异步监听模式
else tcpClientNetworkStream.BeginRead(dataBuffer, 0, dataBuffer.Length, new AsyncCallback(IncomingTCPPacketHandler), tcpClientNetworkStream); } #endif

if (NetworkComms.LoggingEnabled) NetworkComms.Logger.Trace("Listening for incoming data from " + ConnectionInfo); }

我们以异步监听为例，分析一下监听进入的数据的过程（同步监听类似)

/// <summary>
/// Asynchronous incoming connection data delegate /// </summary>
/// <param name="ar">The call back state object</param>
void IncomingTCPPacketHandler(IAsyncResult ar) { //Initialised with true so that logic still works in WP8
bool dataAvailable = true; #if !WINDOWS_PHONE
//Incoming data always gets handled in a timeCritical fashion at this point
Thread.CurrentThread.Priority = NetworkComms.timeCriticalThreadPriority; //int bytesRead;
#endif

try { #if WINDOWS_PHONE
var stream = ar.AsyncState as Stream; var count = stream.EndRead(ar); totalBytesRead = count + totalBytesRead; #else NetworkStream netStream = (NetworkStream)ar.AsyncState; if (!netStream.CanRead) throw new ObjectDisposedException("Unable to read from stream."); totalBytesRead = netStream.EndRead(ar) + totalBytesRead; dataAvailable = netStream.DataAvailable; #endif
if (totalBytesRead > 0) { //收到数据后，更新连接信息类上的数据最后传输时间
ConnectionInfo.UpdateLastTrafficTime(); //If we have read a single byte which is 0 and we are not expecting other data
if (totalBytesRead == 1 && dataBuffer[0] == 0 && packetBuilder.TotalBytesExpected - packetBuilder.TotalBytesCached == 0) { if (NetworkComms.LoggingEnabled) NetworkComms.Logger.Trace(" ... null packet removed in IncomingPacketHandler() from " + ConnectionInfo + ". 1"); } else { //if (NetworkComms.LoggingEnabled) NetworkComms.Logger.Trace(" ... " + totalBytesRead.ToString() + " bytes added to packetBuilder."); //If there is more data to get then add it to the packets lists; //添加数据到“数据包创建器”（packetBuilder)中
packetBuilder.AddPartialPacket(totalBytesRead, dataBuffer); #if !WINDOWS_PHONE
//If we have more data we might as well continue reading syncronously //In order to deal with data as soon as we think we have sufficient we will leave this loop //当接收到的数据小于数据包的大小的时候，循环接收
while (dataAvailable && packetBuilder.TotalBytesCached < packetBuilder.TotalBytesExpected) { int bufferOffset = 0; //We need a buffer for our incoming data //First we try to reuse a previous buffer
if (packetBuilder.TotalPartialPacketCount > 0 && packetBuilder.NumUnusedBytesMostRecentPartialPacket() > 0) dataBuffer = packetBuilder.RemoveMostRecentPartialPacket(ref bufferOffset); else
//If we have nothing to reuse we allocate a new buffer
dataBuffer = new byte[NetworkComms.ReceiveBufferSizeBytes]; //从数据流中接收数据
totalBytesRead = netStream.Read(dataBuffer, bufferOffset, dataBuffer.Length - bufferOffset) + bufferOffset; if (totalBytesRead > 0) { ConnectionInfo.UpdateLastTrafficTime(); //If we have read a single byte which is 0 and we are not expecting other data
if (totalBytesRead == 1 && dataBuffer[0] == 0 && packetBuilder.TotalBytesExpected - packetBuilder.TotalBytesCached == 0) { if (NetworkComms.LoggingEnabled) NetworkComms.Logger.Trace(" ... null packet removed in IncomingPacketHandler() from " + ConnectionInfo + ". 2"); //LastTrafficTime = DateTime.Now;
} else { //if (NetworkComms.LoggingEnabled) NetworkComms.Logger.Trace(" ... " + totalBytesRead.ToString() + " bytes added to packetBuilder for connection with " + ConnectionInfo + ". Cached " + packetBuilder.TotalBytesCached.ToString() + "B, expecting " + packetBuilder.TotalBytesExpected.ToString() + "B.");
packetBuilder.AddPartialPacket(totalBytesRead, dataBuffer); dataAvailable = netStream.DataAvailable; } } else
break; } #endif } } //如果接收到的数据足够的多
if (packetBuilder.TotalBytesCached > 0 && packetBuilder.TotalBytesCached >= packetBuilder.TotalBytesExpected) { //Once we think we might have enough data we call the incoming packet handle handoff //Should we have a complete packet this method will start the appriate task //This method will now clear byes from the incoming packets if we have received something complete. //调用IncomingPacketHandleHandOff方法处理“数据包接收器”(packetBuilder)中已经接收到的字节数据，并把二进制数据还原成相应的类型，进行处理
IncomingPacketHandleHandOff(packetBuilder); } if (totalBytesRead == 0 && (!dataAvailable || ConnectionInfo.ConnectionState == ConnectionState.Shutdown)) CloseConnection(false, -2); else { //We need a buffer for our incoming data //First we try to reuse a previous buffer
if (packetBuilder.TotalPartialPacketCount > 0 && packetBuilder.NumUnusedBytesMostRecentPartialPacket() > 0) dataBuffer = packetBuilder.RemoveMostRecentPartialPacket(ref totalBytesRead); else { //If we have nothing to reuse we allocate a new buffer
dataBuffer = new byte[NetworkComms.ReceiveBufferSizeBytes]; totalBytesRead = 0; } #if WINDOWS_PHONE stream.BeginRead(dataBuffer, totalBytesRead, dataBuffer.Length - totalBytesRead, IncomingTCPPacketHandler, stream); #else
//继续接收TCP连接上的数据
netStream.BeginRead(dataBuffer, totalBytesRead, dataBuffer.Length - totalBytesRead, IncomingTCPPacketHandler, netStream); #endif } } catch (IOException) { CloseConnection(true, 12); } catch (ObjectDisposedException) { CloseConnection(true, 13); } catch (SocketException) { CloseConnection(true, 14); } catch (InvalidOperationException) { CloseConnection(true, 15); } catch (Exception ex) { NetworkComms.LogError(ex, "Error_TCPConnectionIncomingPacketHandler"); CloseConnection(true, 31); } #if !WINDOWS_PHONE Thread.CurrentThread.Priority = ThreadPriority.Normal; #endif }

接收到的数据都交给PacketBuilder(数据包创建器）进行处理了，PacketBuilder如何对接收到的二进制数据进行处理，将在下一章进行分析。

www.cnblogs.com/networkcomms

www.networkcomms.cn