StreamSupport.scala

import scalaz.stream.{Process, Tee, tee}
import Process.{awaitL, awaitR, emit}
import tee.{feed1R, feed1L}

object StreamSupport {

  /* A Tee that drop from left while the predicate `p` is true for the
   * values, then continue with the Tee `rest` */
  def dropWhileL[L,R,O](p: L => Boolean)(rest: Tee[L,R,O]): Tee[L,R,O] =
    awaitL[L].flatMap { v =>
      if (p(v)) dropWhileL(p)(rest)
      else feed1L(v)(rest)
    }

  /* A Tee that drop from right while the predicate `p` is true for the
   * values, then continue with the Tee `rest` */
  def dropWhileR[L,R,O](p: R => Boolean)(rest: Tee[L,R,O]): Tee[L,R,O] =
    awaitR[R].flatMap { v =>
      if (p(v)) dropWhileR(p)(rest)
      else feed1R(v)(rest)
    }

  /* A Tee that emits the equivalent values of left and right. Both
   * sides are expected to be in ascending order.
   *
   * The Tee works recursively by first reading a value from both
   * sides and applying the keyL and keyR functions to the values
   * respectively. If the results are equivalent, it emits the tuple.
   *
   * If the values are not equivalent, it skips from the side with the
   * smaller value until an equal or greater value is found before
   * comparing the values again.
   *
   * For example:
   * {{{
   * // result is List((2 -> "2"), (5 -> "5"), (8 ->"8"))
   * Process(1,2,5,6,8).tee(Process("2","5","7","8"))(join(identity, _.toInt)).
   *   toStream.toList
   * }}}
   */
   def join[L,R,T](keyL: L => T, keyR: R => T)(implicit o: Ordering[T]) = {
     def impl: Tee[L,R,(L,R)] = awaitL[L].flatMap { l =>
       awaitR[R].flatMap { r =>
         val lk = keyL(l)
         val rk = keyR(r)

         o.compare(lk, rk) match {
           case 0 => emit((l,r)) ++ impl
           case -1 => dropWhileL((o.lt(_:T, rk)).compose(keyL))(feed1R(r)(impl))
           case 1 => dropWhileR((o.lt(_:T, lk)).compose(keyR))(feed1L(l)(impl))
         }
       }
     }

     impl
   }
}