Blog: Writing a Concurrent .NET Stream (6)

CompletedReadAndWriteMethods.cs

public bool WritingFinished { get; private set; }

public void CloseWrite()
{
    WritingFinished = true;
    _readWait.Set();
}

public override int Read(byte[] buffer, int offset, int count)
{
    var maxByteCount = offset + count > buffer.Length ? buffer.Length : count;

    if (_buffer.Count == 0)
    {
        _readWait.WaitOne();

    }
    _readWait.Reset();
    var actualBytesRead = 0;

    for (var i = offset; i < maxByteCount && _buffer.TryTake(out var b); i++)
    {
        buffer[i] = b;
        actualBytesRead++;
    }

    return actualBytesRead;
}

public override void Write(byte[] buffer, int offset, int count)
{
    if (WritingFinished)
    {
        throw new InvalidOperationException("Attempt to write after CloseWrite() has been called.");
    }

    var maxByteCount = offset + count > buffer.Length ? buffer.Length : count;
    for (var i = offset; i < maxByteCount; i++)
    {
        _buffer.TryAdd(buffer[i]);
    }
    _readWait.Set();
}