Csharp 在 C# Stream 中延迟转换数据

只需要实现Stream.Read个方法:

• 以区块为单位从源流读取
• 在每个区块后跳过2个字节

using System.Buffers;
using System.Text;

var sourceString = string.Concat(
    Enumerable.Range(1, 10).Select(_ => "Foo \r\nBar \r\nBaz!\r\n"));
Console.WriteLine("Source: " + sourceString);

var encoding = new UTF8Encoding(encoderShouldEmitUTF8Identifier: false);
var sourceBytes = encoding.GetBytes(sourceString);

using var stream = new MemoryStream(sourceBytes);
using var filter = new CrLfFilteringStream(stream, 4);
using var reader = new StreamReader(filter, encoding);

var res = reader.ReadToEnd();
Console.WriteLine("Result: " + res);

public class CrLfFilteringStream : Stream
{
    private readonly Stream _stream;
    private readonly int _chunkSize;
    private readonly byte[] _chunk;
    private int _chunkPosition;
    private int _chunkLength;

    public CrLfFilteringStream(Stream stream, int chunkSize)
    {
        _stream = stream ?? throw new ArgumentNullException(nameof(stream));
        _chunkSize = chunkSize;
        _chunkPosition = chunkSize;
        _chunkLength = chunkSize;
        _chunk = ArrayPool<byte>.Shared.Rent(chunkSize);
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        var toRead = count;
        var bufferPosition = 0;
        Span<byte> sink = stackalloc byte[2];

        while (toRead > 0 && _chunkLength > 0)
        {
            if (_chunkPosition >= _chunkSize)
            {
                _chunkPosition = 0;
                _chunkLength = _stream.Read(_chunk, 0, _chunkSize);

                // Skip CR LF.
                _stream.Read(sink);
            }

            var currentRead = Math.Min(_chunkLength, toRead);
            Array.Copy(_chunk, _chunkPosition, buffer, bufferPosition, currentRead);
            toRead -= currentRead;
            bufferPosition += currentRead;
            _chunkPosition += currentRead;
        }

        return count - toRead;
    }

    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();
    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();
    }

    protected override void Dispose(bool disposing)
    {
        ArrayPool<byte>.Shared.Return(_chunk);

        base.Dispose(disposing);
    }
}

这段代码从ArrayPool租用一个区块大小的缓冲区，并且不分配任何其他内容(除了CrLfFilteringStream个实例).