c# - 如何创建可以转换流的流包装流

Question

var incomingStream = ...
var outgoingStream = ...

await incomingStream.CopyToAsync(outgoingStream);

上面的代码很简单，将传入流复制到传出流。两个流都被分块传输通过互联网传入/传出。

现在，假设我想用 Func<Stream,Stream,Task> 之类的东西转换流如果不读取所有数据，我将如何做到这一点。

因为我可以做到

var ms = new MemoryStream();
incomingStream.CopyTo(ms);

--- do transform of streams and seek
ms.CopyTo(outgoingStream)

但是那会把漏洞读入 ms，是否有任何内置的东西允许我从传入流中读取并写入一个新的流，它不会缓冲所有内容，而是只为缓冲数据保留一个小的内部流并且它不会在数据再次从流中读取之前从传入流中读取。

我想做的是:

    protected async Task XmlToJsonStream(Stream instream, Stream outStream)
    {
        XmlReaderSettings readerSettings = new XmlReaderSettings();
        readerSettings.IgnoreWhitespace = false;
        var reader = XmlReader.Create(instream, readerSettings);
        var jsonWriter = new JsonTextWriter(new StreamWriter(outStream));
        jsonWriter.WriteStartObject();

        while (await reader.ReadAsync())
        {
            jsonWriter.writeReader(reader);
        }
        jsonWriter.WriteEndObject();
        jsonWriter.Flush();
    }
    protected async Task XmlFilterStream(Stream instream, Stream outStream)
    {
        XmlReaderSettings readerSettings = new XmlReaderSettings();
        readerSettings.IgnoreWhitespace = false;
        var reader = XmlReader.Create(instream, readerSettings);
        var writer = XmlWriter.Create(outStream, new XmlWriterSettings { Async = true, CloseOutput = false })

        while (reader.Read())
        {
            writer.writeReader(reader);
        }


    }

但是我不知道怎么连接。

var incomingStream = ...
var outgoingStream = ...
var temp=...  
XmlFilterStream(incomingStream,temp);
XmlToJsonStream(temp,outgoingstream);

因为如果我使用 MemoryStream 作为临时文件，它会不会在最后将它全部存储在流中。寻找读取数据后再次丢弃数据的 at 流。

以上所有只是示例代码，缺少一些处理和查找原因，但我希望我能够说明我要做什么。能够基于设置在复制流、进行 xml 过滤和可选的将其转换为 json 之间即插即用。

Answer 1

流是字节的序列，因此流转换类似于Func<ArraySegment<byte>, ArraySegment<byte>>。 .然后您可以以流式传输的方式应用它:

async Task TransformAsync(this Stream source, Func<ArraySegment<byte>, ArraySegment<byte>> transform, Stream destination, int bufferSize = 1024)
{
  var buffer = new byte[bufferSize];
  while (true)
  {
    var bytesRead = await source.ReadAsync(buffer, 0, bufferSize);
    if (bytesRead == 0)
      return;
    var bytesToWrite = transform(new ArraySegment(buffer, 0, bytesRead));
    if (bytesToWrite.Count != 0)
      await destination.WriteAsync(bytesToWrite.Buffer, bytesToWrite.Offset, bytesToWrite.Count);
  }
}

它比那更复杂一点，但这是一般的想法。它需要一些逻辑来确保 WriteAsync写入所有字节；除了 transform 之外，通常还需要一个“刷新”方法方法，在源流完成时调用，因此转换算法有最后机会返回其最终数据以写入输出流。

如果您想要其他 的流，例如 XML 或 JSON 类型，那么最好使用 Reactive Extensions .