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