How to implement Lazy Chain Pattern in Scala

Question

I am trying to define a set of "LazyChains" that will be used to process incoming messages at a future time. I want the API of the LazyChains to be indistinguishable from the Scala collections APIs (ie: Seq, Stream, etc). This would allow me to declaratively define filtering/transformations/actions ahead of time before the messages arrive. It's possible that this is a well known pattern that I don't know the name of, so that is making it hard for me to find any results on this.

This is an example of what I'm trying to accomplish:

val chainA = LazyChain()
 .filter(_ > 1)
 .map(x => x * 2)
 .foreach(println _)

val chainB = LazyChain()
 .filter(_ > 5)
 .flatMap(x => Seq(x, x))
 .foreach(println _)

 chainA.apply(2)  // should print "4"
 chainA.apply(1)  // should print nothing
 chainB.apply(10) // should print "10" - twice

Does this pattern already exist in the Scala collections API? If not, how can I implement this class LazyChain?

This is my current attempt at this. I can't seem to work out how to get the types to work:

case class LazyChain[I, O](val filter : Option[I => Boolean],
                      val transform : I => O,
                      val action : Option[O => Unit]) {

  def filter(otherFilter : I => Boolean): LazyChain[I, O]  = {
      val newFilter = Some({ x : I => {
        filter.map(_.apply(x)).getOrElse(true) && otherFilter.apply(x)
      }})
      copy(filter = newFilter)
  }

  def map[X](otherTransform : O => X) : LazyChain[I, X] = {
    new LazyChain[I, X](
      filter = filter,
      transform = (x: I) => {
        otherTransform.apply(transform.apply(x))
      },
      /*
        type mismatch;
        [error]  found   : Option[O => Unit]
        [error]  required: Option[X => Unit]
      */
      action = action 
    )
  }

  def flatMap[X](otherTransform : O => Seq[X]) : LazyChain[I, X] = {
    new LazyChain[I, X](
      filter = filter,
      transform = (x: I) => {
        /**
           type mismatch;
             [error]  found   : Seq[X]
             [error]  required: X
        */
        otherTransform.apply(transform.apply(x))
      }
    )
  }

  def foreach(newAction : O => Unit) = {
    copy(action = Some(newAction))
  }

  def apply(element : I) = {
    if (filter.map(_.apply(element)).getOrElse(true)) {
      val postTransform = transform.apply(element)
      action.foreach(_.apply(postTransform))
    }
  }
}
object LazyChain {
  def apply[X]() : LazyChain[X, X] = {
    new LazyChain(filter = None, transform = x => x, action = None)
  }
}

Answer 1

All you want is to wrap a function I => List[O] with some fancy methods. You could write your implicit class to add these methods to this type, but Kleisli does most of this for free, via various cats type classes, mainly FilterFunctor.

  import cats.implicits._
  import cats.data.Kleisli

  type LazyChain[I, O] = Kleisli[List, I, O]
  def lazyChain[A]: LazyChain[A, A] = Kleisli[List, A, A](a => List(a))

  val chainA = lazyChain[Int]
    .filter(_ > 1)
    .map(x => x * 2)
    .map(println)

  val chainB = lazyChain[Int]
    .filter(_ > 5)
    .flatMapF(x => List(x, x))
    .map(println)

  chainA(2)  // should print "4"
  chainA(1)  // should print nothing
  chainB(10) // should print "10" - twice

It might look a bit too magical, so here is a handmade version:

case class LazyChain[A, B](run: A => List[B]) {

  def filter(f: B => Boolean): LazyChain[A, B] = chain(_.filter(f))

  def map[C](f: B => C): LazyChain[A, C] = chain(_.map(f))

  def flatMap[C](f: B => List[C]): LazyChain[A, C] = chain(_.flatMap(f))

  def chain[C](f: List[B] => List[C]): LazyChain[A, C] = LazyChain(run andThen f)
}
object LazyChain {
  def apply[I]: LazyChain[I, I] = new LazyChain(a => List(a))
}

Chaining transformations is a common concern and, as the comments say, using something like monix.Observable, iteratees, etc is the proper way to approach this problem (rather than a plain List and streams are naturally lazy.