播放迭代:将字节流重新组合成大小可变的块

Question

假设我有一个枚举器[数组[字节]]，产生可变大小的字节数组块（例如控制器中接受的POST正文）

我知道这个流实际上包含数据包，每个数据包在哪里：

4 个字节，表示
• 整数，表示数据包正文大小的大小（以字节为单位）
• 该大小的数据包正文

每个数据包可以有不同的正文大小。

我们如何实现一个将初始流转换为字节数组流的枚举者，其中每个数组都是一个数据包正文。

带有 ints 的简化示例（第一个是表示数据包正文大小的 int）：

List(1, 2, 3), List(4, 5), List(6), List(2, 8), List(9) -> List(2) List(4, 5 ,6), List(8, 9)

Answer 1

我在尝试从 HTTP 响应流解析 JPG 标头时遇到了完全相同的问题。

首先，您需要使用 Enumerator.mapConcat 将流转换为字节而不是块。然后，您必须根据第一个字节按组对字节进行分组。为此，有一个函数Enumeratee.grouped，它将Iteratee[From, To]作为参数，并反复使用它来处理传入的From流并将其转换为To流。

这就是我想出的：

val takeSegment: Iteratee[Byte, List[Byte]] = for {
    //find segment length (the first four bytes of the segment, assumed unsigned and bigEndian)
    size <- Enumeratee.take(4) &>> Iteratee.fold[Byte, Int](0){ 
              case (acc, b) => acc * 256 + (b & 0xFF)
            }
    //fetch rest of segment
    rest <- Enumeratee.take(size) &>> Iteratee.getChunks[Byte]
    } yield rest
val segmentStream: Enumerator[List[Byte]] = 
                    inputStream &> 
                    // mapConcat takes an argument of type From => Seq[To]
                    Enumeratee.mapConcat[Array[Byte], Byte](bytes => bytes) &>
                    Enumeratee.grouped[Byte, List[Byte]](takeSegment)

当然，如果你的整数编码方式不同（我假设是无符号的bigEndian，但你可能有不同的情况），你需要相应地改变 的大小行以进行理解。