<commit_before>package scalaz.stream

import scala.collection.immutable.{IndexedSeq,SortedMap,Queue,Vector}

import scalaz.{Catchable,Monad,MonadPlus,Monoid,Nondeterminism,Semigroup}
import scalaz.concurrent.Task
import scalaz.Leibniz.===
import scalaz.{\/,-\/,\/-,~>,Leibniz}
import \/._

/** 
 * A `Process[F,O]` represents a stream of `O` values which can interleave 
 * external requests to evaluate expressions of the form `F[A]`. It takes
 * the form of a state machine with three possible states: `Emit`, which 
 * indicates that `h` should be emitted to the output stream, `Halt`,
 * which indicates that the `Process` is finished making requests and 
 * emitting values to the output stream, and `Await` which asks the driver 
 * to evaluate some `F[A]` and resume processing once the result is available. 
 * See the constructor definitions in the `Process` companion object.
 */
sealed abstract class Process[+F[_],+O] {
  
  import Process._

  /** Transforms the output values of this `Process` using `f`. */
  final def map[O2](f: O => O2): Process[F,O2] = this match {
    case Await(req,recv,fb,c) => 
      Await[F,Any,O2](req, recv andThen (_ map f), fb map f, c map f) 
    case Emit(h, t) => Emit[F,O2](h map f, t map f)
    case Halt => Halt
  }

  /** 
   * Generate a `Process` dynamically for each output of this `Process`, and
   * sequence these processes using `append`. 
   */
  final def flatMap[F2[x]>:F[x], O2](f: O => Process[F2,O2]): Process[F2,O2] = this match {
    case Halt => Halt
    case Emit(o, t) => 
      if (o.isEmpty) t.flatMap(f)
      else f(o.head) ++ emitSeq(o.tail, t).flatMap(f)
    case Await(req,recv,fb,c) => 
      Await(req, recv andThen (_ flatMap f), fb flatMap f, c flatMap f)
  }
  
  /** 
   * Run this `Process`, then, if it halts without an error, run `p2`. 
   * Note that `p2` is appended to the `fallback` argument of any `Await`
   * produced by this `Process`. If this is not desired, use `then`. 
   */
  final def append[F2[x]>:F[x], O2>:O](p2: => Process[F2,O2]): Process[F2,O2] = this match {
    case Halt => p2
    case Emit(h, t) => emitSeq(h, t append p2)
    case Await(req,recv,fb,c) => 
      Await(req, recv andThen (_ append p2), fb append p2, c)
  }

  /** Operator alias for `append`. */
  final def ++[F2[x]>:F[x], O2>:O](p2: => Process[F2,O2]): Process[F2,O2] = 
    this append p2

  /** 
   * Run this `Process`, then, if it self-terminates, run `p2`. 
   * This differs from `append` in that `p2` is not consulted if this
   * `Process` terminates due to the input being exhausted. That is, 
   * we do not modify the `fallback` arguments to any `Await` produced
   * by this `Process`.
   */
  final def then[F2[x]>:F[x],O2>:O](p2: => Process[F2,O2]): Process[F2,O2] = this match {
    case Halt => p2
    case Emit(h, t) => emitSeq(h, t then p2)
    case Await(req,recv,fb,c) => 
      Await(req, recv andThen (_ then p2), fb, c)
  }

  /** 
   * Removes all emitted elements from the front of this `Process`.
   * The second argument returned by this method is guaranteed to be 
   * an `Await` or a `Halt`--if there are multiple `Emit`s at the
   * front of this process, the sequences are concatenated together.
   * 
   * If this `Process` does not begin with an `Emit`, returns the empty
   * sequence along with `this`. 
   */
  final def unemit: (Seq[O], Process[F,O]) = {
    @annotation.tailrec
    def go(acc: Seq[O], cur: Process[F,O]): (Seq[O], Process[F,O]) = 
      cur match {
        case Emit(h, t) => go(acc ++ h, t) 
        case _ => (acc, cur)
      }
    go(Seq(), this)
  }

  private[stream] final def unconsAll: Process[F, (Seq[O], Process[F,O])] = this match {
    case Halt => Halt
    case Emit(h, t) => if (h.isEmpty) t.unconsAll else emit((h,t))
    case Await(req,recv,fb,c) => await(req)(recv andThen (_.unconsAll), fb.unconsAll, c.unconsAll) 
  }
  private[stream] final def uncons: Process[F, (O, Process[F,O])] = 
    unconsAll map { case (h,t) => (h.head, emitAll(h.tail) ++ t) }
    
  /** 
   * Run this process until it halts, then run it again and again, as
   * long as no errors occur. 
   */
  final def repeat[F2[x]>:F[x],O2>:O]: Process[F2,O2] = {
    def go(cur: Process[F,O]): Process[F,O] = cur match {
      case Halt => go(this)
      case Await(req,recv,fb,c) => Await(req, recv andThen go, fb, c)
      case Emit(h, t) => emitSeq(h, go(t))
    }
    go(this)
  }

  /**
   * Halt this process, but give it an opportunity to run any requests it has 
   * in the `cleanup` argument of its next `Await`.
   */
  @annotation.tailrec
  final def kill: Process[F,Nothing] = this match {
    case Await(req,recv,fb,c) => c.drain 
    case Emit(h, t) => t.kill
    case Halt => Halt
  }

  /** 
   * Switch to the `fallback` case of the _next_ `Await` issued by this `Process`.
   */
  final def fallback: Process[F,O] = this match {
    case Await(req,recv,fb,c) => fb 
    case Emit(h, t) => emitSeq(h, t.fallback)
    case Halt => Halt
  }

