如何在不释放Stream的情况下标记读写操作完成？

Great question! This is a common scenario when working with protocol libraries that only support full messages but need to handle large payloads via chunking, without closing the underlying stream via Dispose. Here's a practical, protocol-aligned solution tailored to your requirements:

Instead of relying on Dispose to signal stream completion, we'll use protocol-level chunk markers to indicate the final chunk, paired with custom stream wrappers that mimic "end-of-stream" behavior without closing the underlying resource. This keeps the original stream intact for reuse while adhering to your protocol's chunking rules.

1. Sender Side: Chunked Write Wrapper #

Create a wrapper stream that handles chunk serialization, including a flag to mark the final chunk. This avoids calling Dispose on the underlying stream, since completion is signaled via the protocol itself.

public class ChunkedWriteStream : Stream
{
    private readonly Stream _innerStream;
    private bool _finalChunkSent;

    public ChunkedWriteStream(Stream innerStream) => _innerStream = innerStream;

    // Explicit method to write chunks with final flag
    public void WriteChunk(byte[] chunkData, bool isFinal)
    {
        if (_finalChunkSent)
            throw new InvalidOperationException("Final chunk already sent; cannot write additional data.");

        // Step 1: Write chunk header (final flag + chunk length)
        var header = new byte[8];
        BitConverter.GetBytes(isFinal ? 1 : 0).CopyTo(header, 0);
        BitConverter.GetBytes(chunkData.Length).CopyTo(header, 4);
        _innerStream.Write(header, 0, header.Length);

        // Step 2: Write chunk payload
        _innerStream.Write(chunkData, 0, chunkData.Length);
        _innerStream.Flush();

        if (isFinal)
            _finalChunkSent = true;
    }

    // Implement required Stream members (disable unsupported operations)
    public override bool CanRead => false;
    public override bool CanSeek => false;
    public override bool CanWrite => true;
    public override long Length => throw new NotSupportedException();
    public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
    public override void Flush() => _innerStream.Flush();
    public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException();
    public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
    public override void SetLength(long value) => throw new NotSupportedException();
    public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();

    // Override Dispose to avoid closing the inner stream
    protected override void Dispose(bool disposing)
    {
        if (!_finalChunkSent)
            throw new InvalidOperationException("Final chunk was not sent before disposing the stream.");
        // DO NOT call _innerStream.Dispose() - keep the underlying stream alive
    }
}

Usage for Sender: #

• Wrap your original stream with ChunkedWriteStream
• Write each data chunk with WriteChunk(data, false)
• For the last chunk, call WriteChunk(finalData, true) to signal completion via protocol

2. Receiver Side: Chunked Read Wrapper #

On the receiving end, create a wrapper that reads chunks, checks the final flag, and returns 0 from Read() once the final chunk is fully processed—mimicking the behavior of a closed stream without disposing the underlying resource.

public class ChunkedReadStream : Stream
{
    private readonly Stream _innerStream;
    private bool _finalChunkReceived;
    private int _remainingBytesInChunk;

    public ChunkedReadStream(Stream innerStream) => _innerStream = innerStream;

    private bool ReadChunkHeader()
    {
        if (_finalChunkReceived)
            return false;

        // Read header (final flag + chunk length)
        var headerBuffer = new byte[8];
        if (_innerStream.Read(headerBuffer, 0, headerBuffer.Length) != headerBuffer.Length)
            throw new EndOfStreamException("Incomplete chunk header received.");

        bool isFinal = BitConverter.ToInt32(headerBuffer, 0) == 1;
        _remainingBytesInChunk = BitConverter.ToInt32(headerBuffer, 4);

        if (isFinal)
            _finalChunkReceived = true;

        return true;
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        // Return 0 when final chunk is fully processed (simulate end-of-stream)
        if (_finalChunkReceived && _remainingBytesInChunk == 0)
            return 0;

        // Read next chunk header if current chunk is exhausted
        if (_remainingBytesInChunk == 0)
        {
            if (!ReadChunkHeader())
                return 0;
        }

        // Read up to the remaining bytes in the current chunk
        int bytesToRead = Math.Min(count, _remainingBytesInChunk);
        int bytesRead = _innerStream.Read(buffer, offset, bytesToRead);
        _remainingBytesInChunk -= bytesRead;

        return bytesRead;
    }

    // Implement required Stream members (disable unsupported operations)
    public override bool CanRead => true;
    public override bool CanSeek => false;
    public override bool CanWrite => false;
    public override long Length => throw new NotSupportedException();
    public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
    public override void Flush() => throw new NotSupportedException();
    public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
    public override void SetLength(long value) => throw new NotSupportedException();
    public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();

    // Override Dispose to avoid closing the inner stream
    protected override void Dispose(bool disposing)
    {
        // DO NOT call _innerStream.Dispose() - keep the underlying stream alive
    }
}

Usage for Receiver: #

• Wrap your original stream with ChunkedReadStream
• Read from it like a regular stream—when Read() returns 0, you know all chunks (including the final one) are processed

3. Critical Considerations for Mutexed Send/Receive #

Since your operations are mutually exclusive:

• Implement a state machine (e.g., Idle → Sending → Receiving → Idle) to enforce that send operations complete (including final chunk) before switching to receive mode, and vice versa.
• Use a lock or synchronization primitive to prevent cross-thread access to the stream during state transitions.
• Handle edge cases like partial chunk reads or network timeouts by resetting chunk state before resuming operations.

内容的提问来源于stack exchange，提问作者Roman Vaughan