akkling-streams

akkling_stream.fs

let foreachAsync parallelCount action =
    let asyncF =
        fun input ->
            async {
                do! action input
                return Akka.NotUsed.Instance
            }
    Flow.empty
    |> Flow.asyncMap parallelCount asyncF
    |> Flow.toMat Sink.ignore Keep.right

gistfile1.txt

    let graphFlow b sink =
        graph b {
            //block elements
            let! source = (Source.ofSeq rows).Log("Source")
            let! broadcast = Broadcast 2
            let! zip = ZipWith.create combineResults
            let! someFlow = (Flow.empty |> Flow.asyncMap 10 someFlow).Log("someFlow")
            let! anotherFlow = ..
            let! flow3 = ... 
            let! beforeSink = (Flow.empty
                              |> Flow.groupedWithin 40
                                     (TimeSpan.FromSeconds 10.0)).Log("Before sink")
            //glue the elements
            b.From broadcast =>> someFlow =>> anotherFlow =>> zip.In0 |> ignore
            b.From broadcast =>> flow3 =>> zip.In1 |> ignore
            b.From source =>> broadcast.In |> ignore
            b.From zip.Out =>> beforeSink =>> sink|> ignore
        }

    let sink = foreachAsync 1 writeFinalResult
    sink
    |> Graph.create1 graphFlow
    |> Graph.runnable
    |> Graph.run mat