  /**
   * Replaces the outermost `fallback` parameter of `Await`.
   */
  final def orElse[F2[x]>:F[x],O2>:O](fallback: Process[F2,O2]): Process[F2,O2] = this match {
    case Await(req,recv,fb,c) => Await(req,recv,fallback,c)
    case Emit(h, t) => Emit(h, t.orElse(fallback))
    case Halt => Halt
  }

  /**
   * Replaces the outermost `fallback` parameter of `Await`.
   */
  final def onFallback[F2[x]>:F[x],O2>:O](fallback: Process[F2,O2]): Process[F2,O2] = 
    orElse(fallback)

  /**
   * Replaces the outermost `cleanup` parameter of `Await`.
   */
  final def onError[F2[x]>:F[x],O2>:O](handler: Process[F2,O2]): Process[F2,O2] = this match {
    case Await(req,recv,fb,c) => Await(req,recv,fb,handler)
    case Emit(h, t) => Emit(h, t.onError(handler))
    case Halt => Halt
  }

  /**
   * Replaces the outermost `fallback` and `cleanup` parameters of `Await`.
   */
  final def onFallbackOrError[F2[x]>:F[x],O2>:O](handler: Process[F2,O2]): Process[F2,O2] = this match {
    case Await(req,recv,fb,c) => Await(req,recv,handler,handler)
    case Emit(h, t) => Emit(h, t.onFallbackOrError(handler))
    case Halt => Halt
  }

  /**
   * Switch to the `fallback` case of _all_ subsequent awaits.
   */
  final def disconnect: Process[Nothing,O] = this match {
    case Await(req,recv,fb,c) => fb.disconnect 
    case Emit(h, t) => emitSeq(h, t.disconnect) 
    case Halt => Halt
  }

  /** 
   * Switch to the `cleanup` case of the next `Await` issued by this `Process`.
   */
  final def cleanup: Process[F,O] = this match {
    case Await(req,recv,fb,c) => fb 
    case Halt => Halt
    case Emit(h, t) => emitSeq(h, t.cleanup)
  }

  /**
   * Switch to the `cleanup` case of _all_ subsequent awaits.
   */
  final def hardDisconnect: Process[Nothing,O] = this match {
    case Await(req,recv,fb,c) => c.hardDisconnect 
    case Halt => Halt
    case Emit(h, t) => emitSeq(h, t.hardDisconnect) 
  }

  /** 
   * Remove any leading emitted values from this `Process`. 
   */
  @annotation.tailrec
  final def trim: Process[F,O] = this match {
    case Emit(h, t) => t.trim
    case _ => this
  }

  /** Correctly typed deconstructor for `Await`. */
  def asAwait: Option[(F[AwaitF.Req], AwaitF.Req => Process[F,O], Process[F,O], Process[F,O])] = 
    this match {
      case Await(req,recv,fb,c) => Some((req.asInstanceOf[F[AwaitF.Req]],recv,fb,c))
      case _ => None
    }

  /**
   * Ignores output of this `Process`. A drained `Process` will never `Emit`.  
   */
  def drain: Process[F,Nothing] = this match {
    case Halt => Halt
    case Emit(h, t) => t.drain
    case Await(req,recv,fb,c) => Await(
      req, recv andThen (_ drain), 
      fb.drain, c.drain)
  }

  /** 
   * Feed the output of this `Process` as input of `p2`. The implementation  
   * will fuse the two processes, so this process will only generate
   * values as they are demanded by `p2`. If `p2` signals termination, `this`
   * is killed using `kill`, giving it the opportunity to clean up. 
   */
  final def pipe[O2](p2: Process1[O,O2]): Process[F,O2] =
    // Since `Process1[O,O2] <: Tee[O,Any,O2]`, but it is a `Tee` that
    // never reads from its right input, we can define this in terms of `tee`!
    (this tee Halt)(p2)

  /** Operator alias for `pipe`. */
  final def |>[O2](p2: Process1[O,O2]): Process[F,O2] = 
    this pipe p2

  /** 
   * Use a `Tee` to interleave or combine the outputs of `this` and
   * `p2`. This can be used for zipping, interleaving, and so forth.
   * Nothing requires that the `Tee` read elements from each 
   * `Process` in lockstep. It could read fifty elements from one 
   * side, then two elements from the other, then combine or
   * interleave these values in some way, etc.
   * 
   * If at any point the `Tee` awaits on a side that has halted, 
   * we gracefully kill off the other side, then halt.
   */ 
  final def tee[F2[x]>:F[x],O2,O3](p2: Process[F2,O2])(t: Tee[O,O2,O3]): Process[F2,O3] = 
    // Somewhat evil: we are passing `null` for the `Nondeterminism` and `Catchable` here, 
    // safe because the types guarantee that a `Tee` cannot issue a `Both` 
    // request that would result in the `Nondeterminism` instance being used.  
    // This lets us reuse a single function, `wye`, as the implementation for
    // both `tee` and `pipe`!
    (this wye p2)(t)(null,null)

