Rendezvous

build.sbt

scalaVersion := "2.9.2"

resolvers ++= Seq(
  "typesafe" at "http://repo.typesafe.com/typesafe/releases",
  "sonatype" at "http://oss.sonatype.org/content/repositories/releases/"
)

libraryDependencies <++= (scalaVersion) { scalaVersion =>
  Seq(
    "joda-time" %  "joda-time" % "2.1",
    "org.joda" %  "joda-convert" % "1.2" ,
    "com.typesafe.akka" % "akka-actor" % "2.0.1",
    "org.scalatest" %% "scalatest" % "[1.7,)" % "test"
  )
}

RendezvousSpec.scala

import org.scalatest._
import org.scalatest.matchers._

import org.joda.time.DateTime
import java.util.concurrent.Executors
import akka.dispatch._
import akka.util.Duration
import akka.util.duration._

class RendezvousSpec extends FlatSpec with ShouldMatchers {

  behavior of "rendezvous"

  implicit val ec = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(4))

  case class DurationSleeper(d: Duration) {
    def sleep(): Unit = Thread.sleep(d.toMillis)
  }
  implicit def durationToSleeper(d: Duration) = DurationSleeper(d)

  it should "work with concurrent.ops.future" in {

    import scala.concurrent.ops.future
    // TODO Scala 2.10 import scala.concurrent.future

    val start = DateTime.now

    val after5sec = future {
      5.seconds.sleep()
      5
    }
    val after3sec = future {
      3.seconds.sleep()
      3
    }
    val after4sec = future {
      4.seconds.sleep()
      4
    }

    Seq(after5sec, after3sec, after4sec).map(_.apply()).sum should equal(12)
    val spentMillis = DateTime.now.getMillis - start.getMillis
    spentMillis should be < (5100L)
  }

  it should "work with Akka" in {

    val start = DateTime.now

    val after5sec = Future {
      5.seconds.sleep()
      5
    }
    val after3sec = Future {
      3.seconds.sleep()
      3
    }
    val after4sec = Future {
      4.seconds.sleep()
      4
    }

    Seq(after5sec, after3sec, after4sec).map(f => Await.result(f, 10 seconds)).sum should equal(12)
    val spentMillis = DateTime.now.getMillis - start.getMillis
    spentMillis should be < (5100L)
  }

  it should "work with Akka onSuccess" in {

    val start = DateTime.now

    class SharedValue(var value: Int)
    var shared = new SharedValue(0)

    val after5sec = Future {
      5.seconds.sleep()
      1 
    } onSuccess { case res =>
      2.second.sleep()
      shared.synchronized { shared.value += res * 5 }
    }
    val after3sec = Future {
      3.seconds.sleep()
      1 
    } onSuccess { case res => 
      4.second.sleep()
      shared.synchronized { shared.value += res * 3 }
    }
    val after4sec = Future {
      4.seconds.sleep()
      1
    } onSuccess { case res => 
      3.second.sleep()
      shared.synchronized { shared.value += res * 4 }
    }
    val fs = Seq(after5sec, after3sec, after4sec)

    val result = fs.map(f => Await.result(f, 10 seconds)).sum
    result should equal(3)
    val spentMillis = DateTime.now.getMillis - start.getMillis
    spentMillis should be < (5100L)

    shared.value should equal(0) // onSuccess callback has not been finished yet.

    while (fs.find(f => !f.isCompleted).isDefined) {
      100.millis.sleep()
    }
    shared.value should equal(0) // onSuccess callback has not been finished yet.

    3.seconds.sleep()
    shared.value should equal(12)
  }


}

// vim: set ts=4 sw=4 et: