Feature Request: Add Stream.CreateConnectedStreams API

[geoffkizer]

Background and Motivation

It's often convenient to have a simple in-memory producer/consumer stream. We have a few variations of this in our test code, e.g. VirtualNetwork. You can also achieve this by using a loopback socket or a named pipe; however these approaches are unnecessarily difficult to use and the perf likely isn't great, since it's copying buffers through the OS.

Proposed API

namespace System.IO
{
  class Stream
  {
    (Stream stream1, Stream stream2) CreateConnectedStreams(int maxBufferSize = 16384, bool duplex = true);
  }
}

If duplex is true, then both streams can both read and write. If duplex is false, then stream1 is write-only and stream2 is read-only. Alternatively, we could have two separate methods for clarity.

We may want to have a distinguished type for the returned Streams, like ConnectedStream, just in case we ever want to add more APIs to it.

Usage Examples

    (Stream stream1, Stream stream2) = CreateConnectedStreams();
    stream1.Write(Encoding.UTF8.GetBytes("Hello world!\n"));
    byte[] buffer = new byte[4096];
    int bytesRead = stream2.Read(buffer);
    Console.WriteLine(Encoding.UTF8.GetString(buffer, 0, bytesRead));

[danmoseley]

I haven't used them but my understanding is that this is what Pipelines are for? Or you need to pass to API that expects Stream?

[davidfowl]

Right, Pipe is basically this and you can use AsStream to get the Stream from the pipe.

[stephentoub]

Or you need to pass to API that expects Stream?

Right, Stream is the lingua-franca of our APIs and is used everywhere. The various places Geoff refers to specifically includes a bunch of places we're testing Streams, e.g. wrapping SslStream around another Stream.

Pipe is basically this and you can use AsStream to get the Stream from the pipe.

Pipe is one-directional. To use Pipe here, you actually need two of them, then create four streams via AsStream, and then write a custom stream type that wraps two of them and maps the wrapper's reads to one and writes to the other, then creating two instances of those for each end.

[davidfowl]

Ah I see this is talking about an IDuplexPipe 😄

[geoffkizer]

Yeah, it's basically "new Pipe()" for Streams. And supports unidirectional or bidirectional.

[scalablecory]

This API doesn't allow for the scenario where I have a duplex stream and need to read the entire buffer before writing a response -- what do you think about returning a type with those capabilities?

[geoffkizer]

This API doesn't allow for the scenario where I have a duplex stream and need to read the entire buffer before writing a response -- what do you think about returning a type with those capabilities?

Because it doesn't have a way to EndWrite, short of Close/Dispose?

Yeah, I agree adding something like that would be interesting.

It would be nice if we had a common way to do this across NetworkStream, QuicStream, this Stream, etc...

[geoffkizer]

Do you have an API proposal?

[scalablecory]

This needs some play time, but how about I start us off with:

abstract class DuplexStream : Stream
{
    public abstract void ShutdownWrites();

    public static (DuplexStream, DuplexStream) CreateConnectedPair(int maxBufferSize = 16384);
}

class NetworkStream : DuplexStream {}
class QuicStream : DuplexStream {}

A better name would be useful as QuicStream might not be duplex.

[stephentoub]

I think there are essentially three ways this could be done:

1. Add a new abstract base class that derives from Stream and that our various existing streams inherit where appropriate (likely at least NetworkStream, SslStream, PipeStream, QuicStream), as suggested above. I agree the "duplex" naming isn't quite right, though I don't have a better suggestion at the moment.
2. Just add a new virtual method to Stream itself, ideally with a reasonable default implementation, e.g. just calling Dispose (if there's no good default behavior, then potentially also a Can/Is capability test property).
3. Add a mix-in via an interface that can be tested for, e.g. if (stream is IDuplexStream d) d.ShutdownWrites;