  /** 
   * Like `tee`, but we allow the `Wye` to read nondeterministically 
   * from both sides at once, using the supplied `Nondeterminism` 
   * instance.
   * 
   * If `y` is in the state of awaiting `Both`, this implementation 
   * will continue feeding `y` until either it halts or _both_ sides
   * halt. 
   * 
   * If `y` is in the state of awaiting `L`, and the left 
   * input has halted, we halt. Likewise for the right side. 
   * 
   * For as long as `y` permits it, this implementation will _always_ 
   * feed it any leading `Emit` elements from either side before issuing
   * new `F` requests. More sophisticated chunking and fairness 
   * policies do not belong here, but should be built into the `Wye`
   * and/or its inputs. 
   */ 
  final def wye[F2[x]>:F[x],O2,O3](p2: Process[F2,O2])(y: Wye[O,O2,O3])(implicit F2: Nondeterminism[F2], E: Catchable[F2]): Process[F2,O3] = {
    import F2.monadSyntax._
    try y match {
      case Halt => this.kill ++ p2.kill
      case Emit(h,y2) => 
        Emit(h, this.wye(p2)(y2))
      case Await(_,_,_,_) => 
        val u1 = this.unemit; val h1 = u1._1; val t1 = u1._2
        val u2 = p2.unemit; val h2 = u2._1; val t2 = u2._2
        val ready = These.align(h1, h2)
        val (y2, ready2) = if (ready.isEmpty) (y, ready) else { // we have some values queued up, try feeding them to the Wye
          y.feed(ready) { // .feed is a tail recursive function
            case Await(req, recv, fb, c) => (req.tag: @annotation.switch) match {
              case 0 => // Left 
                val recv_ = recv.asInstanceOf[O => Wye[O,O2,O3]]
                (e: These[O,O2]) => e match { 
                  case These.This(o) => (None, Some(recv_(o)))
                  case These.That(_) => (None, None)
                  case These.Both(o,o2) => (Some(These.That(o2)), Some(recv_(o)))
                }
              case 1 => // Right 
                val recv_ = recv.asInstanceOf[O2 => Wye[O,O2,O3]]
                (e: These[O,O2]) => e match { 
                  case These.This(_) => (None, None)
                  case These.That(o2) => (None, Some(recv_(o2)))
                  case These.Both(o,o2) => (Some(These.This(o)), Some(recv_(o2)))
                }
              case 2 => // Both 
                val recv_ = recv.asInstanceOf[These[O,O2] => Wye[O,O2,O3]]
                (e: These[O,O2]) => (None, Some(recv_(e)))
            }
            case _ => _ => (None, None)
          }
        }
        val (h1Next, h2Next) = These.unalign(ready2)
        val (thisNext_, p2Next_) = (emitSeq(h1Next, t1), emitSeq(h2Next, t2))
        val thisNext = thisNext_.asInstanceOf[Process[F2,O]] 
        val p2Next = p2Next_.asInstanceOf[Process[F2,O2]]
        y2 match {
          case Await(req,_,_,_) => (req.tag: @annotation.switch) match {
            case 0 => // Left
              thisNext match {
                case AwaitF(reqL,recvL,fbL,cL) => // unfortunately, casts required here
                  await(reqL)(recvL andThen (_.wye(p2Next)(y2)), fbL.wye(p2Next)(y2), cL.wye(p2Next)(y2))
                case Halt => p2Next.kill ++ y2.disconnect
                case e@Emit(_,_) => thisNext.wye(p2Next)(y2) 
              }
            case 1 => // Right 
              p2Next match {
                case AwaitF(reqR,recvR,fbR,cR) => // unfortunately, casts required here
                  // in the event of a fallback or error, `y` will end up running the right's fallback/cleanup
                  // actions on the next cycle - it still needs that value on the right and will run any awaits
                  // to try to obtain that value!
                  await(reqR)(recvR andThen (p2 => thisNext.wye[F2,O2,O3](p2)(y2)), 
                               thisNext.wye(fbR)(y2), 
                               thisNext.wye(cR)(y2))
                case Halt => thisNext.kill ++ y2.disconnect
                case e@Emit(_,_) => thisNext.wye(p2Next)(y2) 
              }
            case 2 => thisNext match { // Both
              case Halt => p2Next |> y2.detachL
              case AwaitF(reqL, recvL, fbL, cL) => p2Next match {
                case Halt => 
                  Await(reqL, recvL andThen (_.wye(p2Next)(y2)), fbL.wye(p2Next)(y2), cL.wye(p2Next)(y2))
                // If both sides are in the Await state, we use the
                // Nondeterminism instance to request both sides
                // concurrently.
                case AwaitF(reqR, recvR, fbR, cR) =>
                  wrap(
                    F2.choose(
                      E.attempt(reqL.map(recvL)).map {
                        _.fold({ case End => fbL 
                                 case t: Throwable => cL ++ (throw t) 
                               }, 
                               e => e)
                      },
                      E.attempt(reqR.map(recvR)).map {
                        _.fold({ case End => fbR 
                                 case t: Throwable => cR ++ (throw t) 
                               }, 
                               e => e)
                      }
                    )
                  ).flatMap { 
                    _.fold(l => wrap(l._2).flatMap(p2 => l._1.wye(p2)(y2)),
                           r => wrap(r._1).flatMap(t => t.wye(r._2)(y2)))
                  }
              }
            }
          }
          case _ => thisNext.wye(p2Next)(y2)
        }
    }
    catch { case e: Throwable => this.kill ++ p2.kill ++ (throw e) }
  }

  /** Translate the request type from `F` to `G`, using the given polymorphic function. */
  def translate[G[_]](f: F ~> G): Process[G,O] = this match {
    case Emit(h, t) => Emit(h, t.translate(f))
    case Halt => Halt
    case Await(req, recv, fb, c) => 
      Await(f(req), recv andThen (_ translate f), fb translate f, c translate f)
  }

