Is this extension safe?
extension AsyncStream {
static func make() -> (AsyncStream<Element>, AsyncStream<Element>.Continuation) {
var continuation: AsyncStream<Element>.Continuation?
let asyncStream: AsyncStream<Element> = AsyncStream {
continuation = $0
}
return (asyncStream, continuation!)
}
}
Do we have strong guarantees that the build closure will be called during the init of the AsyncStream?
The existing API (continuation passed to the build closure) is not suitable for my use case. I have an Actor that has a factory method that returns an AsyncStream powered by a Task and I need to tie that Task (and stream) to the lifecycle of the Actor (store it in the actor so I can cancel it in the deinit). I can't add the task from the sync AsyncSequence callback closure because I can't await to execute the function in the actor from there.
I would use it like this:
actor WordService {
private var userTasks: [String: Task<Void, Never>] = [:]
func words(userID: String) -> AsyncStream<String> {
let (asyncStream, continuation) = AsyncStream<String>.make()
let task = Task { [weak self, continuation] in
// Get the words from somewhere and yield them when available
continuation.yield("some word")
continuation.yield("some other word")
}
continuation.onTermination = { _ in
task.cancel()
}
userTasks[userID]?.cancel()
userTasks[userID] = task
return asyncStream
}
deinit {
userTasks.values.forEach { task in
task.cancel()
}
}
}
Having an extension like the above one would allow me to implement what I need to implement without having to deal with optionals for the continuation.
Is the extension safe? Are there better ways to implement my use case?