TcpClient更新一次，然后继续侦听

Question

我正在做一个基于客户端服务器的XNA游戏，我在将Message Framing机制集成到网络部分时遇到了一些困难。

这是我从这里获得的消息框架协议类，有一些重构：

using System;

namespace XnaCommonLib.Network
{
    // Original source: http://blog.stephencleary.com/2009/04/sample-code-length-prefix-message.html
    /// <summary>
    /// Maintains the necessary buffers for applying a length-prefix message framing protocol over a stream.
    /// </summary>
    /// <remarks>
    /// <para>Create one instance of this class for each incoming stream, and assign a handler to <see cref="MessageArrived"/>. As bytes arrive at the stream, pass them to <see cref="DataReceived"/>, which will invoke <see cref="MessageArrived"/> as necessary.</para>
    /// <para>If <see cref="DataReceived"/> raises <see cref="System.Net.ProtocolViolationException"/>, then the stream data should be considered invalid. After that point, no methods should be called on that <see cref="PacketProtocol"/> instance.</para>
    /// <para>This class uses a 4-byte signed integer length prefix, which allows for message sizes up to 2 GB. Keepalive messages are supported as messages with a length prefix of 0 and no message data.</para>
    /// <para>This is EXAMPLE CODE! It is not particularly efficient; in particular, if this class is rewritten so that a particular interface is used (e.g., Socket's IAsyncResult methods), some buffer copies become unnecessary and may be removed.</para>
    /// </remarks>
    public class PacketProtocol
    {
        private const int LengthBufferSize = sizeof(int);

        /// <summary>
        /// Wraps a message. The wrapped message is ready to send to a stream.
        /// </summary>
        /// <remarks>
        /// <para>Generates a length prefix for the message and returns the combined length prefix and message.</para>
        /// </remarks>
        /// <param name="message">The message to send.</param>
        public static byte[] WrapMessage(byte[] message)
        {
            // Get the length prefix for the message
            var lengthPrefix = BitConverter.GetBytes(message.Length);

            // Concatenate the length prefix and the message
            var ret = new byte[lengthPrefix.Length + message.Length];
            lengthPrefix.CopyTo(ret, 0);
            message.CopyTo(ret, lengthPrefix.Length);

            return ret;
        }

        /// <summary>
        /// Wraps a keepalive (0-length) message. The wrapped message is ready to send to a stream.
        /// </summary>
        public static byte[] WrapKeepaliveMessage()
        {
            return BitConverter.GetBytes(0);
        }

        /// <summary>
        /// Initializes a new <see cref="PacketProtocol"/>, limiting message sizes to the given maximum size.
        /// </summary>
        /// <param name="maxMessageBufferSize">The maximum message size supported by this protocol. This may be less than or equal to zero to indicate no maximum message size.</param>
        public PacketProtocol(int maxMessageBufferSize)
        {
            // We allocate the buffer for receiving message lengths immediately
            lengthBuffer = new byte[LengthBufferSize];
            maxMessageSize = maxMessageBufferSize;
        }

        /// <summary>
        /// The buffer for the length prefix; this is always 4 bytes long.
        /// </summary>
        private readonly byte[] lengthBuffer;

        /// <summary>
        /// The buffer for the data; this is null if we are receiving the length prefix buffer.
        /// </summary>
        private byte[] dataBuffer;

        /// <summary>
        /// The number of bytes already read into the buffer (the length buffer if <see cref="dataBuffer"/> is null, otherwise the data buffer).
        /// </summary>
        private int bytesReceived;

        /// <summary>
        /// The maximum size of messages allowed.
        /// </summary>
        private readonly int maxMessageSize;

        /// <summary>
        /// Indicates the completion of a message read from the stream.
        /// </summary>
        /// <remarks>
        /// <para>This may be called with an empty message, indicating that the other end had sent a keepalive message. This will never be called with a null message.</para>
        /// <para>This event is invoked from within a call to <see cref="DataReceived"/>. Handlers for this event should not call <see cref="DataReceived"/>.</para>
        /// </remarks>
        public Action<byte[]> MessageArrived
        {
            get; set;
        }