  /** 
   * Catch exceptions produced by this `Process`. The resulting `Process` halts after the
   * the first emitted failure, though this can handled and resumed by a subsequent `Process`. 
   */ 
  def attempt[F2[x]>:F[x]](implicit F: Catchable[F2]): Process[F2, Throwable \/ O] = 
  this match {
    case Emit(h, t) => Emit(h map (\/-(_)), t.attempt[F2])
    case Halt => Halt
    case Await(req, recv, fb, c) => 
      await(F.attempt(req))(
        _.fold(err => emit(-\/(err)), recv andThen (_.attempt[F2])), 
        fb.attempt[F2], c.attempt[F2])
  }

  /** 
   * Catch some of the exceptions generated by this `Process`, rethrowing any
   * not handled by the given `PartialFunction`. 
   */
  def handle[F2[x]>:F[x],O2>:O](f: PartialFunction[Throwable,O2])(implicit F: Catchable[F2]) = 
    attempt(F) |> (process1.lift { 
      case \/-(o) => o
      case -\/(err) => f.lift(err).getOrElse(throw err) })

  /** 
   * Map over this `Process` to produce a stream of `F`-actions, 
   * then evaluate these actions. 
   */
  def evalMap[F2[x]>:F[x],O2](f: O => F2[O2]): Process[F2,O2] = 
    map(f).eval

  /** 
   * Map over this `Process` to produce a stream of `F`-actions, 
   * then evaluate these actions, returning results in whatever
   * order they come back. `bufSize` controls the maximum number
   * of evaluations that can be queued at any one time. 
   */
  def gatherMap[F2[x]>:F[x],O2](bufSize: Int)(f: O => F2[O2])(
                                implicit F: Nondeterminism[F2]): Process[F2,O2] = 
    map(f).gather(bufSize)
 
  /** 
   * Collect the outputs of this `Process[F,O]`, given a `Monad[F]` in
   * which we can catch exceptions. This function is not tail recursive and
   * relies on the `Monad[F]` to ensure stack safety. 
   */
  final def collect[F2[x]>:F[x], O2>:O](implicit F: Monad[F2], C: Catchable[F2]): F2[IndexedSeq[O2]] = {
    def go(cur: Process[F2,O2], acc: collection.mutable.ListBuffer[O2]): F2[IndexedSeq[O2]] =
      cur match {
        case Emit(h,t) => 
          acc ++= h.asInstanceOf[Seq[O2]]
          go(t.asInstanceOf[Process[F2,O2]], acc)
        case Halt => F.point(acc.toIndexedSeq)
        case Await(req,recv,fb,c) => 
           F.bind (C.attempt(req.asInstanceOf[F2[AnyRef]])) {
             _.fold(
               { case End => go(fb.asInstanceOf[Process[F2,O2]], acc)
                 case err => c match {
                   case Halt => C.fail(err)
                   case _ => go(c.asInstanceOf[Process[F2,O2]] ++ wrap(C.fail(err)), new collection.mutable.ListBuffer()) 
                 }
               }, o => go(recv.asInstanceOf[AnyRef => Process[F2,O2]](o), acc))
           }
      }
    go(this, new collection.mutable.ListBuffer[O2]())
  }

  /** Run this `Process` solely for its final emitted value, if one exists. */
  final def runLast[F2[x]>:F[x], O2>:O](implicit F: Monad[F2], C: Catchable[F2]): F2[Option[O2]] = 
    F.map(this.last.collect[F2,O2])(_.lastOption)

  /** Run this `Process` solely for its final emitted value, if one exists, using `o2` otherwise. */
  final def runLastOr[F2[x]>:F[x], O2>:O](o2: => O2)(implicit F: Monad[F2], C: Catchable[F2]): F2[O2] = 
    F.map(this.last.collect[F2,O2])(_.lastOption.getOrElse(o2))

  /** Run this `Process`, purely for its effects. */
  final def run[F2[x]>:F[x]](implicit F: Monad[F2], C: Catchable[F2]): F2[Unit] = 
    F.void(drain.collect(F, C))

  /** Alias for `this |> process1.buffer(n)`. */
  def buffer(n: Int): Process[F,O] = 
    this |> process1.buffer(n)

  /** Alias for `this |> process1.bufferBy(f)`. */
  def bufferBy(f: O => Boolean): Process[F,O] = 
    this |> process1.bufferBy(f)

  /** Alias for `this |> process1.bufferBy(f)`. */
  def bufferAll: Process[F,O] = 
    this |> process1.bufferAll

  /** Ignores the first `n` elements output from this `Process`. */
  def drop(n: Int): Process[F,O] = 
    this |> processes.drop[O](n)

  /** Ignores elements from the output of this `Process` until `f` tests false. */
  def dropWhile(f: O => Boolean): Process[F,O] = 
    this |> processes.dropWhile(f)

  /** Skips any output elements not matching the predicate. */
  def filter(f: O => Boolean): Process[F,O] = 
    this |> processes.filter(f)

  /** Connect this `Process` to `process1.fold(b)(f)`. */
  def fold[B](b: B)(f: (B,O) => B): Process[F,B] = 
    this |> process1.fold(b)(f)
  <commit_msg>Syntax on Process for chunking and buffering<commit_after>package scalaz.stream

import scala.collection.immutable.{IndexedSeq,SortedMap,Queue,Vector}

import scalaz.{Catchable,Monad,MonadPlus,Monoid,Nondeterminism,Semigroup}
import scalaz.concurrent.Task
import scalaz.Leibniz.===
import scalaz.{\/,-\/,\/-,~>,Leibniz}
import \/._

/** 
 * A `Process[F,O]` represents a stream of `O` values which can interleave 
 * external requests to evaluate expressions of the form `F[A]`. It takes
 * the form of a state machine with three possible states: `Emit`, which 
 * indicates that `h` should be emitted to the output stream, `Halt`,
 * which indicates that the `Process` is finished making requests and 
 * emitting values to the output stream, and `Await` which asks the driver 
 * to evaluate some `F[A]` and resume processing once the result is available. 
 * See the constructor definitions in the `Process` companion object.
 */
sealed abstract class Process[+F[_],+O] {
  
