build.sbt

scalaVersion := "2.11.4"

libraryDependencies ++= Seq(
  "com.typesafe.akka"      %% "akka-stream-experimental"    % "1.0-M1"
)

Main.scala

package main

import akka.stream.{ FlowMaterializer, MaterializerSettings }
import akka.stream.scaladsl.{Source,ForeachSink,FlowGraph,FlowGraphImplicits}
import akka.actor.ActorSystem
import scala.concurrent.{Await, ExecutionContext, Future}
import scala.concurrent.duration._
import scala.util.Try

object Main extends App {

  val input =  1 to 10000000  
  def streamDemo() = {
    implicit val system = ActorSystem("query")
    import system.dispatcher
    implicit val materializer = FlowMaterializer()
    val output = scala.collection.mutable.Stack[Int]()
    val f = Source(input).
      filter { x => (x % 100) == 0 }.
      map { _ * 37 }.
      filter { x => (x % 100) == 0 }.
      map { _ / 29 }.
      filter { x => (x % 100) == 0 }.
      foreach(output.push(_))

    f.onComplete { _ =>
      Try {
        println(s"pushed ${output.length} items")
        system.shutdown
      }
    }
    f
  }

  def normalDemo() = {
    val output = scala.collection.mutable.Stack[Int]()
    for {
      x <- input
      if (x % 100) == 0
      y = x * 37
      if (y % 100) == 0
      z = y / 29
      if (z % 100) == 0
    } output.push(z)
    println(s"pushed ${output.length} items")
  }

  def benchmark(what: String)(body: => Unit): Unit = {
    val start = System.currentTimeMillis
    println(s"${what}: start ${start}")
    body
    val end = System.currentTimeMillis
    println(s"${what}: stop  ${end}")
    println(s"${what}: took ${end - start} ms")
  }

  import scala.concurrent.ExecutionContext.Implicits.global
  benchmark("akka streams") {
    Await.result({
      streamDemo()
    }, 1 hour)
  }
  benchmark("for-comprehensions") {
    normalDemo()
  }
}