[Feature - Akka.Streams] ForEachAsync

[george-zubrienko]

Is your feature request related to a problem? Please describe.
Imagine a scenario: consume input, process in a stream, sink to an external destination (write to database, send an api request etc). In this case, sink will be performing operation that is best run as a Task<TResult>. But currently there is no Sink.ForEachAsync to facilitate that, so available options are:

• var mySink = Sink.ForEach<TIn>(async myVar => ...). This creates an async void delegate, which does not really behave as a loop where each interation does an async call. Also, exception propagation will not work, and IDEs/compiler will issue a warning for this kind of code.
• This workaround:

var mySink = Flow.Create<TIn>()
 .SelectAsync(...)
 .ToMaterialized(Sink.ForEach<TOut>(_ => { }), Keep.Right)
 .Named(...) 

Which works fine, but is not super elegant :)

OG Akka has this: https://doc.akka.io/docs/akka/current/stream/operators/Sink/foreachAsync.html

Describe the solution you'd like

private Task<TResult> MySinkLogic(..)
{
  ...
}

var mySink = Sink.ForEachAsync(4, myVar => MySinkLogic(myVar))

[ismaelhamed]

@george-zubrienko ForeachAsync is no fancier than this:

Flow.Create<T>()
    .SelectAsyncUnordered(parallelism, f)
    .ToMaterialized(Sink.Ignore<T>(), Keep.Right);

which you can use in the meantime to achieve the exact same behavior.

[F0b0s]

As i see we already have ForEachAsync there

Can you use it?

[F0b0s]

@george-zubrienko

[george-zubrienko]

@F0b0s I don't see than in pre-1.5 released packages, the code you referenced is in dev branch. As of, can this be used for the issue in question, I think yes, that is 1-1 implementation of foreachAsync from the Scala API I referenced in the description.

[F0b0s]

@george-zubrienko check it in 1.5.1 please