Pipeline

Pipeline.scala

trait PipelineBuilder {
  def createStage[I, O](f: I => O) : Stage[I, O]

  def link[I, O, P >: O, Q](first: Stage[I, O], second: Stage[P,Q]) : Stage[I, Q]
}

//a pipeline is composed of stages
trait Stage[I, O] extends (I => O) {
  owner =>

  def ->[Q](next: Stage[O, Q])(implicit link: (Stage[I,O], Stage[O,Q]) => Stage[I, Q]) = link(owner, next)
}


//a basic pipeline just composes functions
trait SynchronousPipelineBuilder extends PipelineBuilder {
  def createStage[I, O](f: I => O) : Stage[I, O] = new Stage[I, O] {
    def apply(input: I) = f(input)
  }

  implicit def link[I, O, P >: O, Q](first: Stage[I, O], second: Stage[P,Q]) : Stage[I, Q] = new Stage[I, Q]{
    def apply(v1: I) = (first andThen second)(v1)
  }
}

object Main extends SynchronousPipelineBuilder {
  val logger = LoggerFactory.getLogger("test")

  def main(args: Array[String]) {
    val pipe =
      createStage{input: Int => input.toFloat} -> createStage(_.toDouble) -> createStage(_.toString)

    logger.info(pipe(42))
  }