2015-05-27-Akka-Streams-Balancing-Buffer.scala

#!/usr/bin/env scalas
 
// NOTE: This is a self-encapsulated Scala script meant to be run with scalas
// See http://www.scala-sbt.org/0.13/docs/Scripts.html
 
/***         
scalaVersion := "2.11.6"
 
resolvers += Resolver.url("typesafe-ivy-repo", url("http://typesafe.artifactoryonline.com/typesafe/releases"))(Resolver.ivyStylePatterns)

// Common dependency versions
val akkaVersion = "2.3.10"
val akkaHttpVersion = "1.0-RC2"
 
libraryDependencies ++= Seq(
  "com.typesafe.scala-logging" %% "scala-logging" % "3.+",
  "com.typesafe.akka" % "akka-stream-experimental_2.11" % "1.0-RC3",
  "org.scalatest" %% "scalatest" % "2.2.4"
)

*/
 
import com.typesafe.scalalogging.StrictLogging
import akka.actor.ActorSystem
import akka.stream._
import akka.stream.stage._
import akka.stream.scaladsl.FlowGraph.Implicits._
import akka.stream.scaladsl._
 
import org.scalatest._
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.prop.TableDrivenPropertyChecks._
import org.scalactic.Tolerance._
import org.scalactic.TripleEqualsSupport.Spread
 
import scala.collection.immutable._
 
 
def trailingDifference(offset: Int) = Flow() { implicit b =>

  val bcast = b.add(Broadcast[Int](2))
  val drop = b.add(Flow[Int].drop(offset))
  val zip = b.add(Zip[Int, Int])
  val diff = b.add(Flow[(Int, Int)].map { 
    case (num, trailing) => (num, num - trailing) 
  })

  bcast         ~> zip.in0
  bcast ~> drop ~> zip.in1
                   zip.out ~> diff

  (bcast.in, diff.outlet)
}
 
def trailingDifference2(offset: Int) = Flow() { implicit b =>

  val bcast = b.add(Broadcast[Int](2))
  val buffer = b.add(Flow[Int].buffer(offset, OverflowStrategy.backpressure))
  val drop = b.add(Flow[Int].drop(offset))
  val zip = b.add(Zip[Int, Int])
  val diff = b.add(Flow[(Int, Int)].map { 
    case (num, trailing) => (num, num - trailing) 
  })

  bcast ~> buffer ~> zip.in0
  bcast ~> drop   ~> zip.in1
                     zip.out ~> diff

  (bcast.in, diff.outlet)
}
 
// Create/destroy an actor system for testing
trait AkkaStreamsImplicits extends BeforeAndAfterAll { this: Suite =>
 
  implicit var system: ActorSystem = _
  implicit var materializer: FlowMaterializer = _
 
  override def beforeAll = {
    super.beforeAll
    system = ActorSystem(this.getClass.getSimpleName.replace("$", "_"))
    materializer = ActorFlowMaterializer()
  }
 
  override def afterAll = {
    system.shutdown
    super.afterAll
  }
 
}

trait TrailingDifferenceTable {
  val tests = Table(
    ("offset", "input", "expected"),
    (
      3, 
      (100 to 1 by -1).map { n => (n * n) },
      (100 to 1 by -1).dropRight(3).map { n => (n * n, (2*3) * n - (3*3)) }
    ),
    (
      10, 
      (20 to 1 by -1).map { n => (n * n) },
      (20 to 1 by -1).dropRight(10).map { n => (n * n, (2*10) * n - (10*10)) }
    )
  ) 
}

class TrailingDifferenceSpec extends FlatSpec with AkkaStreamsImplicits with TrailingDifferenceTable with Matchers with ScalaFutures {
 
  "Trailing difference" should "be properly calculated for all scenarios" in {
    forAll(tests) { (offset: Int, input: Seq[Int], expected: Seq[(Int, Int)]) =>
 
      val future = Source(input).via(trailingDifference(offset)).runWith(Sink.fold(List[(Int, Int)]())(_ :+ _))

      // Source(100 to 1 by -1).via(trailingDifference(offset)).runWith(Sink.foreach(println))
 
      whenReady(future) { result =>
        //println(s"result: $result")
        result should have size expected.size
        result.zip(expected).foreach { case (actual, expected) =>
          actual shouldBe expected
        }
      }
    }
  }
 
}
 
class TrailingDifference2Spec extends FlatSpec with AkkaStreamsImplicits with TrailingDifferenceTable with Matchers with ScalaFutures {
 
  "Trailing difference" should "be properly calculated for all scenarios" in {
    forAll(tests) { (offset: Int, input: Seq[Int], expected: Seq[(Int, Int)]) =>
 
      val future = Source(input).via(trailingDifference2(offset)).runWith(Sink.fold(List[(Int, Int)]())(_ :+ _))

      Source(100 to 0 by -2).via(trailingDifference2(offset)).runWith(Sink.foreach(println))
 
      whenReady(future) { result =>
        //println(s"result: $result")
        result should have size expected.size
        result.zip(expected).foreach { case (actual, expected) =>
          actual shouldBe expected
        }
      }
    }
  }
 
}
 
// Run the test case
run(new TrailingDifference2Spec)