dataflow.scala

import java.util.concurrent.Executor

import scala.collection.mutable
import scala.concurrent._
import scala.util.{Success, Failure}

implicit val synchronousExecutionContext = ExecutionContext.fromExecutor(new Executor {
  def execute(task: Runnable) = task.run()
})
 
object PriceService {
  val prices = mutable.HashMap[String, Int]()
  val pendingRequests = mutable.HashMap[String, Set[Promise[Int]]]() withDefaultValue Set()
 
  def setPrice(symbol: String, price: Int): Unit = {
    prices += (symbol -> price)
    println(s"${Thread.currentThread.getName}: price received: $symbol: $price")
    pendingRequests(symbol).foreach(_.success(price))
  }
 
  def price(symbol: String): Future[Int] = 
    if (prices.contains(symbol)) Future.successful(prices(symbol))
    else {
      val request = Promise[Int]()
      val updatedRequests = if (pendingRequests.contains(symbol)) pendingRequests(symbol) + request
                            else                                  Set(request)
      pendingRequests += (symbol -> updatedRequests)
      request.future
    }
}
 
def publishValuation(symbol: String, quantity: Int): Unit = 
  PriceService.price(symbol).map(_ * quantity).onComplete {
    case Success(value) => println(s"${Thread.currentThread.getName}: $quantity x $symbol: $value)")
    case Failure(e)     => println(s"error: $e")
  }
 
PriceService.setPrice("AAA", 1013)
publishValuation("AAA", 20)
publishValuation("BBB", 30)
PriceService.setPrice("BBB", 221)