How to test effects with multiple AsyncStreams

[chenf7]

Hi all,
I have a reducer with a .loadStuff action that returns a .run{} effect. Inside the run effect I have multiple AsyncStreams that I listen to. It looks something like:

await withTaskGroup(of: Bool.self) { taskGroup in
    taskGroup.addTask {
        for await a in aStream {
            await send(.setDataA(a))
        }
        return true
    }
    taskGroup.addTask {
        for await b in bStream {
            await send(.setDataB(b))
        }
        return true
    }
    ...
    ...

    await taskGroup.waitForAll()
}

It works great (and I also set the effect as cancellable for correct management).

However, when trying to test this it seems that there's no fixed order for the resulting actions. On my machine the test succeeds consistently, but on the CI it fails.

From what I see, all of the async tasks run on the same thread. My test function (in an XCTestCase ) does the following:

_ = await store.send(.loadStuff)
await store.receive(.setDataA(aData)) {
    $0.data.a = aData
}
await store.receive(.setDataB(bData)) {
    $0.data.b = bData
}
...
...

One option I can try is to break the current .run{} effect into multiple effects, but I prefer to keep it in one effect. Also, if there's a way to make .receive() wait only for a specific action and ignore the rest it would help.

Any suggestions?

[Nevwild]

I have faced the same issue, and have not seen a great solution. I would love to find a syntax for concatenating  effects that’s as easy as the old combine .concatenate {} publisher. Does anyone have any recommendations?Best,Nevill WilderOn Mar 20, 2023, at 09:56, Chen Frenkel ***@***.***> wrote: Hi all, I have a reducer with a .loadStuff action that returns a .run{} effect. Inside the run effect I have multiple AsyncStreams that I listen to. It looks something like: await withTaskGroup(of: Bool.self) { taskGroup in taskGroup.addTask { for await a in aStream { await send(.setDataA(a)) } return true } taskGroup.addTask { for await b in bStream { await send(.setDataB(b)) } return true } ... ... await taskGroup.waitForAll() } It works great (and I also set the effect as cancellable for correct management). However, when trying to test this it seems that there's no fixed order for the resulting actions. On my machine the test succeeds consistently, but on the CI it fails. From what I see, all of the async tasks run on the same thread. My test function (in an XCTestCase ) does the following: _ = await store.send(.loadStuff) await store.receive(.setDataA(aData)) { $0.data.a = aData } await store.receive(.setDataB(bData)) { $0.data.b = bData } ... ... One option I can try is to break the current .run{} effect into multiple effects, but I prefer to keep it in one effect. Also, if there's a way to make .receive() wait only for a specific action and ignore the rest it would help. Any suggestions? —Reply to this email directly, view it on GitHub, or unsubscribe.You are receiving this because you are subscribed to this thread.Message ID: ***@***.***>

[mbrandonw]

Hi @chenf7, how are aStream and bStream constructed? Ideally they are procured from a dependency that you control, in which case for tests they can be backed by an AsyncStream that you can feed data with a continuation. Then you have the ability to send one stream a piece of data, make your assertion than action was received, and then send the other stream a piece of data.

[chenf7]

Well, my streams are dependencies and in testing I do replace them with fake streams. However, the fake streams are set somewhere else in the code, so I had them set with a list of results to yield automatically. For some reason I didn't consider placing them in the tests themselves and controlling the order of emissions with explicit yields.
I'm going to try that, thanks!

[chenf7]

Ok, it solved the issue. I added control over the AsyncStream emmissions by initializing my dependency with PassthroughSubjects that are translated to AsyncStream with:

extension PassthroughSubject where Failure == Never {
    public func toAsyncStream() -> AsyncStream<Output> {
        return AsyncStream<Output>(bufferingPolicy: AsyncStream<_>.Continuation.BufferingPolicy.bufferingNewest(1)) { continuation in
            let cancellable = self.sink {
                continuation.yield($0)
            }
            continuation.onTermination = { @Sendable _ in cancellable.cancel() }
        }
    }
}

I then send to the subjects in my test before each action receive check.

[ylorn]

https://developer.apple.com/documentation/combine/publisher/values-1dm9r

[lukeredpath]

Ok, it solved the issue. I added control over the AsyncStream emmissions by initializing my dependency with PassthroughSubjects that are translated to AsyncStream with:

extension PassthroughSubject where Failure == Never {
    public func toAsyncStream() -> AsyncStream<Output> {
        return AsyncStream<Output>(bufferingPolicy: AsyncStream<_>.Continuation.BufferingPolicy.bufferingNewest(1)) { continuation in
            let cancellable = self.sink {
                continuation.yield($0)
            }
            continuation.onTermination = { @Sendable _ in cancellable.cancel() }
        }
    }
}

I then send to the subjects in my test before each action receive check.

You could also use the .streamWithContinuation helper to get subject like behaviour without needing Combine.