Pipeline example

Pipeline.scala

import play.api.libs.json._
import play.api.libs.json.extensions._

/**
 * Like a function that takes an A as input and produces a B as output.
 * But we make a special type so we can define our own ++, map and
 * flatMap operations.
 */
trait Pipeline[-A,+B] {
  original =>
  def process(a: A): B

  /**
   * Make a new Pipeline from two parts by joining two
   * Pipelines together.
   */
  def ++[B1>:B,C](following: Pipeline[B1,C]): Pipeline[A,C] = new Pipeline[A,C] {
    def process(a: A): C = {
      val b = original.process(a)
      val c = following.process(b)
      c
    }
  }

  /**
   * Make a new Pipeline from an existing one by transforming
   * the Pipeline's output with a function. The map operation
   * is special and can be used in Scala's for comprehensions.
   */
  def map[C](f: B => C): Pipeline[A,C] = new Pipeline[A,C] {
    def process(a: A): C = {
      val b = original.process(a)
      val c = f(b)
      c
    }
  }

  /**
   * Make a new Pipeline from an existing one by transforming
   * the Pipeline's output into a new Pipeline, then calling
   * that new Pipeline with the output of the first Pipeline...
   * The flatMap operation is special and can be used in Scala's
   * for comprehensions.
   */
  def flatMap[B1>:B,C](f: B => Pipeline[B1,C]): Pipeline[A,C] = new Pipeline[A,C] {
    def process(a: A): C = {
      val b = original.process(a)
      val bPipeline = f(b)
      val c = bPipeline.process(b)
      c
    }
  }
}

object Pipeline {
  /**
   * Helpful constructor for making Pipelines from functions.
   * (See CatDog.dogPipeline for an example)
   */
  def apply[A,B](f: A => B): Pipeline[A,B] = new Pipeline[A,B] {
    def process(a: A): B = f(a)
  }
  /**
   * A Pipeline that ignores its input and always produces the
   * same output.
   */
  def const[A](a: A): Pipeline[Any,A] = new Pipeline[Any,A] {
    def process(ignored: Any): A = a
  }
}

object CatDog {

  case class Cat(name: String)
  case class Dog(name: String)

  /**
   * Create a Pipeline which inserts a Cat's name into the
   * JS.
   */
  def catPipeline(cat: Cat) = new Pipeline[JsValue,JsValue] {
    def process(js: JsValue): JsValue = {
      js.set(
        (__ \ "cat" \ "name") -> JsString(cat.name)
      )
    }
  }

  /**
   * Create a Pipeline which inserts a Dog's name into the
   * JS. We use the Pipeline.apply() helper method to create
   * a Pipeline from a function.
   */
  def dogPipeline(dog: Dog) = Pipeline { (js: JsValue) =>
    js.set(
      (__ \ "dog" \ "name") -> JsString(dog.name)
    )
  }

}

object Tester {
  def main(args: Array[String]): Unit = {

    // Your example converted to use a Pipeline
    {
      import CatDog._
      val input = Json.obj()
      val composedPipeline = catPipeline(Cat("Steven")) ++ dogPipeline(Dog("Rex"))
      val output = composedPipeline.process(input)
      println(output)
      // PRINTS: {"cat":{"name":"Steven"},"dog":{"name":"Rex"}}
    }

    // Example that demonstrates building a Pipeline using
    // for comprehension syntax
    {
      import CatDog._
      val input = Json.obj()
      val pipeline = for {
        catJs <- catPipeline(Cat("Steven"))
        dogJs <- dogPipeline(Dog("Rex"))
        expandedJs <- Pipeline.const(Json.arr(catJs, dogJs))
      } yield expandedJs
      val output = pipeline.process(input)
      println(output)
      // PRINTS: [{"cat":{"name":"Steven"}},{"cat":{"name":"Steven"},"dog":{"name":"Rex"}}]
    }

  }
}