        /// <summary>
        /// Notifies the <see cref="PacketProtocol"/> instance that incoming data has been received from the stream. This method will invoke <see cref="MessageArrived"/> as necessary.
        /// </summary>
        /// <remarks>
        /// <para>This method may invoke <see cref="MessageArrived"/> zero or more times.</para>
        /// <para>Zero-length receives are ignored. Many streams use a 0-length read to indicate the end of a stream, but <see cref="PacketProtocol"/> takes no action in this case.</para>
        /// </remarks>
        /// <param name="data">The data received from the stream. Cannot be null.</param>
        /// <exception cref="System.Net.ProtocolViolationException">If the data received is not a properly-formed message.</exception>
        public void DataReceived(byte[] data)
        {
            // Process the incoming data in chunks, as the ReadCompleted requests it

            // Logically, we are satisfying read requests with the received data, instead of processing the
            //  incoming buffer looking for messages.

            var i = 0;
            while (i != data.Length)
            {
                // Determine how many bytes we want to transfer to the buffer and transfer them
                var bytesAvailable = data.Length - i;
                if (dataBuffer != null)
                {
                    // We're reading into the data buffer
                    var bytesRequested = dataBuffer.Length - bytesReceived;

                    // Copy the incoming bytes into the buffer
                    var bytesTransferred = Math.Min(bytesRequested, bytesAvailable);
                    Array.Copy(data, i, dataBuffer, bytesReceived, bytesTransferred);
                    i += bytesTransferred;

                    // Notify "read completion"
                    ReadCompleted(bytesTransferred);
                }
                else
                {
                    // We're reading into the length prefix buffer
                    var bytesRequested = lengthBuffer.Length - bytesReceived;

                    // Copy the incoming bytes into the buffer
                    var bytesTransferred = Math.Min(bytesRequested, bytesAvailable);
                    Array.Copy(data, i, lengthBuffer, bytesReceived, bytesTransferred);
                    i += bytesTransferred;

                    // Notify "read completion"
                    ReadCompleted(bytesTransferred);
                }
            }
        }

        /// <summary>
        /// Called when a read completes. Parses the received data and calls <see cref="MessageArrived"/> if necessary.
        /// </summary>
        /// <param name="count">The number of bytes read.</param>
        /// <exception cref="System.Net.ProtocolViolationException">If the data received is not a properly-formed message.</exception>
        private void ReadCompleted(int count)
        {
            // Get the number of bytes read into the buffer
            bytesReceived += count;

            if (dataBuffer == null)
            {
                // We're currently receiving the length buffer

                if (bytesReceived != LengthBufferSize)
                {
                    // We haven't gotten all the length buffer yet: just wait for more data to arrive
                }
                else
                {
                    // We've gotten the length buffer
                    var length = BitConverter.ToInt32(lengthBuffer, 0);

                    // Sanity check for length < 0
                    if (length < 0)
                        throw new System.Net.ProtocolViolationException("Message length is less than zero");

                    // Another sanity check is needed here for very large packets, to prevent denial-of-service attacks
                    if (maxMessageSize > 0 && length > maxMessageSize)
                        throw new System.Net.ProtocolViolationException("Message length " + length.ToString(System.Globalization.CultureInfo.InvariantCulture) + " is larger than maximum message size " + maxMessageSize.ToString(System.Globalization.CultureInfo.InvariantCulture));

                    // Zero-length packets are allowed as keepalives
                    if (length == 0)
                    {
                        bytesReceived = 0;
                        MessageArrived?.Invoke(new byte[0]);
                    }
                    else
                    {
                        // Create the data buffer and start reading into it
                        dataBuffer = new byte[length];
                        bytesReceived = 0;
                    }
                }
            }
            else
            {
                if (bytesReceived != dataBuffer.Length)
                    // We haven't gotten all the data buffer yet: just wait for more data to arrive
                    return;

                // We've gotten an entire packet
                MessageArrived?.Invoke(dataBuffer);

                // Start reading the length buffer again
                dataBuffer = null;
                bytesReceived = 0;
            }
        }
    }
}

