MemoryStream 等到有东西要读

Question

这是一个非常简单的问题，但如果不与某人交谈，我似乎无法弄清楚。

我需要一个像 MemoryStream 这样的流，它会在读取时异步阻塞，直到有要读取的内容或超时。

更新： 好的。我已经放弃了，自己写了Wrapper类，但是EndRead总是返回0。请看下面的代码。 （不提供面向任务的解决方案。）

    public class BlockingMemoryStream : MemoryStream
    {
        ManualResetEventSlim isReadReady = new ManualResetEventSlim(false);

        public override void Write(byte[] buffer, int offset, int count)
        {
            base.Write(buffer, offset, count);
            //base.Position = offset; //I do not know if I need this!!!!
            isReadReady.Set();
        }

        public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback callback, object state)
        {
            IAsyncResult result = base.BeginRead(buffer, offset, count, callback, state);
            return result;
        }

        public override int EndRead(IAsyncResult asyncResult)
        {
            isReadReady.Wait(/*600000*/);
            int aa = base.EndRead(asyncResult);
            return aa;
        }
    }

Answer 1

尝试这样的事情，它所做的只是循环并等待直到有要读取的对象。

    private static readonly MemoryStream _reader;
    private static object _data;

    static Program()
    {
        _reader = new MemoryStream();
    }

    private static void Main(string[] args)
    {
        Task.Run(async delegate()
        {
            while (true)
            {
                if (_data == null)
                    await Task.Delay(1000); // so the cpu can have rest
                                            // you can lower the value of this
                else
                {
                    // read here
                    await Task.Delay(1000);
                }
            }
        });
    }

Answer 2

如果我理解正确，您想要一个您写入的流，另一个线程从中读取。您可以自己滚动，即：

public sealed class MyStream : Stream
{
    private readonly MemoryStream underlyingStream = new MemoryStream();
    private readonly AutoResetEvent waitHandle = new AutoResetEvent(false);

    public int Timeout { get; set; }

    public MyStream()
    {
        Timeout = 5000;
    }

    public override void Write(byte[] buffer, int offset, int count)
    {
        // Write to the stream and notify any waiting threads
        underlyingStream.Write(buffer, offset, count);
        waitHandle.Set();
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        int bytes;
        while ((bytes = underlyingStream.Read(buffer, offset, count)) == 0)
        {
            // 0 bytes read (end of stream), wait Timeout ms for someone to write
            if (!waitHandle.WaitOne(Timeout))
            {
                throw new TimeoutException();
            }
        }

        return bytes;
    }

    // TODO other mandatory methods
}

我在没有测试的情况下编写了上面的示例，Stream 的实现是不完整的，它只是为您的解决方案显示了一个可能的方向。如果有多个线程来自它Read，则可能需要多个Timeout 才能完成。

如果您不喜欢摆弄等待句柄，您也可以在 TCP 环回上使用套接字并使用 NetworkStream。