Как записать поток .NET в два других потока одновременно без буферов?

Question

В моем предыдущем вопросе кто-то прокомментировал, что возможно записать поток .NET в два других потока одновременно. Я мог найти только один способ, но используя временную переменную (буфер), которая будет хранить содержимое (полностью или частично сразу) входного потока.

1. Есть ли способ сделать это без использования буферов?
2. Есть ли способ сделать это одновременно?

Answer 1

Используя буфер, вы можете выполнять асинхронную запись в оба потока, предполагая, что потоки поддерживают правильный асинхронный доступ. Это эффективно одновременно - хотя у вас будет сложная работа по поддержанию синхронизации всех буферов, учитывая, что один может закончить запись первого буфера задолго до того, как другой сделает это.

Насколько я знаю, нет способа скопировать из одного (общего) потока в другой без использования буфера, по крайней мере, одного байта. (И использование одного байта, конечно, будет крайне неэффективно.)

Answer 2

Если вы готовы ограничить ваш параллелизм скоростью самого медленного потока, то это довольно просто и не требует дополнительного буфера, я записал его из памяти, поэтому, возможно, немного испортил его.

Кроме того, он не реализует все необходимые переопределения, такие вещи, как CanRead, CanSeek и т. П., Все пропущены для простоты, только основная параллельная запись выполняется.

using System;
using System.IO;
using System.Threading;

namespace MultiCastStream
{   
  public class MultiWriteStream : Stream
  {
    private readonly Stream[] streams;      
    private AutoResetEvent[] waits;
    private readonly IAsyncResult[] results;

    public MultiWriteStream(params Stream[] streams)
    {
      this.streams = (Stream[])streams.Clone();
      this.results = new IAsyncResult[this.streams.Length];
    }

    private void prepWaits()
    {
      if (waits == null)
      {
        this.waits = new AutoResetEvent[this.streams.Length];
        for(int i= 0; i < this.waits.Length; i++)
        {
          this.waits[i] = new AutoResetEvent(false);
        }
      }
      else
      {
        for(int i= 0; i < this.waits.Length; i++)
        {
          this.waits[i].Reset();
        }
      }
    }

    public override void Write (byte[] buffer, int offset, int count)
    {
      prepWaits();
      for( int i = 0; i < this.streams.Length; i++)
      {
        this.results[i] = streams[i].BeginWrite(
          buffer, offset, count, this.Release, waits[i]);   
      }
      WaitHandle.WaitAll(waits);
      for( int i = 0; i < this.streams.Length; i++)
      {
        this.results[i] = this.streams[i].EndWrite(results[i]); 
      }
    }

    private void Release(IAsyncResult result)
    {
      ((AutoResetEvent)result.AsyncState).Set();
    }

    public override void WriteByte (byte value)
    {
      // no point doing this asynchronously
      foreach (Stream s in this.streams) 
      s.WriteByte (value);
    }

    protected override void Dispose (bool disposing)
    {
      base.Dispose (disposing);
      if (this.waits != null)
      {
        foreach (AutoResetEvent w in this.waits)
          w.Close();
      }
      foreach (Stream s in this.streams)
        s.Dispose();
    }       
  }
}