实现TCP框架的正确模式是什么？它是一个过滤器堆栈吗？

Question

我正在尝试实现一个强大的TCP库,它允许用户选择一个应用程序协议或实现他们自己的,并简单地"插入"到客户端/服务器.

通过协议,我的意思是简单地定义如何将流构造成消息的能力.

我正在为堆栈的其余部分使用内置的异步TCP库,并开发了一个客户端,无论何时建立连接,读取或写入数据或引发异常,都会引发事件.

我有两种实现成帧协议的选择.第一个已经工作的是扩展客户端类并覆盖接收的数据事件,这样只有在收到完整消息时才会引发此事件.(即在引擎盖下,我缓冲来自套接字的原始数据,并根据协议决定何时有完整的消息,然后才提出数据接收事件.)这类似于Nito.Asynch库的工作方式.

这种方法的问题在于它意味着每个新协议都需要新的客户端实现.我更喜欢客户端维护可以添加或删除的内部过滤器堆栈.

当在套接字上接收到数据时,它将被传递到第一个缓冲区,该缓冲区将一直缓冲,直到它决定传递删除了头或元数据的完整消息.然后将其传递给堆栈等中的下一个过滤器等.

这样,可以独立于库定义/开发过滤器,并根据配置(在运行时)将过程注入到客户端.

为了实现这一点,我考虑将过滤器定义为由客户端内部保存的System.IO.Stream(传入和传出)的实现对.

从套接字读取的数据将写入堆栈的底部传入流.然后,从该流读取的数据将被写入下一个流等,直到最后一个流(堆栈顶部)返回数据,然后由客户端返回.(我的计划是使用Stream的CopyTo()函数).

写入客户端的数据将写入顶部传出流并向下复制到堆栈,直到底部传出Stream写入底层套接字.

显然有很多事情需要考虑,我试图让我的头脑以正确的方式表现为Stream对象.示例:当有人调用Flush()时,我该怎么办？

这是实现这一目标的好方法还是我在这里重新发明轮子？

Nito.Asynch库

Answer 1

我回答我自己的问题，希望我的解决方案能得到一些好的批评，并可能对其他人有所帮助。

我为协议过滤器和数据帧定义了两个接口。（为了清楚地表达术语，我避免使用“数据包”一词，以避免与较低级别协议中定义的数据包混淆。）

虽然这不是我自己的意图，但我想这可以在任何传输协议（即命名管道、TCP、串行）之上使用。

首先是数据框的定义。这由“数据”（有效负载）以及将传输数据框架为原子“消息”的任何字节组成。

/// <summary>
/// A packet of data with some form of meta data which frames the payload for transport in via a stream.
/// </summary>
public interface IFramedData
{
    /// <summary>
    /// Get the data payload from the framed data (excluding any bytes that are used to frame the data)
    /// i.e. The received data minus protocl specific framing
    /// </summary>
    public readonly byte[] Data { get; }

    /// <summary>
    /// Get the framed data (payload including framing bytes) ready to send
    /// </summary>
    /// <returns>Framed data</returns>
    public byte[] ToBytes();
}

然后是协议过滤器，它从某个源（例如 TCP 套接字，甚至是另一个过滤器，如果它们在堆栈中使用）读取数据并将数据写回。

过滤器应读入数据（包括帧）并为每个完整的帧读取引发 DataReceived 事件。通过 IFramedData 实例的“Data”属性访问有效负载。

当数据写入过滤器时，它应该适当地“构建”它，然后在每次准备发送完整的数据帧时引发 DataToSend 事件。（在我的情况下，这将是立即的，但我尝试允许一种协议，该协议可能会发送固定长度的消息或在返回准备发送的完整帧之前出于某种其他原因缓冲输入。

/// <summary>
/// A protocol filter can be used to read and write data from/to a Stream and frame/deframe the messages.
/// </summary>
/// <typeparam name="TFramedData">The data frame that is handled by this filter</typeparam>
public interface IProtocolFilter<TFramedData> where TFramedData : IFramedData
{
    /// <summary>
    /// Should be raised whenever a complete data frame is ready to send.
    /// </summary>
    /// <remarks>
    /// May be raised after a call to <see cref="FlushSend()"/>
    /// </remarks>
    public event Action<TFramedData> DataToSend;

    /// <summary>
    /// Should be raised whenever a complete data frame has been received.
    /// </summary>
    /// <remarks>
    /// May be raised after a call to <see cref="FlushReceive()"/>
    /// </remarks>
    public event Action<TFramedData> DataReceived;

    /// <summary>
    /// Should be raised if any data written or read breaks the protocol.
    /// This could be due to any asynchronous operation that cannot be raised by the calling function.
    /// </summary>
    /// <remarks>
    /// Behaviour may be protocol specific such as flushing the read or write cache or even resetting the connection.
    /// </remarks>
    public event Action<Exception> ProtocolException;

    /// <summary>
    /// Read data into the recieve buffer
    /// </summary>
    /// <remarks>
    /// This may raise the DataReceived event (possibly more than once if multiple complete frames are read)
    /// </remarks>
    /// <param name="buffer">Data buffer</param>
    /// <param name="offset">Position within the buffer where data must start being read.</param>
    /// <param name="count">Number of bytes to read.</param>
    /// <returns></returns>
    public int Read(byte[] buffer, int offset, int count);

    /// <summary>
    /// Write data to the send buffer.
    /// </summary>
    /// <remarks>
    /// This may raise the DataToSend event (possibly more than once if the protocl requires the data is broken into multiple frames)
    /// </remarks>
    /// <param name="buffer">Data buffer</param>
    /// <param name="offset">Position within the buffer where data must start being read.</param>
    /// <param name="count">Number of bytes to read from the buffer</param>
    public void Write(byte[] buffer, int offset, int count);

    /// <summary>
    /// Flush any data from the receive buffer and if appropriate, raise a DataReceived event.
    /// </summary>
    public void FlushReceive();

    /// <summary>
    /// Flush any data from the send buffer and if appropriate, raise a DataToSend event.
    /// </summary>
    public void FlushSend();
}

然后，我围绕 TcpClient 编写了一个非常简单的包装器，它会异步读取和写入，并在协议栈顶部的过滤器引发 DataReceived 事件或底部的过滤器引发 DataToSend 事件时引发事件（我还将数据写入套接字，但这允许应用程序监视它写入客户端的数据何时实际发送）。