从流中不断读取？

Question

我有一个Stream对象偶尔会获得一些数据,但是间隔时间不可预测.流上出现的消息是明确定义的,并提前声明其有效负载的大小(大小是每个消息的前两个字节中包含的16位整数).

我想要一个StreamWatcher类来检测Stream何时有一些数据.一旦它发生,我想要引发一个事件,以便订阅的StreamProcessor实例可以处理新消息.

这可以用C#事件完成而不直接使用Threads吗？看起来它应该是直截了当的,但我无法完全理解设计这个问题的正确方法.

Answer 1

当你说不直接使用线程时,我假设你仍然希望通过异步调用间接使用它们,否则这将不是非常有用.

您需要做的就是包装其异步方法Stream并将结果存储在缓冲区中.首先,让我们定义规范的事件部分:

public delegate void MessageAvailableEventHandler(object sender,
    MessageAvailableEventArgs e);

public class MessageAvailableEventArgs : EventArgs
{
    public MessageAvailableEventArgs(int messageSize) : base()
    {
        this.MessageSize = messageSize;
    }

    public int MessageSize { get; private set; }
}

现在,异步读取流中的一个16位整数,并在准备就绪时报告:

public class StreamWatcher
{
    private readonly Stream stream;

    private byte[] sizeBuffer = new byte[2];

    public StreamWatcher(Stream stream)
    {
        if (stream == null)
            throw new ArgumentNullException("stream");
        this.stream = stream;
        WatchNext();
    }

    protected void OnMessageAvailable(MessageAvailableEventArgs e)
    {
        var handler = MessageAvailable;
        if (handler != null)
            handler(this, e);
    }

    protected void WatchNext()
    {
        stream.BeginRead(sizeBuffer, 0, 2, new AsyncCallback(ReadCallback),
            null);
    }

    private void ReadCallback(IAsyncResult ar)
    {
        int bytesRead = stream.EndRead(ar);
        if (bytesRead != 2)
            throw new InvalidOperationException("Invalid message header.");
        int messageSize = sizeBuffer[1] << 8 + sizeBuffer[0];
        OnMessageAvailable(new MessageAvailableEventArgs(messageSize));
        WatchNext();
    }

    public event MessageAvailableEventHandler MessageAvailable;
}

我认为就是这样.这假设处理消息的任何类也可以访问Stream并准备基于事件中的消息大小同步或异步地读取它.如果您希望观察者类实际读取整个消息,那么您将不得不添加更多代码来执行此操作.

Answer 2

是的，这是可以做到的。将非阻塞Stream.BeginRead方法与AsyncCallback结合使用。当数据可用时，异步调用回调。在回调中，调用Stream.EndRead获取数据，并再次调用Stream.BeginRead获取下一块数据。将传入数据缓冲在足以容纳消息的字节数组中。一旦字节数组已满（可能需要多个回调调用），引发事件。然后读取下一个消息大小，创建一个新的缓冲区，重复，完成。