conditionalVia.scala

import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Flow, Sink, Source}

object ConditionalVia extends App {

  implicit val system = ActorSystem()
  implicit val materializer = ActorMaterializer()

  def conditionalVia[E](predicate: E => Boolean, flow: Flow[E, E, NotUsed]): Flow[E, E, NotUsed] =
    Flow[E]
      .flatMapConcat {
        case elem if predicate(elem) => Source.single(elem).via(flow)
        case elem => Source.single(elem)
      }

  Source(1 to 10)
    .via(conditionalVia[Int](_ % 2 == 0, Flow[Int].map(_ * 2)))
    .via(conditionalVia[Int](_ % 3 == 0, Flow[Int].map(_ * 3)))
    .runWith(Sink.foreach(println))
}