内存流是否像文件流一样阻塞

Question

我正在使用一个库,要求我提供一个实现此接口的对象:

public interface IConsole {
    TextWriter StandardInput { get; }
    TextReader StandardOutput { get; }
    TextReader StandardError { get; }
}

然后,对象的读者将被库使用:

IConsole console = new MyConsole();
int readBytes = console.StandardOutput.Read(buffer, 0, buffer.Length);

通常,实现IConsole的类具有来自外部进程的StandardOutput流.在这种情况下,console.StandardOutput.Read通过阻塞来调用,直到有一些数据写入StandardOutput流.

我要做的是创建一个测试IConsole实现,它使用MemoryStreams和echo在StandardInput上显示的任何内容返回到StandardInput.我试过了:

MemoryStream echoOutStream = new MemoryStream();
StandardOutput = new StreamReader(echoOutStream);

但问题是console.StandardOutput.Read将返回0而不是阻塞,直到有一些数据.无论如何,如果没有可用数据或者我可以使用的内存流有什么不同,我可以得到一个MemoryStream来阻止吗？

Answer 1

受你的回答启发,这是我的多线程,多写版本:

public class EchoStream : MemoryStream
{
    private readonly ManualResetEvent _DataReady = new ManualResetEvent(false);
    private readonly ConcurrentQueue<byte[]> _Buffers = new ConcurrentQueue<byte[]>();

    public bool DataAvailable{get { return !_Buffers.IsEmpty; }}

    public override void Write(byte[] buffer, int offset, int count)
    {
        _Buffers.Enqueue(buffer);
        _DataReady.Set();
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        _DataReady.WaitOne();

        byte[] lBuffer;

        if (!_Buffers.TryDequeue(out lBuffer))
        {
            _DataReady.Reset();
            return -1;
        }

        if (!DataAvailable)
            _DataReady.Reset();

        Array.Copy(lBuffer, buffer, lBuffer.Length);
        return lBuffer.Length;
    }
}

对于您的版本,您应该在写入时读取流,而不能连续写入.我的版本在ConcurrentQueue中缓冲任何写入的缓冲区(将其更改为简单的队列并将其锁定非常简单)

Answer 2

最后,通过继承MemoryStream并接管Read和Write方法,我找到了一种简单的方法.

public class EchoStream : MemoryStream {

    private ManualResetEvent m_dataReady = new ManualResetEvent(false);
    private byte[] m_buffer;
    private int m_offset;
    private int m_count;

    public override void Write(byte[] buffer, int offset, int count) {
        m_buffer = buffer;
        m_offset = offset;
        m_count = count;
        m_dataReady.Set();
    }

    public override int Read(byte[] buffer, int offset, int count) {
        if (m_buffer == null) {
            // Block until the stream has some more data.
            m_dataReady.Reset();
            m_dataReady.WaitOne();    
        }

        Buffer.BlockCopy(m_buffer, m_offset, buffer, offset, (count < m_count) ? count : m_count);
        m_buffer = null;
        return (count < m_count) ? count : m_count;
    }
}