SE-0406: Backpressure support for AsyncStream

[FranzBusch]

@wadetregaskis

My lingering concern is the sharp edge that is multiple consumers - nothing is stopping someone from trying to do that with AsyncStream. I like things to be gracefully resilient to programmer error in proportion to how close to the stdlib they are. Since almost everyone uses stdlib & Foundation, there's a wide range of competencies (and experience levels). For that reason, it'd be nice if multiple consumers were reasonably supported even if the behaviour isn't configurable or ideal for every use case (or even any use case - even String has some parts with underwhelming performance, but overall it works well-enough for most folks).

@tcldr

Rx derivative libraries were faced with exactly the same issue, and they made their stock Source types (Subjects) multicast by default. There's nothing in the specification for a Subject that says they must be multicast, but in the interest of progressive disclosure, they seem to have decided that this was the most intuitive way forward. It's just now we take for granted that Subjects are multicast. (You can create a single-shot Subject if you need one for performance.)

Replying to both concerns raised with multiple consumer support. My goal with this proposal was to clarify the behaviour of the current Async[Throwing]Stream since the current implementation has diverging behavior between the throwing and non-throwing one. Furthermore, none of the two provide a true broadcast API which I think is what most users expect.

I do agree that there is a significant need for broadcasting of AsyncSequences because that is a very common pattern in applications; however, I am still of the opinion that this should be composed rather than baked into AsyncStream itself. Two reasons for this: First, implementing an efficient broadcasting algorithm is hard and would make the implementation of AsyncStream significantly more complex. Secondly, when our root asynchronous sequence type is always broadcasting it limits some of it's usability in scenarios where one does not want a broadcasted sequence.

Lastly, there is an interesting future direction here that we haven't discussed yet. Once ~Copyable and ~Escaping types land we have to revisit the sequence and asynchronous sequence protocols. This would allow us to spell out a type where the func makeAsyncIterator() becomes consuming and would enforce the unicast behaviour at compile time compared to the proposals runtime enforcement.