  import Process._

  /** Transforms the output values of this `Process` using `f`. */
  final def map[O2](f: O => O2): Process[F,O2] = this match {
    case Await(req,recv,fb,c) => 
      Await[F,Any,O2](req, recv andThen (_ map f), fb map f, c map f) 
    case Emit(h, t) => Emit[F,O2](h map f, t map f)
    case Halt => Halt
  }

  /** 
   * Generate a `Process` dynamically for each output of this `Process`, and
   * sequence these processes using `append`. 
   */
  final def flatMap[F2[x]>:F[x], O2](f: O => Process[F2,O2]): Process[F2,O2] = this match {
    case Halt => Halt
    case Emit(o, t) => 
      if (o.isEmpty) t.flatMap(f)
      else f(o.head) ++ emitSeq(o.tail, t).flatMap(f)
    case Await(req,recv,fb,c) => 
      Await(req, recv andThen (_ flatMap f), fb flatMap f, c flatMap f)
  }
  
  /** 
   * Run this `Process`, then, if it halts without an error, run `p2`. 
   * Note that `p2` is appended to the `fallback` argument of any `Await`
   * produced by this `Process`. If this is not desired, use `then`. 
   */
  final def append[F2[x]>:F[x], O2>:O](p2: => Process[F2,O2]): Process[F2,O2] = this match {
    case Halt => p2
    case Emit(h, t) => emitSeq(h, t append p2)
    case Await(req,recv,fb,c) => 
      Await(req, recv andThen (_ append p2), fb append p2, c)
  }

  /** Operator alias for `append`. */
  final def ++[F2[x]>:F[x], O2>:O](p2: => Process[F2,O2]): Process[F2,O2] = 
    this append p2

  /** 
   * Run this `Process`, then, if it self-terminates, run `p2`. 
   * This differs from `append` in that `p2` is not consulted if this
   * `Process` terminates due to the input being exhausted. That is, 
   * we do not modify the `fallback` arguments to any `Await` produced
   * by this `Process`.
   */
  final def then[F2[x]>:F[x],O2>:O](p2: => Process[F2,O2]): Process[F2,O2] = this match {
    case Halt => p2
    case Emit(h, t) => emitSeq(h, t then p2)
    case Await(req,recv,fb,c) => 
      Await(req, recv andThen (_ then p2), fb, c)
  }

  /** 
   * Removes all emitted elements from the front of this `Process`.
   * The second argument returned by this method is guaranteed to be 
   * an `Await` or a `Halt`--if there are multiple `Emit`s at the
   * front of this process, the sequences are concatenated together.
   * 
   * If this `Process` does not begin with an `Emit`, returns the empty
   * sequence along with `this`. 
   */
  final def unemit: (Seq[O], Process[F,O]) = {
    @annotation.tailrec
    def go(acc: Seq[O], cur: Process[F,O]): (Seq[O], Process[F,O]) = 
      cur match {
        case Emit(h, t) => go(acc ++ h, t) 
        case _ => (acc, cur)
      }
    go(Seq(), this)
  }

  private[stream] final def unconsAll: Process[F, (Seq[O], Process[F,O])] = this match {
    case Halt => Halt
    case Emit(h, t) => if (h.isEmpty) t.unconsAll else emit((h,t))
    case Await(req,recv,fb,c) => await(req)(recv andThen (_.unconsAll), fb.unconsAll, c.unconsAll) 
  }
  private[stream] final def uncons: Process[F, (O, Process[F,O])] = 
    unconsAll map { case (h,t) => (h.head, emitAll(h.tail) ++ t) }
    
  /** 
   * Run this process until it halts, then run it again and again, as
   * long as no errors occur. 
   */
  final def repeat[F2[x]>:F[x],O2>:O]: Process[F2,O2] = {
    def go(cur: Process[F,O]): Process[F,O] = cur match {
      case Halt => go(this)
      case Await(req,recv,fb,c) => Await(req, recv andThen go, fb, c)
      case Emit(h, t) => emitSeq(h, go(t))
    }
    go(this)
  }

  /**
   * Halt this process, but give it an opportunity to run any requests it has 
   * in the `cleanup` argument of its next `Await`.
   */
  @annotation.tailrec
  final def kill: Process[F,Nothing] = this match {
    case Await(req,recv,fb,c) => c.drain 
    case Emit(h, t) => t.kill
    case Halt => Halt
  }

  /** 
   * Switch to the `fallback` case of the _next_ `Await` issued by this `Process`.
   */
  final def fallback: Process[F,O] = this match {
    case Await(req,recv,fb,c) => fb 
    case Emit(h, t) => emitSeq(h, t.fallback)
    case Halt => Halt
  }

  /**
   * Replaces the outermost `fallback` parameter of `Await`.
   */
  final def orElse[F2[x]>:F[x],O2>:O](fallback: Process[F2,O2]): Process[F2,O2] = this match {
    case Await(req,recv,fb,c) => Await(req,recv,fallback,c)
    case Emit(h, t) => Emit(h, t.orElse(fallback))
    case Halt => Halt
  }

