ChannelComposition.scala

object Procs {

  implicit class RichProcess[F[_], A](p: Process[F, A]) {
    def through2[B,C](c1: Channel[F, A, B],
                      c2: Channel[F, B, C]): Process[F, C] =
      p.flatMap(composeChannels(c1, c2))
  }

  implicit class RichTaskProcess[A](p: Process[Task, A]) {
    def observeThrough2[B,C](c1: Channel[Task, A, B],
                             c2: Channel[Task, B, C]): Process[Task, (A, B, C)] =
      p.flatMap(composeThroughChannels(c1, c2))
  }

  def composeChannels[F[_], A, B, C](c1: Channel[F, A, B],
                                     c2: Channel[F, B, C]) = {
    i: A => Process.emit(i).through(c1).through(c2)
  }

  def composeThroughChannels[A, B, C](c1: Channel[Task, A, B],
                                     c2: Channel[Task, B, C]):
    A => Process[Task, (A,B,C)] = { i: A =>
    Process.emit(i)
      .observeThrough(c1)
      .observeThrough(c2.contramap(_._2))
      .map{ case ((a,b), c) => (a,b,c) }
  }
}