Scala:读取Enumerator[T]的一些数据并返回剩余的Enumerator[T]

Question

我使用的是playframework的异步I/O库，它使用迭代器和枚举器。我现在有一个IteratorT作为数据接收器(为了简化起见，假设它是一个将其内容存储到文件中的IteratorByte )。此IteratorByte被传递给处理写入的函数。

但在编写之前，我想在文件开头添加一些统计信息(为了简化起见，假设它是一个字节)，因此在将其传递给write函数之前，我以以下方式传递迭代器：

def write(value: Byte, output: Iteratee[Byte]): Iteratee[Byte] =
    Iteratee.flatten(output.feed(Input.El(value)))

现在，当我从磁盘读取存储的文件时，我得到了它的EnumeratorByte。首先，我希望读取并删除额外的数据，然后将EnumeratorByte的其余部分传递给一个处理读取的函数。因此，我还需要转换枚举数：

def read(input: Enumerator[Byte]): (Byte, Enumerator[Byte]) = {
   val firstEnumeratorEntry = ...
   val remainingEnumerator = ...
   (firstEnumeratorEntry, remainingEnumerator)
}

但我不知道该怎么做。如何从枚举器中读取一些字节并获取剩余的枚举器？

将IterateeByte替换为OutputStream，将EnumeratorByte替换为InputStream，这将非常容易：

def write(value: Byte, output: OutputStream) = {
    output.write(value)
    output
}
def read(input: InputStream) = (input.read,input)

但是我需要play框架的异步I/O。

Answer 1

这是通过在Iteratee和适当的(种类)状态累加器(这里是一个元组)中折叠来实现这一点的一种方法。

我去读取routes文件，第一个字节将作为Char读取，另一个字节将作为UTF-8字节字符串附加到String。

  def index = Action {
    /*let's do everything asyncly*/
    Async {
      /*for comprehension for read-friendly*/
      for (
        i <- read; /*read the file */
        (r:(Option[Char], String)) <- i.run /*"create" the related Promise and run it*/
      ) yield Ok("first : " + r._1.get + "\n" + "rest" + r._2) /* map the Promised result in a correct Request's Result*/
    }
  }


  def read = {
    //get the routes file in an Enumerator
    val file: Enumerator[Array[Byte]] = Enumerator.fromFile(Play.getFile("/conf/routes"))

    //apply the enumerator with an Iteratee that folds the data as wished
    file(Iteratee.fold((None, ""):(Option[Char], String)) { (acc, b) =>
       acc._1 match {
         /*on the first chunk*/ case None => (Some(b(0).toChar), acc._2 + new String(b.tail, Charset.forName("utf-8")))
         /*on other chunks*/ case x => (x, acc._2 + new String(b, Charset.forName("utf-8")))
       }
    })

  }

编辑

我找到了另一种使用Enumeratee的方法，但它需要创建2个Enumerator (一个是短暂的)。然而，它是不是更优雅一点。我们使用的是“某种”枚举类型，但Traversal级别的枚举比枚举级别更好(块级别)。我们使用take 1，它只占用1个字节，然后关闭流。在另一种情况下，我们使用drop来删除第一个字节(因为我们使用的是Enumerator[ArrayByte])

此外，现在read2的签名比您希望的更接近，因为它返回2个枚举数(离Promise不远，枚举数)。

def index = Action {
  Async {
    val (first, rest) = read2
    val enee = Enumeratee.map[Array[Byte]] {bs => new String(bs, Charset.forName("utf-8"))}

    def useEnee(enumor:Enumerator[Array[Byte]]) = Iteratee.flatten(enumor &> enee |>> Iteratee.consume[String]()).run.asInstanceOf[Promise[String]]

    for {
      f <- useEnee(first);
      r <- useEnee(rest)
    } yield Ok("first : " + f + "\n" + "rest" + r)
  }
}

def read2 = {
  def create = Enumerator.fromFile(Play.getFile("/conf/routes"))

  val file: Enumerator[Array[Byte]] = create
  val file2: Enumerator[Array[Byte]] = create

  (file &> Traversable.take[Array[Byte]](1), file2 &> Traversable.drop[Array[Byte]](1))

}

Answer 2

我想知道你是否可以从另一个角度来实现你的目标。

这个函数将使用剩余的枚举数，我们称其为remaining，假设它应用于一个迭代器来处理剩余部分：remaining |>> iteratee，产生另一个迭代器。让我们将生成的迭代器称为iteratee2...你能检查一下你是否能找到iteratee2的参考资料吗？如果是这种情况，那么您可以使用第一个迭代器head获取并处理第一个字节，然后通过flatMap组合head和iteratee2：

val head = Enumeratee.take[Byte](1) &>> Iteratee.foreach[Byte](println)
val processing = for { h <- head; i <- iteratee2 } yield (h, i)
Iteratee.flatten(processing).run

如果您不能获得iteratee2 -如果您的枚举器与您没有实现的被枚举器组合在一起-那么这种方法将不起作用。

Answer 3

实际上，我们喜欢Iteratees，因为它们组成。因此，与其从原始的迭代器创建多个Enumerator，不如按顺序组合两个迭代器(read-first和read-rest)，并用单个枚举器提供给它。

为此，您需要一个顺序组合方法，现在我将其称为andThen。下面是一个粗略的实现。请注意，返回未使用的输入有点苛刻，也许可以使用基于输入类型的类型类自定义行为。此外，它不会处理从第一个迭代器到第二个迭代器的剩余内容(Exercise :)。

object Iteratees {
  def andThen[E, A, B](a: Iteratee[E, A], b: Iteratee[E, B]): Iteratee[E, (A,B)] = new Iteratee[E, (A,B)] {
    def fold[C](
        done: ((A, B), Input[E]) => Promise[C],
        cont: ((Input[E]) => Iteratee[E, (A, B)]) => Promise[C],
        error: (String, Input[E]) => Promise[C]): Promise[C] = {

      a.fold(
        (ra, aleft) => b.fold(
          (rb, bleft) => done((ra, rb), aleft /* could be magicop(aleft, bleft)*/),
          (bcont) => cont(e => bcont(e) map (rb => (ra, rb))),
          (s, err) => error(s, err)
        ),
        (acont) => cont(e => andThen[E, A, B](acont(e), b)),
        (s, err) => error(s, err)
      )
    }
  }
}

现在，您只需使用以下内容：

object Application extends Controller {

  def index = Action { Async {

    val strings: Enumerator[String] = Enumerator("1","2","3","4")
    val takeOne = Cont[String, String](e => e match {
      case Input.El(e) => Done(e, Input.Empty)
      case x => Error("not enough", x)
    })
    val takeRest = Iteratee.consume[String]()
    val firstAndRest = Iteratees.andThen(takeOne, takeRest)

    val futureRes = strings(firstAndRest) flatMap (_.run)

    futureRes.map(x => Ok(x.toString)) // prints (1,234)
  } }

}