  /**
   * Replaces the outermost `fallback` parameter of `Await`.
   */
  final def onFallback[F2[x]>:F[x],O2>:O](fallback: Process[F2,O2]): Process[F2,O2] = 
    orElse(fallback)

  /**
   * Replaces the outermost `cleanup` parameter of `Await`.
   */
  final def onError[F2[x]>:F[x],O2>:O](handler: Process[F2,O2]): Process[F2,O2] = this match {
    case Await(req,recv,fb,c) => Await(req,recv,fb,handler)
    case Emit(h, t) => Emit(h, t.onError(handler))
    case Halt => Halt
  }

  /**
   * Replaces the outermost `fallback` and `cleanup` parameters of `Await`.
   */
  final def onFallbackOrError[F2[x]>:F[x],O2>:O](handler: Process[F2,O2]): Process[F2,O2] = this match {
    case Await(req,recv,fb,c) => Await(req,recv,handler,handler)
    case Emit(h, t) => Emit(h, t.onFallbackOrError(handler))
    case Halt => Halt
  }

  /**
   * Switch to the `fallback` case of _all_ subsequent awaits.
   */
  final def disconnect: Process[Nothing,O] = this match {
    case Await(req,recv,fb,c) => fb.disconnect 
    case Emit(h, t) => emitSeq(h, t.disconnect) 
    case Halt => Halt
  }

  /** 
   * Switch to the `cleanup` case of the next `Await` issued by this `Process`.
   */
  final def cleanup: Process[F,O] = this match {
    case Await(req,recv,fb,c) => fb 
    case Halt => Halt
    case Emit(h, t) => emitSeq(h, t.cleanup)
  }

  /**
   * Switch to the `cleanup` case of _all_ subsequent awaits.
   */
  final def hardDisconnect: Process[Nothing,O] = this match {
    case Await(req,recv,fb,c) => c.hardDisconnect 
    case Halt => Halt
    case Emit(h, t) => emitSeq(h, t.hardDisconnect) 
  }

  /** 
   * Remove any leading emitted values from this `Process`. 
   */
  @annotation.tailrec
  final def trim: Process[F,O] = this match {
    case Emit(h, t) => t.trim
    case _ => this
  }

  /** Correctly typed deconstructor for `Await`. */
  def asAwait: Option[(F[AwaitF.Req], AwaitF.Req => Process[F,O], Process[F,O], Process[F,O])] = 
    this match {
      case Await(req,recv,fb,c) => Some((req.asInstanceOf[F[AwaitF.Req]],recv,fb,c))
      case _ => None
    }

  /**
   * Ignores output of this `Process`. A drained `Process` will never `Emit`.  
   */
  def drain: Process[F,Nothing] = this match {
    case Halt => Halt
    case Emit(h, t) => t.drain
    case Await(req,recv,fb,c) => Await(
      req, recv andThen (_ drain), 
      fb.drain, c.drain)
  }

  /** 
   * Feed the output of this `Process` as input of `p2`. The implementation  
   * will fuse the two processes, so this process will only generate
   * values as they are demanded by `p2`. If `p2` signals termination, `this`
   * is killed using `kill`, giving it the opportunity to clean up. 
   */
  final def pipe[O2](p2: Process1[O,O2]): Process[F,O2] =
    // Since `Process1[O,O2] <: Tee[O,Any,O2]`, but it is a `Tee` that
    // never reads from its right input, we can define this in terms of `tee`!
    (this tee Halt)(p2)

  /** Operator alias for `pipe`. */
  final def |>[O2](p2: Process1[O,O2]): Process[F,O2] = 
    this pipe p2

  /** 
   * Use a `Tee` to interleave or combine the outputs of `this` and
   * `p2`. This can be used for zipping, interleaving, and so forth.
   * Nothing requires that the `Tee` read elements from each 
   * `Process` in lockstep. It could read fifty elements from one 
   * side, then two elements from the other, then combine or
   * interleave these values in some way, etc.
   * 
   * If at any point the `Tee` awaits on a side that has halted, 
   * we gracefully kill off the other side, then halt.
   */ 
  final def tee[F2[x]>:F[x],O2,O3](p2: Process[F2,O2])(t: Tee[O,O2,O3]): Process[F2,O3] = 
    // Somewhat evil: we are passing `null` for the `Nondeterminism` and `Catchable` here, 
    // safe because the types guarantee that a `Tee` cannot issue a `Both` 
    // request that would result in the `Nondeterminism` instance being used.  
    // This lets us reuse a single function, `wye`, as the implementation for
    // both `tee` and `pipe`!
    (this wye p2)(t)(null,null)

