This is perfectly possible with a tuple, you don't have to destructure it and are free to store as an aggregate value in a single constant/variable.
let (stream, continuation) = AsyncStream.makeStream(...)
let aggregate = AsyncStream.makeStream(...)
OTOH, destructuring is not available for a concrete type and the first line in the snippet above would not compile.
At this point, I really would like to get somebody from the @language-workgroup to chime in and state what they prefer both in terms of usability and back-deployment.
It's a bit premature to be pinging the language workgroup for this change after 5 days of pitch over a weekend.
Something that might help is to refer to the test criteria for standard library additions that have been laid out in the past: is it non-trivially composed, helps avoid common pitfalls, significantly improves readability. This does meet these criteria AFAICT.
If the pitch is achieving consensus after a while, and the proposal is easy to implement, the next step is to put up an evolution proposal and an implementation as PRs against the appropriate repos. But I think the pitch needs longer to get to that stage.
Two concerns from me (not language group feedback, just my own):
The shape of this API is unusual. We don't normally put creation static methods on protocols. Usually initializers on concrete types are preferred. The need for an Element meta type in the signature is also a sign something is off here.
I'd like to see some scrutiny/discussion of the safety aspects of this design. Can it be misused in a way that leads to leaks or unexpected behavior? Personally I'm not familiar with the domain to be confident but maybe others can chime in (@Philippe_Hausler maybe)
Back deployment isn't something the language group generally comments on – that is a decision for the platform owner not evolution. But since this doesn't declare any new types, it can probably make use of the currently-in-review @backDeploy feature.
Thanks for chiming in here and sorry for the early ping!
We are in agreement here, I just misread your point. Sorry about that!
I would disagree on this point. I think this avoids a very common pitfall which is that it is not really clear to users that you can escape the Continuation from the closure based init. Much less is it clear that it safe to do so. Just from the feedback here in this thread in 5 days we see that a lot of people have created conveniences for this.
This proposal is not about adding static methods on any protocols. It is rather about extending the concrete types AsyncStream and AsyncThrowingStream with these two new convenience static methods. We cannot use an init for this since we need to return two separate types. The Element type in the parameters of the makeStream() method is just to make it easier for users to spell it and follows our spellings for task groups.
let (stream, continuation) = AsyncStream<Int>.makeStream()
can become this:
let (stream, continuation) = AsyncStream.makeStream(of: Int.self)
This is one of the goals that this proposal wants to achieve. It should be clear that actually escaping the Continuation is safe and with this new API we make this very clear. I have already talked with @Philippe_Hausler about this and would love if he can chime in here as well!
This is good to know since one of the open discussion points is a new concrete type vs the proposed tuple approach. I have read in a few places already that concrete types can't be back deployed and from your statement I guess that still stands. That would make the tuple more favourable in my opinion since this kind of convenience benefits a lot from being backdeployable.
AFAICT Ben said it fulfills those criteria so seems you are in agreement.
Oh you are right, I must have imagined a not in there. Thanks for clarifying that!
Since you mention that an alternative is a concrete wrapper type with both stream and continuation as properties, then would it be the case then that you could use init instead of a static creation method with that alternative?
Since platform owners and not the Swift Evolution process control what is back-deployed, I would be very hesitant to compromise on the proposed design of a feature on the basis of what can be back-deployed. It would set a very poor precedent to contort the design of permanent additions to the standard library on the basis of transitional bring-up matters that are not even within the full control of the Swift project.
Yeah, we could but that is super hard to discover in my opinion. It would end up being something like this:
let newStream = AsyncStream<Int>.StreamWithContinuation()
I will add it to the alternatives considered section though since it is worth mentioning!
I think that's a fair point! Thanks for bringing it up.
I went with the tuple because I heard concerns about backdeployability of the concrete type but I agree that we shouldn't compromise because of individual platforms ability here.
Couldn't it be (StreamAccessors being a struct with the continuation/stream) something like:
let newStream: AsyncStream.Accessors = AsyncStream<Int>()
extension AsyncStream {
struct Accessors {
let stream: AsyncStream<Element>
let continuation: AsyncStream<Element>.Continuation
}
public init(
of elementType: Element.Type = Element.self,
bufferingPolicy limit: Continuation.BufferingPolicy = .unbounded
) -> AsyncStream.Accessors {
let storage: _Storage = .create(limit)
let stream = AsyncStream<Element>(storage: storage)
let continuation = Continuation(storage)
return AsyncStream.Accessors(stream: stream, continuation: continuation)
}
}
(perhaps I bungled some syntax, but basically a scoped subtype to e.g. AsyncStream)
That doesn't work because inits cannot have return types since they are always returning Self implicitly.
Oh, right. Sorry for the noise, then it's basically makeXXX.
I guess we could put the init on that struct type instead and have it create the stream there instead?
Still out on holiday for a bit, but hopping back over here to chime in - This API is quite useful. I have seen variants of this being used pretty extensively and for good reason too. When originally designing AsyncStream this was one of the considerations early on but was dropped because of the slight oddity of the construction. Retrospectively looking at it, this perhaps was the more general way of approaching the problem; the only advantage that the closure gives is isolation of where the Continuation is scoped to.
All in all I think this is a good fit, a good tool, and helps developers avoid confusion about the escapes of the continuation and usages of force unwrapping. Which for the cases it is useful, quite honestly it is the right tool for the job.
However... there is one danger zone that is easily tempting with the API to go down; that of the structure versus tuple. Personally I am of the belief that 9 times out of 10 (or more) tuples are the wrong choice for the job and structures are better picks because of extensibility and general ergonomics. This API is one of the exceptions to that belief. The reason being is that it is very tempting to then store that type somewhere for vending out to multiple consumers. This immediately becomes a bit of a sticky situation per the iteration on multiple tasks. Being a tuple it isolates the concerns of each item and infers the resultant is for independent consumption; which in my view aides in the use of the API that results in expected behaviors.
Keeping it as a tuple return reinforces the construction nature of things and more importantly isolates the concerns of sending and the sequence part.
I also feel that back deploying this is reasonable and warranted - it prevents folks from writing wonky code that might be error prone and offers an ergonomic that as I said before perhaps should have been there from the start.
Seeing as we're now escaping the closure anyway, is there a reason we couldn't adapt AsyncStream to accept a Continuation – or more likely a type that wraps the continuation – as a parameter for a new initialiser on AsyncStream:
AsyncStream.init(_ source: AsyncStreamSource<Element>)
With AsyncStreamSource having essentially the same API as the AsyncStream's Continuation type.
The one issue I see is that you could initialise multiple AsyncStreams with the same storage, but that would simply be programmer error, not dissimilar to attempting to iterate an AsyncStream from multiple tasks.
API wise it feels a bit cleaner then the tuple setup, and while it might take a bit of juggling under the hood, it shouldn't impact any existing APIs.
Hi,
As we explore some alternate solutions regarding the ergonomics of the API, I’d like to mention the case of AsyncChannel in Swift async algorithms. This type acts has a communication bus between tasks, by allowing some to send and some others to iterate over it. AsyncChannel conforms to AsyncSequence and offers a suspending function to send elements (and a non suspending one to finish).
I feel that what we are trying to achieve by giving a « factory » function to AsyncStream that returns both a « sender » type and an AsyncSequence is very close to the notion of Channel.
In the end I fear that the coexistence in the same code base of AsyncChannel and AsyncStream, each one offering a different instantiation and usage mechanism will be confusing.
let channel = AsyncChannel<Int>()
await channel.send(1)
…
for await element in channel { … }
let (continuation, stream) = AsyncStream<Int>.makeStream()
continuation.yield(1)
…
for await element in stream { … }
IMO both are made to make tasks communicate but offer a different back pressure management.
Shouldn’t we try to harmonise that to improve the discoverability and readability of the API ?
One thing that I think is important in this conversation is the ability for developers to express which types or scopes have the responsibility to send events on a stream vs merely receiving them. It makes code much easier to reason about.
One advantage of separating the continuation from the stream is that it allows developers to bestow this responsibility. Passing a channel around, we’d have to type erase / lean on existentials to make the sending/receiving responsibility clear, right?
Maybe @twittemb’s onto something here. Maybe async channels need to live in the standard library and async streams be created from channels. Then need for an escaping continuation starts to disappear.
Yes I think we should pass the AsyncSequence protocol when only consumption is expected, this is the read-only version of the AsyncChannel (the primary associated type should help, when implemented)
Thanks for all the feedback and healthy discussions!
In my opinion, this is too dangerous and too easy to misuse. As you already pointed out, you could then pass the Continuation/Source to many AsyncStream.inits and that would not be okay. I think we shouldn't try to change the world here, but rather just make the AsyncStream creation more convenient.
Just going to pick this one quote here but replying to all that you said. I disagree with the statement. AsyncStream and AsyncChannel are catering to two very separate use-cases. AsyncStream allows sync-to-async bridges to be build like bridging delegate based APIs (while giving some means of back pressure). Of course it also supports an async-to-async bridge. AsyncChannel on the other hand is only allowing async-to-async communication (by the fact that func send() is async.
On the general notation of separating the producing interface of a root AsyncSequence e.g. AsyncStream.Continuation or NIOAsyncSequenceProducer.Source from the consuming i.e. AsyncSequence is in my opinion good and makes it easier to implement things like early termination. However, I really believe this needs to be evaluated on a type by type basis and is not something that this proposal aims to unify. It is something that should be discussed in the AsyncChannel review.
I don't think AsyncChannel should live in the standard library. Like with many other things that have been discussed on this forums the stdlib should not become a place to just put everything. Having important things live in separate packages is IMO good.
I agree with the general notion that only the required interface should be passed when only that one is needed. However, I don't think passing an AsyncSequence protocol is the right approach. First, it is currently not possible since AsyncSequence hasn't adopted primary associated types right now. If it does we should not arbitrarily pass any AsyncSequence<Foo> everywhere since that has performance implications due to the existential. In a lot of cases, having a more concrete AsyncSequence implementation that is exposed is better for both performance and evolvability of the API. What I have recommend in a few places already is to create a concrete type like FooAsyncSequence and back it with something like AsyncStream. This allows you to switch the underlying type without breaking API.
Hi! Long time listener, first time caller 
Thank you for the proposal. I've been reaching for AsyncStreams frequently the last few months and I think the original API shape led to a lot of confusion at first because it seemed wrong to escape the continuation, but that is actually okay in this case. While I do agree that this proposal makes the instructions for using this API much clearer, I do worry about the safety.
By making continuation part of the return type, it signals that it is meant to be used, but in doing so, I think it also clouds the understanding that the continuation is meant to be the (singular) other end of the AsyncStream "pipe". So I worry that this piece of the proposal would inadvertently suggest to the reader that they should be able to pass that continuation around to whoever wants it...turning a single entry/exit point into a multi-entry/single exit point.
One thing that I've done in my usage of AsyncStream is to effectively create a stateful factory of AsyncStreams that shields both the producer and consumer of the stream of the idea of a continuation. I don't recall who gave me this inspiration but this wasn't my idea.
public actor AsyncStreamManager<T: Sendable> {
private var continuations: [UUID: AsyncStream<T>.Continuation]
public init() {
self.continuations = [:]
}
private func insert(continuation: AsyncStream<T>.Continuation) {
let uuid = UUID()
continuations[uuid] = continuation
}
public func send(_ value: T) {
continuations.values.forEach { continuation in
continuation.yield(value)
}
}
public func finish(_ value: T) {
send(value)
continuations.values.forEach { continuation in
continuation.finish()
}
continuations.removeAll()
}
public func makeStream() -> AsyncStream<T> {
let stream = AsyncStream { continuation in
insert(continuation: continuation)
}
return stream
}
}
I wonder if a similar idea should be employed as part of the AsyncStream API? The continuation is ancillary but a bit distracting to the purpose of AsyncStream which is to consume values over time. If we can help developers by not needing to worry about the medium by which they can publish their values, I think it would go a long way to making AsyncStream safer.
All that said, I'm by no means a language expert, just a guy with a couple thoughts.
The continuation of an AsyncStream is thread-safe (hence Sendable) which is partly why the original API seemed so restrictive. So effectively an AsyncStream is multi producer – but unfortunately not multi-consumer.