A dirty example of monadic value processing pipeline

Pipeline.scala

import cats.Foldable
import shapeless.ops.product._
import shapeless.ops.tuple._
import cats._
import cats.implicits._

trait Pipeline[+A] {
  @inline def map[B](f: A => B): Pipeline[B]

  @inline def flatMap[B](f: A => Pipeline[B]): Pipeline[B]

  @inline def eff(f: A => Unit): Pipeline[A]

  @inline def end(f: A => Unit): Unit

  @inline def end(): Unit

  @inline def value: A
}


case class PipelineImpl[+A](value: A) extends Pipeline[A] {
  @inline final def map[B](f: A => B): Pipeline[B] = PipelineImpl(f(this.value))

  @inline final def flatMap[B](f: A => Pipeline[B]): Pipeline[B] = f(this.value)

  @inline def eff(f: A => Unit): this.type = {
    f(value)
    this
  }

  @inline def end(f: A => Unit): Unit = f(value)

  @inline def end(): Unit = {}

  @inline def get: A = value
}


object Pipeline {
  def apply[A](value: A): Pipeline[A] = new PipelineImpl[A](value)

  sealed trait TPipeline[+A] {
    this: Pipeline[A] =>
  }

  type TupledPipeline[+A] = Pipeline[A] with TPipeline[A]

  implicit class ValueOps[V](v: Pipeline[V]) {
    @inline final def pull[B](f: => B): Pipeline[(V, B)] = {
      PipelineImpl(Tuple2(v.value, f))
    }

    @inline final def pull[R[_] : Foldable, K[_] <: Traversable[_], E]
    (f: => R[K[E]])
    (implicit ev: scala.collection.generic.CanBuildFrom[Traversable[E], E, K[E]]
      , ev1: scala.collection.generic.CanBuildFrom[K[E], E, Traversable[E]]
    )
    : Pipeline[K[E]] = {
      val folded = f.foldLeft(Traversable.empty[E]) {
        case (acc, r) =>
          acc ++ ev1(r).result()
      }
      PipelineImpl(ev(folded).result())
    }

    @inline final def fork[B](f: V => B): Pipeline[(V, B)] = {
      PipelineImpl(Tuple2(v.value, f(v.value)))
    }

    @inline final def tupled: TupledPipeline[Tuple1[V]] = {
      new PipelineImpl(Tuple1(v.value)) with TPipeline[Tuple1[V]]
    }

    @inline def flatten[K](implicit ev: V <:< Pipeline[K]): Pipeline[K] = {
      ev(v.value)
    }
  }

  implicit class ValueProductOps[V <: Product](v: Pipeline[V]) {
    @inline final def tupled[T](implicit toTuple: ToTuple.Aux[V, T]): TupledPipeline[T] = {
      new PipelineImpl(toTuple(v.value)) with TPipeline[T]
    }
  }

  implicit class ValueTupledOps[V <: Product](v: Pipeline[V] with TPipeline[V]) {
    @inline final def pull[B](f: => B)(implicit prepend: Prepend[V, Tuple1[B]]): TupledPipeline[prepend.Out] = {
      new PipelineImpl(prepend(v.value, Tuple1(f))) with TPipeline[prepend.Out]
    }

    @inline final def pullt[B <: Product](f: => B)(implicit prepend: Prepend[V, B]): TupledPipeline[prepend.Out] = {
      new PipelineImpl(prepend(v.value, f)) with TPipeline[prepend.Out]
    }

    @inline final def fork[B](f: V => B)(implicit prepend: Prepend[V, Tuple1[B]]): TupledPipeline[prepend.Out] = {
      new PipelineImpl(prepend(v.value, Tuple1(f(v.value)))) with TPipeline[prepend.Out]
    }

    @inline final def forkt[B](f: V => B)(implicit prepend: Prepend[V, B]): TupledPipeline[prepend.Out] = {
      new PipelineImpl(prepend(v.value, f(v.value))) with TPipeline[prepend.Out]
    }

    @inline def flatten[K](implicit ev: V <:< TupledPipeline[K]): TupledPipeline[K] = {
      ev(v.value)
    }
  }

}

Pipeline(1)
  .tupled
  .pull(2)
  .pull(3)
  .pullt((3, 4))

Pipeline((0, 1))
  .tupled

Pipeline((0, 1))
  .tupled
  .pull(2)
  .pull(3)
  .pullt((3, 4))

Pipeline(1).fork(_ * 2)


Pipeline(Pipeline(1)).flatten

Pipeline((0, 1))
  .tupled
  .pull(2)
  .pull(3)
  .pullt((3, 4))
  .fork(_._1 * 2)
  .forkt(v => v)
  .pullt((3, 4))
  .map(v => v)
  .tupled
  .pull(5)