如何编写一个枚举器来沿着不同的边界对一个枚举器进行分块

Question

因此,Play2.0 Enumeratee页面显示了使用&>或through方法将其更改Enumerator[String]为Enumerator[Int]:的示例:

val toInt: Enumeratee[String,Int] = Enumeratee.map[String]{ s => s.toInt }
val ints: Enumerator[Int] = strings &> toInt

还有一个Enumeratee.grouped枚举器可以从各个元素创建块的枚举器.这似乎工作正常.

但我所看到的是,通常的输入将是Array[Byte](由Enumerator.fromFileand和返回Enumerator.fromStream)的形式.考虑到这一点,我想采取这些Array[Byte]输入并将它们变成一个Enumerator[String],例如每个字符串是一行(以a结尾'\n').线条和Array[Byte]元素的边界通常不匹配.如何编写可以将分块数组转换为分块字符串的枚举器？

目的是在每个行Array[Byte]变为可用时将这些行重新组合回浏览器,并保留不属于完整行的剩余字节,直到下一个输入块出现.

理想情况下,我希望有一个方法给出一个iter: Iteratee[Array[Byte], T]和一个Enumerator[Array[Byte]]会给我一个Enumerator[T],我的T元素被解析iter.

附加信息:我有一点时间来清理我的代码,这是我正在尝试做的一个具体的例子.我有以下迭代检测下一行:

import play.api.libs.iteratee._
type AB = Array[Byte]

def takeWhile(pred: Byte => Boolean): Iteratee[AB, AB] = {
  def step(e: Input[AB], acc: AB): Iteratee[AB, AB] = e match {
    case Input.EOF => Done(acc, Input.EOF)
    case Input.Empty => Cont(step(_, acc))
    case Input.El(arr) =>
      val (taking, rest) = arr.span(pred)
      if (rest.length > 0) Done(acc ++ taking, Input.El(rest)) 
      else Cont(step(_, acc ++ taking)) 
  }
  Cont(step(_, Array()))
}

val line = for {
  bytes <- takeWhile(b => !(b == '\n' || b == '\r'))
  _     <- takeWhile(b =>   b == '\n' || b == '\r')
} yield bytes

而我想做的是这样的事情:

Ok.stream(Enumerator.fromFile(filename) &> chunkBy(line)).as("text/plain")

Answer 1

https://github.com/playframework/Play20/commit/f979006a7e2c1c08ca56ee0bae67b5463ee099c1#L3R131与您正在做的事情类似.我修复分组以处理剩余的输入.代码基本上看起来像:

val upToNewLine = 
  Traversable.splitOnceAt[String,Char](_ != '\n')  &>>
  Iteratee.consume()

Enumeratee.grouped(upToNewLine)

另外,我必须以同样的方式修复重复