  /** 
   * Like `tee`, but we allow the `Wye` to read nondeterministically 
   * from both sides at once, using the supplied `Nondeterminism` 
   * instance.
   * 
   * If `y` is in the state of awaiting `Both`, this implementation 
   * will continue feeding `y` until either it halts or _both_ sides
   * halt. 
   * 
   * If `y` is in the state of awaiting `L`, and the left 
   * input has halted, we halt. Likewise for the right side. 
   * 
   * For as long as `y` permits it, this implementation will _always_ 
   * feed it any leading `Emit` elements from either side before issuing
   * new `F` requests. More sophisticated chunking and fairness 
   * policies do not belong here, but should be built into the `Wye`
   * and/or its inputs. 
   */ 
  final def wye[F2[x]>:F[x],O2,O3](p2: Process[F2,O2])(y: Wye[O,O2,O3])(implicit F2: Nondeterminism[F2], E: Catchable[F2]): Process[F2,O3] = {
    import F2.monadSyntax._
    try y match {
      case Halt => this.kill ++ p2.kill
      case Emit(h,y2) => 
        Emit(h, this.wye(p2)(y2))
      case Await(_,_,_,_) => 
        val u1 = this.unemit; val h1 = u1._1; val t1 = u1._2
        val u2 = p2.unemit; val h2 = u2._1; val t2 = u2._2
        val ready = These.align(h1, h2)
        val (y2, ready2) = if (ready.isEmpty) (y, ready) else { // we have some values queued up, try feeding them to the Wye
          y.feed(ready) { // .feed is a tail recursive function
            case Await(req, recv, fb, c) => (req.tag: @annotation.switch) match {
              case 0 => // Left 
                val recv_ = recv.asInstanceOf[O => Wye[O,O2,O3]]
                (e: These[O,O2]) => e match { 
                  case These.This(o) => (None, Some(recv_(o)))
                  case These.That(_) => (None, None)
                  case These.Both(o,o2) => (Some(These.That(o2)), Some(recv_(o)))
                }
              case 1 => // Right 
                val recv_ = recv.asInstanceOf[O2 => Wye[O,O2,O3]]
                (e: These[O,O2]) => e match { 
                  case These.This(_) => (None, None)
                  case These.That(o2) => (None, Some(recv_(o2)))
                  case These.Both(o,o2) => (Some(These.This(o)), Some(recv_(o2)))
                }
              case 2 => // Both 
                val recv_ = recv.asInstanceOf[These[O,O2] => Wye[O,O2,O3]]
                (e: These[O,O2]) => (None, Some(recv_(e)))
            }
            case _ => _ => (None, None)
          }
        }
        val (h1Next, h2Next) = These.unalign(ready2)
        val (thisNext_, p2Next_) = (emitSeq(h1Next, t1), emitSeq(h2Next, t2))
        val thisNext = thisNext_.asInstanceOf[Process[F2,O]] 
        val p2Next = p2Next_.asInstanceOf[Process[F2,O2]]
        y2 match {
          case Await(req,_,_,_) => (req.tag: @annotation.switch) match {
            case 0 => // Left
              thisNext match {
                case AwaitF(reqL,recvL,fbL,cL) => // unfortunately, casts required here
                  await(reqL)(recvL andThen (_.wye(p2Next)(y2)), fbL.wye(p2Next)(y2), cL.wye(p2Next)(y2))
                case Halt => p2Next.kill ++ y2.disconnect
                case e@Emit(_,_) => thisNext.wye(p2Next)(y2) 
              }
            case 1 => // Right 
              p2Next match {
                case AwaitF(reqR,recvR,fbR,cR) => // unfortunately, casts required here
                  // in the event of a fallback or error, `y` will end up running the right's fallback/cleanup
                  // actions on the next cycle - it still needs that value on the right and will run any awaits
                  // to try to obtain that value!
                  await(reqR)(recvR andThen (p2 => thisNext.wye[F2,O2,O3](p2)(y2)), 
                               thisNext.wye(fbR)(y2), 
                               thisNext.wye(cR)(y2))
                case Halt => thisNext.kill ++ y2.disconnect
                case e@Emit(_,_) => thisNext.wye(p2Next)(y2) 
              }
            case 2 => thisNext match { // Both
              case Halt => p2Next |> y2.detachL
              case AwaitF(reqL, recvL, fbL, cL) => p2Next match {
                case Halt => 
                  Await(reqL, recvL andThen (_.wye(p2Next)(y2)), fbL.wye(p2Next)(y2), cL.wye(p2Next)(y2))
                // If both sides are in the Await state, we use the
                // Nondeterminism instance to request both sides
                // concurrently.
                case AwaitF(reqR, recvR, fbR, cR) =>
                  wrap(
                    F2.choose(
                      E.attempt(reqL.map(recvL)).map {
                        _.fold({ case End => fbL 
                                 case t: Throwable => cL ++ (throw t) 
                               }, 
                               e => e)
                      },
                      E.attempt(reqR.map(recvR)).map {
                        _.fold({ case End => fbR 
                                 case t: Throwable => cR ++ (throw t) 
                               }, 
                               e => e)
                      }
                    )
                  ).flatMap { 
                    _.fold(l => wrap(l._2).flatMap(p2 => l._1.wye(p2)(y2)),
                           r => wrap(r._1).flatMap(t => t.wye(r._2)(y2)))
                  }
              }
            }
          }
          case _ => thisNext.wye(p2Next)(y2)
        }
    }
    catch { case e: Throwable => this.kill ++ p2.kill ++ (throw e) }
  }

  /** Translate the request type from `F` to `G`, using the given polymorphic function. */
  def translate[G[_]](f: F ~> G): Process[G,O] = this match {
    case Emit(h, t) => Emit(h, t.translate(f))
    case Halt => Halt
    case Await(req, recv, fb, c) => 
      Await(f(req), recv andThen (_ translate f), fb translate f, c translate f)
  }

  /** 
   * Catch exceptions produced by this `Process`. The resulting `Process` halts after the
   * the first emitted failure, though this can handled and resumed by a subsequent `Process`. 
   */ 
  def attempt[F2[x]>:F[x]](implicit F: Catchable[F2]): Process[F2, Throwable \/ O] = 
  this match {
    case Emit(h, t) => Emit(h map (\/-(_)), t.attempt[F2])
    case Halt => Halt
    case Await(req, recv, fb, c) => 
      await(F.attempt(req))(
        _.fold(err => emit(-\/(err)), recv andThen (_.attempt[F2])), 
        fb.attempt[F2], c.attempt[F2])
  }

