Fetching available elements in scalaz-stream

available.markdown

  case class Env[-I,-I2]() {
    sealed trait Y[-X] {
      def tag: Int
      def fold[R](l: => R, r: => R, both: => R): R
    }
    sealed trait T[-X] extends Y[X]
    sealed trait Is[-X] extends T[X]
    case object Left extends Is[I] {
      def tag = 0
      def fold[R](l: => R, r: => R, both: => R): R = l
    }
    case object Right extends T[I2] {
      def tag = 1
      def fold[R](l: => R, r: => R, both: => R): R = r
    }
    case object Both extends Y[ReceiveY[I,I2]] {
      def tag = 2
      def fold[R](l: => R, r: => R, both: => R): R = both
    }
  }

  private val Left_ = Env[Any,Any]().Left
  private val Right_ = Env[Any,Any]().Right
  private val Both_ = Env[Any,Any]().Both

  def Get[I]: Env[I,Any]#Is[I] = Left_
  def L[I]: Env[I,Any]#Is[I] = Left_
  def R[I2]: Env[Any,I2]#T[I2] = Right_
  def Both[I,I2]: Env[I,I2]#Y[ReceiveY[I,I2]] = Both_

So Get[I], which is the primitive request that can be made by Process1[I,_], only fetches a single value. So we have await1 being implemented directly in terms of this primitive:

def await1[A]: Process1[A,A] = eval(Get[I])

What I am saying is let's make Get[I] give us a Seq[I] of the available elements.

  case class Env[-I,-I2]() {
    sealed trait Y[-X] {
      def tag: Int
      def fold[R](l: => R, r: => R, both: => R): R
    }
    sealed trait T[-X] extends Y[X]
    sealed trait Is[-X] extends T[X]
    case object Left extends Is[Seq[I]] {
      def tag = 0
      def fold[R](l: => R, r: => R, both: => R): R = l
    }
    case object Right extends T[Seq[I2]] {
      def tag = 1
      def fold[R](l: => R, r: => R, both: => R): R = r
    }
    case object Both extends Y[ReceiveY[Seq[I],Seq[I2]]] {
      def tag = 2
      def fold[R](l: => R, r: => R, both: => R): R = both
    }
  }

  private val Left_ = Env[Any,Any]().Left
  private val Right_ = Env[Any,Any]().Right
  private val Both_ = Env[Any,Any]().Both

  def Get[I]: Env[I,Any]#Is[Seq[I]] = Left_
  def L[I]: Env[I,Any]#Is[Seq[I]] = Left_
  def R[I2]: Env[Any,I2]#T[Seq[I2]] = Right_
  def Both[I,I2]: Env[I,I2]#Y[ReceiveY[Seq[I],Seq[I2]]] = Both_

Now, available is primitive, and it returns however many elements are currently available:

def available[A]: Process1[A,Seq[A]] = eval(Get[I])

def await1[A]: Process1[A,A] = 
  available[A] flatMap { emitAll }

def exactly[A](n: Int): Process1[A, (Seq[A], Seq[A])] = 
  ??
  // calls available until `n` elements accumulate, then emits exactly that size sequence, along with any 
  // trailing elements in the final available chunk

// likewise for tee/wye, we have
def availableL[A]: Tee[A,Any,Seq[A]] = ...
def availableR[A]: Tee[Any,A,Seq[A]] = ...