sourcebatch.scala

object TestMain extends App {
  import akka.stream.scaladsl._
  import akka.actor._
  import akka.stream.ActorMaterializer
  import scala.concurrent._
  import scala.concurrent.duration._

  implicit val system = ActorSystem()
  implicit val ec     = system.dispatcher
  implicit val mat    = ActorMaterializer()

  val future = Source(1 to 20000)
    .batch(max = 30, first => Seq(first)) { (batch, elem) =>
      batch :+ elem // 下流が遅いときに要素がまとめられる
    }
    .runWith(Sink.foreach { x =>
      Thread.sleep(4000)
      println("Out:" + x)
    })

  Await.result(future, Duration.Inf)
}