我已经调试了代码，它似乎运行正常。但问题是我如何使用这段代码。据我所知，每次接收数据时，我都需要调用PacketProtocol::DataReceived。但从我的观点来看，由于我使用的是TCP，所以很难理解什么才是Data Received，因为TCP使用Stream，而UDP则使用数据报，所以对于TCP来说，DataReceived的定义非常困难。

我尝试使用以下助手方法来使其工作：

using System.IO;
using System.Net.Sockets;

namespace XnaCommonLib.Network
{
    public static class HelperMethods
    {
        public static void Receive(TcpClient connection, BinaryReader reader, PacketProtocol packetProtocol)
        {
            var buffer = new byte[connection.ReceiveBufferSize];

            while (reader.Read(buffer, 0, buffer.Length) > 0) // this is where it gets stuck
            {
                packetProtocol.DataReceived(buffer);
                buffer = new byte[connection.ReceiveBufferSize];
            }
        }
    }
}

这种方法的用法如下：

ConnectionHandler -客户端网络管理类

private void ConnectionHandler_InteractWithServer()
{
    while (Connection.Connected)
    {
        try
        {
            HelperMethods.Receive(Connection, Reader, PacketProtocol);
        }
        catch (Exception)
        {
            Connection.Close();
            break;
        }

        Thread.Sleep(Constants.Time.UpdateThreadSleepTime);
    }
}

这个方法是在线程中调用的，所以它会不断地运行。PacketProtocol的客户端回调如下：

private void PacketProtocol_MessageRecievedCallback(byte[] data)
{
    if (data.Length == 0)
        return;

    var stringData = Encoding.UTF8.GetString(data);
    ProcessServerUpdate(stringData);
    WritePlayerData();
}

private void ProcessServerUpdate(string message)
{
    UpdatePing();

    var incomingUpdate = JsonConvert.DeserializeObject<ServerToClientUpdateMessage>(message);
    EmsServerEndpoint.BroadcastIncomingEvents(incomingUpdate.Broadcasts);

    foreach (var update in incomingUpdate.PlayerUpdates)
        ApplyUpdate(update);
}

private void ApplyUpdate(PlayerUpdate update)
{
    var entity = new Entity(update.Guid);
    if (!ClientGameManager.EntityPool.Exists(entity))
    {
        var newGo = ClientGameManager.BeginAllocateRemote(entity.Id);
        newGo.Components.Get<NetworkPlayer>().Update(update);
        ClientGameManager.EndAllocate(newGo);
    }
    else
    {
        var remoteComponents = ClientGameManager.EntityPool.GetComponents(entity);
        remoteComponents.Get<NetworkPlayer>().Update(update);
    }
}

private void WritePlayerData()
{
    var message = new ClientToServerUpdateMessage
    {
        Broadcasts = EmsServerEndpoint.Flush(),
        PlayerUpdate = new PlayerUpdate(GameObject.Components)
    };

    var messageBytes = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(message));
    var wrapperMessage = PacketProtocol.WrapMessage(messageBytes);
    Writer.Write(wrapperMessage);
}

它是回调方法和它调用的方法一起使用的方法。

这是服务器对PacketProtocol的回调：

private void PacketProtocol_MessageArrivedCallback(byte[] bytes)
{
    if (bytes.Length == 0)
        return;

    var stringData = Encoding.UTF8.GetString(bytes);
    ProcessClientUpdate(stringData);
    SendServerUpdate();
}

