How to accomplish the equivalent of SelectManyAsync in a Flow

[seantempleton]

Hello. I am just starting into learning Akka.NET streams and am looking for help particularly with async/await and creating Sources and Flows.

My scenario is I will have one or more Source stages that need to emit multiple output elements of type N asynchronously, as they will be using async/await calls. Those will lead to a second Flow stage which will accept the input elements of type N and asychronously emit multiple output elements of type S for each input element. I currently have both of these operations implemented as IAsyncEnumerable functions.

1. I see RunAsAsyncEnumerable on Source and it sounds like that would work, but what if I have multiple Sources?
2. I would have thought there would be a SelectManyAsync Flow stage but I do not see one. Am I missing an easy way to transform one input type to a different type of many outputs asynchronously?

Thank you for any help you can provide!

[to11mtm]

I see RunAsAsyncEnumerable on Source and it sounds like that would work, but what if I have multiple Sources?

RunAsAsyncEnumerable is... a curious beast. But not what you would want to use. It's mostly useful for taking a Source<T> and getting back a single IAsyncEnumerable instance that yields when the stream doesn't have an element ready.

My scenario is I will have one or more Source stages that need to emit multiple output elements of type N asynchronously, as they will be using async/await calls. Those will lead to a second Flow stage which will accept the input elements of type N and asychronously emit multiple output elements of type S for each input element. I currently have both of these operations implemented as IAsyncEnumerable functions.

I would have thought there would be a SelectManyAsync Flow stage but I do not see one. Am I missing an easy way to transform one input type to a different type of many outputs asynchronously?

I think a lot of this depends on what you mean by 'asynchronusly emit multiple output elements of type S'.

Are you talking about a stage that Gets 1 T, Then doing something that may emit 0 or more elements of S, but you want each element to be pushed to a downstream as fast as possible? (It sounds like it but I want to confirm.) If so, Do the S elements come out as a group, or separately? (i.e. 1 T results in X Async operations producing Y distinct S elements)?

[seantempleton]

i.e. 1 T results in X Async operations producing Y distinct S elements

Yes, you've got it. The two stages, the first a Source, and the second a Flow connected to the Source, will each make one or more async/await calls in order to produce their output elements, and in each stage they will have distinct output elements passed one by one down to the next stage, not in a group.
They do not need to be pushed as fast as possible though, more likely they will need to wait for demand.

For example the Source stage could be initialized with 10 items to process. (Not calling them elements since the Source doesn't have an input, at least in a stream sense.) The Source stage will make async/await calls to emit its elements and might emit 5 elements of type N one at a time for each of its items. The second stage does the same thing, takes each of the previous 50 input elements of type N one at a time, make potentially multiple async/await calls for each input, and produce 10 more elements of type S per input to emit downstream one at a time.

I think I was able to figure out the Source part by using UnfoldAsync. I passed in an initial state of a custom class that has an iterator as a null property. The stage would initialize the iterator on the first call and use it on each invocation until the iterator ends.