Akka stream helpers

akkastreamHelpers.scala

object FlowHelper {
  def mapAsyncUnorderedWithBoundedParallelism[A, B](parallelism: Int)(f: A => Future[B]): Flow[A, B] =
    Flow[A].section(OperationAttributes.inputBuffer(initial = parallelism, max = parallelism)) { sectionFlow =>
      sectionFlow.mapAsyncUnordered(f)
    }

  def mapAsyncWithBoundedParallelism[A, B](parallelism: Int)(f: A => Future[B]): Flow[A, B] =
    Flow[A].section(OperationAttributes.inputBuffer(initial = parallelism, max = parallelism)) { sectionFlow =>
      sectionFlow.mapAsync(f)
    }

  def mapAsyncWithOrderedSideEffect[A, B](f: A => Future[B]): Flow[A, B] = mapAsyncWithBoundedParallelism(1)(f)
}