How to copy a stream to another stream only when destination consumed

Question

Similar to How do I copy the contents of one stream to another?

But my understanding of sourceStream.CopyTo(destStream) means it'll read the entire sourceStream from start to end (chunks or whatever) in order to copy it, and then the consumer goes back through and reads the stream again (its copy), resulting in O(2n) rather than O(n), right? And if destStream is a temporary copy (i.e. MemoryStream), then I will also end up loading the entire source stream into memory for each copy.

Is there a way to do it so that it's only copied as destStream is consumed/read?

Specifically, in .NET C# I need to make a copy of an input stream and write it to multiple "destinations" (via various helper libraries, some of which dispose of the stream they're given). The input could be very large, and is usually actually a FileStream, so I'd rather not load the entire file into memory when I can rewind it and buffer it from the disk.

Example Scenario:

void WriteToMany(Stream sourceStream, IEnumerable<ICanPutStream> destinations) {
    foreach(var endpoint in destinations) {
        // <-- I need to make a copy of `stream` here because...
        endpoint.PutStream(sourceStream); // ...some endpoints automatically dispose the stream
    }
}

If I make a copy before PutStream is called, it's going to read through the source stream. I can live with that, but if I copy it to a MemoryStream it also loads it into memory for each endpoint (with the added weirdness of trying to dispose of something that may/not be disposed already). Ideally it would only be during the internal workings of PutStream that the original stream gets copied/read.

Answer 1

Unless you can seek back to the beginning of the stream, you must copy the entire stream to memory to have multiple consumers. Otherwise the stream data is only available to first consumer.

If you have a seekable stream (like FileStream) and you want to pass it to multiple consumers without having it disposed, you could implement a Stream proxy which delegates all members to the underlying stream except for Dispose. It would look something like:

class StreamProxy : Stream
{
    private readonly Stream _stream;

    public StreamProxy(Stream stream)
    {
        if (stream == null) throw new ArgumentNullException(nameof(stream));
        _stream = stream;
    }

    protected override void Dispose(bool disposing)
    {
        //don't dispose inner stream
    }

    public override void Flush()
    {
        _stream.Flush();
    }

    public override long Seek(long offset, SeekOrigin origin)
    {
        return _stream.Seek(offset, origin);
    }

    public override void SetLength(long value)
    {
        _stream.SetLength(value);
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        return _stream.Read(buffer, offset, count);
    }

    public override void Write(byte[] buffer, int offset, int count)
    {
        _stream.Write(buffer, offset, count);
    }

    public override bool CanRead
    {
        get { return _stream.CanRead; }
    }

    public override bool CanSeek
    {
        get { return _stream.CanSeek; }
    }

    public override bool CanWrite
    {
        get { return _stream.CanWrite; }
    }

    public override long Length
    {
        get { return _stream.Length; }
    }

    public override long Position
    {
        get { return _stream.Position; }
        set { _stream.Position = value; }
    }
}

This way, each consumer can dispose their "copy" of the stream (instance of StreamProxy), without disposing the underlying stream. Once the consumer is done, seek the underlying stream back to the beginning and pass the proxy to another consumer.

Regarding your question: Is there a way to do it so that it's only copied as destStream is consumed/read? You could augment the above StreamProxy class so that it keeps track of its own position within the inner stream. StreamProxy will then be responsible, for each read operation, to seek the inner stream to the appropriate position and read the next chunk. This way, each consumer receives their own instance of StreamProxy, and can be reading from the inner stream at independent positions.

I can't see any advantage to this approach over the initially proposed StreamProxy, unless your consumers are running in parallel. If they are, you'll also need a synchronization mechanism in StreamProxy so that reads are not overlapped, since the inner stream can only be at one position at a time. This effectively serializes the consumers (which is an inherent limitation of starting with a single Stream and not copying its contents to memory), and makes the approach overall less efficient (unless there is a huge disparity between the read performance of the inner stream and the write performance of the consumers).

Your new read method might look like:

public override int Read(byte[] buffer, int offset, int count)
{
    lock(_stream)
    {
        //position the inner stream to end of last read (another consumer may have moved it)
        _stream.Seek(Position, SeekOrigin.Begin);

        //read the bytes, up to count
        var count = _stream.Read(buffer, offset, count);

        //update the next read position
        Position += count;

        return count;
    }
}

public override long Position{get;set;}

Answer 2

Stream.CopyTo has a buffer internally (81920 bytes, if you don't set it on the overload method). It's implementation is very simple, so you could just alter it and use it like so:

void ConsumeStream(Stream source, Stream destination, int bufferSize)
{
    byte[] buffer = new byte[bufferSize];
    int count;
    while ((count = source.Read(buffer, 0, buffer.Length)) != 0)
    {
        destination.Write(buffer, 0, count);
        //Other stuff
    }
}