private void SendServerUpdate()
{
    var message = JsonConvert.SerializeObject(new ServerToClientUpdateMessage
    {
        Broadcasts = EmsServerEndpoint.Flush(),
        PlayerUpdates = PlayerUpdates()
    });

    var messageBytes = Encoding.UTF8.GetBytes(message);
    Writer.Write(PacketProtocol.WrapMessage(messageBytes));
}

private void ProcessClientUpdate(string clientMessageString)
{
    var clientMessage = JsonConvert.DeserializeObject<ClientToServerUpdateMessage>(clientMessageString);
    EmsServerEndpoint.BroadcastIncomingEvents(clientMessage.Broadcasts);
    UpdateClient(clientMessage.PlayerUpdate);
}

private IList<PlayerUpdate> PlayerUpdates()
{
    return GameManager.EntityPool.AllThat(PlayerUpdate.IsPlayer).Select(c => new PlayerUpdate(c)).ToList();
}

private void UpdateClient(PlayerUpdate playerUpdate)
{
    var components = GameObject.Components;
    components.Get<DirectionalInput>().Update(playerUpdate.Input);
}

所以现在对于实际的问题:代码对于第一次更新正常工作-消息被正确地接收并且更新发生。而在此之后，客户机和服务器都会被困在while (reader.Read(buffer, 0, buffer.Length) > 0)行中。

要开始实际的通信，这是ConnectionHandler所做的工作：

WriteLoginDataToServer(name, team);
ReadLoginResponseFromServer();
WritePlayerData();
UpdateThread.Start();

在启动更新线程(与服务器运行通信循环)之前，它调用WritePlayerData，后者将初始更新发送到服务器。我认为这会使他调用SendServerUpdate，客户端会捡起它，循环会继续，但它不会发生。

Answer 1

using System.IO;
using System.Net.Sockets;

namespace XnaCommonLib.Network
{
    public static class HelperMethods
    {
        public static void Receive(TcpClient connection, BinaryReader reader, PacketProtocol packetProtocol)
        {
            var bufferSize = connection.Available;
            var buffer = reader.ReadBytes(bufferSize);
            packetProtocol.DataReceived(buffer); 
        }
    }
}

这就是Receive现在看起来的样子，而且它可以工作。感谢null在评论中的家伙！

Answer 2

这就是我如何处理接收信息的方法：

接收数据的类：

class DataReceiver
{    
 public Delegate MessageReceived(string message);
 pulic MessageReceived MessageReceivedEvent;

 private void Socket;
 private bool IsReceiving;
 public Class DataReceiver(Socket sock)
 {
 Socket = sock;
 }
 public void StartReceiving()
  => Task.Run((Action)ReceiveLoop);
 public void StopReceiving()
  => IsReceiving = false;

 private void ReceiveLoop()
 {
  IsReceiving = true;
  while(IsReceiving)
  {
   byte[] LengthHeader = new byte[4];
   Socket.Receive(LenthHeader, 0, 4);
   byte[] Buffer = new byte[BitConverter.ToInt32(LengthHeader);
   Socket.Receive(Buffer, 0,Buffer.Length);
   MessageReceivedEvent?.Invoke(Encoding.UTF8.GetString(Buffer);
  }
 }          
}

而不是在消息完成之前积累字节，而是接收包含消息长度的header，然后完全接收该消息。然后将其转换为字符串，您可以通过MessageReceivedEvent处理它。

若要发送数据，请使用此函数：

public void Send(string message, Socket sock)
{
  byte[] LengthHeader;
  byte[] Buffer;
  Buffer = Encoding.UTF8.GetBytes(message);
  LengthHeader = BitConverter.GetBytes(Buffer.Length);
  sock.Send(LengthHeader, 0, 4);
  socket.Send(Buffer, 0, Buffer.Length;
}

您必须在客户端和服务器上实现这一点，并简化消息接收/发送，您只需序列化发送，反序列化即可接收。