Akka-stream-slide

Flows.scala

val source = Source(1 to 10)

val sink = Sink.fold[String, String]("")(_ + _)

val flow = Flow[Int]
               .filter(_ % 2 == 0)
               .map(_.toString)
               
val result = source.via(flow).runWith(sink)    //Future[String]

Await.result(result, 100.millis) //246810

Partial.scala

source.via(measure).via(anotherFlow).runWith(sink)
    
val measure = 
    Flow.fromGraph( FlowGraph.create { implicit builder =>
      
      val conflate = builder.add(
        Flow[Int].conflate(_ => 1)((sum, _) => sum + 1)
      )

      val zip = builder.add(
        ZipWith[Int, Unit, Int]((t, _) => t)
          .withAttributes(OperationAttributes.inputBuffer(initial = 1, max = 1)))

      val tickr = builder.add(
        Source(0.seconds, 50.millis, ())
      )
      
      val logger = builder.add(
        Flow[Int].map { counter => println("TOTAL: "+ counter); counter}
      )

      conflate.outlet    ~>    zip.in0
         tickr.outlet    ~>    zip.in1
                               zip.out ~> logger.inlet

      FlowShape(conflate.inlet, logger.outlet)
    }    
)