  /** 
   * Catch some of the exceptions generated by this `Process`, rethrowing any
   * not handled by the given `PartialFunction`. 
   */
  def handle[F2[x]>:F[x],O2>:O](f: PartialFunction[Throwable,O2])(implicit F: Catchable[F2]) = 
    attempt(F) |> (process1.lift { 
      case \/-(o) => o
      case -\/(err) => f.lift(err).getOrElse(throw err) })

  /** 
   * Map over this `Process` to produce a stream of `F`-actions, 
   * then evaluate these actions. 
   */
  def evalMap[F2[x]>:F[x],O2](f: O => F2[O2]): Process[F2,O2] = 
    map(f).eval

  /** 
   * Map over this `Process` to produce a stream of `F`-actions, 
   * then evaluate these actions, returning results in whatever
   * order they come back. `bufSize` controls the maximum number
   * of evaluations that can be queued at any one time. 
   */
  def gatherMap[F2[x]>:F[x],O2](bufSize: Int)(f: O => F2[O2])(
                                implicit F: Nondeterminism[F2]): Process[F2,O2] = 
    map(f).gather(bufSize)
 
  /** 
   * Collect the outputs of this `Process[F,O]`, given a `Monad[F]` in
   * which we can catch exceptions. This function is not tail recursive and
   * relies on the `Monad[F]` to ensure stack safety. 
   */
  final def collect[F2[x]>:F[x], O2>:O](implicit F: Monad[F2], C: Catchable[F2]): F2[IndexedSeq[O2]] = {
    def go(cur: Process[F2,O2], acc: collection.mutable.ListBuffer[O2]): F2[IndexedSeq[O2]] =
      cur match {
        case Emit(h,t) => 
          acc ++= h.asInstanceOf[Seq[O2]]
          go(t.asInstanceOf[Process[F2,O2]], acc)
        case Halt => F.point(acc.toIndexedSeq)
        case Await(req,recv,fb,c) => 
           F.bind (C.attempt(req.asInstanceOf[F2[AnyRef]])) {
             _.fold(
               { case End => go(fb.asInstanceOf[Process[F2,O2]], acc)
                 case err => c match {
                   case Halt => C.fail(err)
                   case _ => go(c.asInstanceOf[Process[F2,O2]] ++ wrap(C.fail(err)), new collection.mutable.ListBuffer()) 
                 }
               }, o => go(recv.asInstanceOf[AnyRef => Process[F2,O2]](o), acc))
           }
      }
    go(this, new collection.mutable.ListBuffer[O2]())
  }

  /** Run this `Process` solely for its final emitted value, if one exists. */
  final def runLast[F2[x]>:F[x], O2>:O](implicit F: Monad[F2], C: Catchable[F2]): F2[Option[O2]] = 
    F.map(this.last.collect[F2,O2])(_.lastOption)

  /** Run this `Process` solely for its final emitted value, if one exists, using `o2` otherwise. */
  final def runLastOr[F2[x]>:F[x], O2>:O](o2: => O2)(implicit F: Monad[F2], C: Catchable[F2]): F2[O2] = 
    F.map(this.last.collect[F2,O2])(_.lastOption.getOrElse(o2))

  /** Run this `Process`, purely for its effects. */
  final def run[F2[x]>:F[x]](implicit F: Monad[F2], C: Catchable[F2]): F2[Unit] = 
    F.void(drain.collect(F, C))

  /** Alias for `this |> process1.buffer(n)`. */
  def buffer(n: Int): Process[F,O] = 
    this |> process1.buffer(n)

  /** Alias for `this |> process1.bufferBy(f)`. */
  def bufferBy(f: O => Boolean): Process[F,O] = 
    this |> process1.bufferBy(f)

  /** Alias for `this |> process1.bufferAll`. */
  def bufferAll: Process[F,O] = 
    this |> process1.bufferAll

  /** Alias for `this |> process1.chunk(n)`. */
  def chunk(n: Int): Process[F,Vector[O]] = 
    this |> process1.chunk(n)

  /** Alias for `this |> process1.chunkBy(f)`. */
  def chunkBy(f: O => Boolean): Process[F,Vector[O]] = 
    this |> process1.chunkBy(f)

  /** Alias for `this |> process1.chunkAll`. */
  def chunkAll: Process[F,Vector[O]] = 
    this |> process1.chunkAll

  /** Ignores the first `n` elements output from this `Process`. */
  def drop(n: Int): Process[F,O] = 
    this |> processes.drop[O](n)

  /** Ignores elements from the output of this `Process` until `f` tests false. */
  def dropWhile(f: O => Boolean): Process[F,O] = 
    this |> processes.dropWhile(f)

  /** Skips any output elements not matching the predicate. */
  def filter(f: O => Boolean): Process[F,O] = 
    this |> processes.filter(f)

  /** Connect this `Process` to `process1.fold(b)(f)`. */
  def fold[B](b: B)(f: (B,O) => B): Process[F,B] = 
    this |> process1.fold(b)(f)
  
{'fraction_non_alphanumeric': 0.15057171107674366, 'fraction_numerical': 0.01526253074056233, 'mean_word_length': 2.415064935064935, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 364, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4812494', 'n_tokens_mistral': 14447, 'n_tokens_neox': 13770, 'n_words': 3946}