[BUG] attempt to await next() on more than one task

[CyonAlexRDX]

Describe the bug
When using .multicast I got unexpected error: attempt to await next() on more than one task. It is my - maybe incorrect? - understanding that I could use several tasks consuming an Async Multicasted sequence?

To Reproduce
I have these variables:

actor Connection {
 private let asyncStream: AsyncThrowingStream<MySendableStruct, Swift.Error>
 private let asyncContinuation: AsyncThrowingStream<MySendableStruct, Swift.Error>.Continuation
 private let multicastSubject = AsyncThrowingPassthroughSubject<MySendableStruct, Swift.Error>()

 ...
 // using this static func: https://github.com/pointfreeco/swift-composable-architecture/blob/53ddc5904c065190d05c035ca0e4589cb6d45d61/Sources/ComposableArchitecture/Effects/ConcurrencySupport.swift#L222-L228
 (asyncStream, asyncContinuation) = AsyncThrowingStream.streamWithContinuation()
 ...

func autoconnectingMulticastedAsync() -> AsyncMulticastSequence<AsyncThrowingStream<MySendableStruct, Swift.Error>, AsyncThrowingPassthroughSubject<MySendableStruct, Swift.Error>> {
 asyncStream
 .multicast(multicastSubject)
 .autoconnect()
 }
}

and callsite

...
func connect() async throws {
 connectTask?.cancel()
 connectTask = Task {
 for try await value in autoconnectingMulticastedAsync() {
 guard !Task.isCancelled else { return }
 // do stuff with `value` that in some case might trigger a new nested `Task`
 ...
 }
}
..

Provide code snippets, or links to gist or repos if possible.

Expected behavior
I would expect to be able to await values in multiple tasks from a multicasted autoconnected Async Seqyuence.

Screenshots
If applicable, add screenshots to help explain your problem.

Environment:
Mac Studio (M1), running macOS Ventura: 13.0.1 (22A400)
Xcode Version 14.1 (14B47b)
AsyncExtension version 0.5.1

Additional context
Add any other context about the problem here.

[twittemb]

Hi,

Perhaps it as to do with the fact that an AsyncStream, once it is cancelled, cannot be iterated again. Perhaps try to replace it with an AsyncThrowingBufferedChannel?

[twittemb]

Ultimately you are right … the multicast operator should support several tasks no matter the upstream sequence … I’ll look into it

[CyonAlexRDX]

@twittemb Im sorry I did not give you a reproducible example, I have a quite large and complex code base and Im novice at Structured Concurrency so it is a bit hard for me to give a small repro code snippet.

Thanks for an amazing project and looking forward to your Broadcast PR in AsyncAlgorithm repo to be merged!

Maybe I should use share instead of multicast..? Hmm

[twittemb]

Thanks a lot.

Share is just a shortcut for multicast + AsyncPassthroughSubject. You’ll get the same result.

I’ll take a look as soon as possible to your usecase

[twittemb]

Could you test the new version 0.5.2, it should have fixed that?

thanks.

[CyonAlexRDX]

Thanks! Testing it now, will report back results soon!

[CyonAlexRDX]

I did not manage to reproduce the bug with neither 0.5.1 nor 0.5.2 now when using AyncBufferedChannel.

[twittemb]

Did you try with keeping an AsyncStream ?