ValidationPromise.scala

import scalaz._

object ValidationPromise extends Application {
  import scalaz.Scalaz._
  import scalaz.concurrent.Promise
  import scalaz.concurrent.Strategy

  // ------------------------------------------------
  //  Applicative composition of concurrent functions
  // ------------------------------------------------

  type ValidationFunction = String => Validation[String, String]

  // Composes n (independent) functions of type ValidationFunction to a composite function of type ValidationFunction
  // that executes its components concurrently depending on an implicitly defined scalaz.concurrent.Strategy.
  def comp(fs: ValidationFunction*)(aggr: (String, String) => String)(implicit cs: Strategy): ValidationFunction =
    (s: String) => {
      val promises = s.pure[List] <*> fs.toList ∘ (_.promise(cs)) ∘
        (kleisliFn[Promise, String, Validation[String, String]](_))
      val promise = promises.tail.foldLeft(promises.head) {
        (z, m) => m <*> z ∘ aggr.lift[PartialApply1Of2[Validation, String]#Apply].curried
      }
      promise.get
    }

  // Alternative to comp using MA#sequence (not used below)
  def comps(fs: ValidationFunction*)(aggr: (String, String) => String)(implicit cs: Strategy): ValidationFunction =
    (s: String) => {
      val xsPromise = (s.pure[List] <*> fs.toList ∘ (_.promise(cs)) ∘
        (kleisliFn[Promise, String, Validation[String, String]](_))).sequence
      val xPromise = xsPromise.map(xs => xs.tail.foldLeft(xs.head) {
        (z, m) => m <*> z ∘ aggr.curried
      })
      xPromise.get
    }

  // Creates a ValidationFunction that fails when created with "c" otherwise succeeds
  def f(t: String): ValidationFunction = (s:String) => {
    Thread sleep 3000
    if (t == "c") Failure(t) else Success(s + t)
  }

  // Combines two success values into one
  val c = (s1: String, s2: String) => "%s - %s" format (s1, s2)

  {
    // Creates a separate thread for each function (component)
    import scalaz.concurrent.Strategy.Naive

    // Compose 3 functions that succeed
    val compf = comp(f("a"), f("b"), f("d"))(c)

    // Takes approx. 3 seconds (parallel computation)
    compf("x") assert_≟ Success("xa - xb - xd")
  }

  {
    // Runs each function (component) in the caller thread
    import scalaz.concurrent.Strategy.Sequential

    // Compose 3 functions that succeed
    val compf = comp(f("a"), f("b"), f("d"))(c)

    // Takes approx. 9 seconds (sequential computation)
    compf("x") assert_≟ Success("xa - xb - xd")
  }

  {
    // Creates a separate thread for each function (component)
    import scalaz.concurrent.Strategy.Naive

    // Compose 2 functions that succeed and one that fails
    val compf = comp(f("a"), f("c"), f("d"))(c)

    // Takes approx. 3 seconds (parallel computation)
    compf("x") assert_≟ Failure("c")
  }
}