流处理与增量I/O中的简单流转换

### 流处理与增量 I/O 中的简单流转换 #### 1. 流处理器与 Pull 数据类型 在进行 I/O 操作时，为了恢复我们习惯的高级风格，引入了流处理器的概念。流处理器指定了从一个流到另一个流的转换，这里的“流”是一个通用术语，可指可能由外部源延迟生成或提供的序列，比如文件中的行流、HTTP 请求、鼠标点击位置等。 为了表达流转换，我们引入了一个简单的数据类型 `Pull`： ```scala enum Pull[+O, +R]: case Result[+R](result: R) extends Pull[Nothing, R] case Output[+O](value: O) extends Pull[O, Unit] case FlatMap[X, +O, +R]( source: Pull[O, X], f: X => Pull[O, R] ) extends Pull[O, R] def step: Either[R, (O, Pull[O, R])] = this match case Result(r) => Left(r) case Output(o) => Right(o, Pull.done) case FlatMap(source, f) => source match case FlatMap(s2, g) => s2.flatMap(x => g(x).flatMap(y => f(y))).step case other => other.step match case Left(r) => f(r).step case Right((hd, tl)) => Right((hd, tl.flatMap(f))) @annotation.tailrec final def fold[A](init: A)(f: (A, O) => A): (R, A) = step match case Left(r) => (r, init) case Right((hd, tl)) => tl.fold(f(init, hd))(f) def toList: List[O] = fold(List.newBuilder[O])((bldr, o) => bldr += o)(1).result def flatMap[O2 >: O, R2]( f: R => Pull[O2, R2] ): Pull[O2, R2] = Pull.FlatMap(this, f) def >>[O2 >: O, R2](next: => Pull[O2, R2]): Pull[O2, R2] = flatMap(_ => next) def map[R2](f: R => R2): Pull[O, R2] = flatMap(r => Result(f(r))) ``` `Pull[O, R]` 可以输出任意数量的 `O` 类型的值，然后以单个 `R` 类型的值终止。可以通过 `step` 方法对 `Pull` 进行增量评估，要么输出最终的 `R` 值，要么输出一个 `O` 值和一个表示输入流剩余部分的新 `Pull`。 #### 2. 创建 Pull - **简单的 Pull**： - 输出无值并以单位值终止的 `Pull`： ```scala val done: Pull[Nothing, Unit] = Result(()) ``` - 输出单个值并连接的 `Pull`： ```scala scala> val p = Pull.Output(1) >> Pull.Output(2) val p: Pull[Int, Unit] = FlatMap(Output(1),...) scala> val q = p.toList val q: List[Int] = List(1, 2) ``` - **从列表和惰性列表创建 `Pull`**： ```scala def fromList[O](os: List[O]): Pull[O, Unit] = os match case Nil => done case hd :: tl => Output(hd) >> fromList(tl) def fromLazyList[O](os: LazyList[O]): Pull[O, Unit] = os match case LazyList() => done case hd #:: tl => Output(hd) >> fromLazyList(tl) ``` - **通用的 `Pull` 构造函数 `unfold`**： ```scala def unfold[O, R](init: R)(f: R => Either[R, (O, R)]): Pull[O, R] = f(init) match case Left(r) => Result(r) case Right((o, r2)) => Output(o) >> unfold(r2)(f) ``` #### 3. 无限流与部分评估 - **无限流构造函数**： ```scala def continually[A](a: A): Pull[A, Nothing] = Output(a) >> continually(a) enum Pull[+O, +R]: def repeat: Pull[O, R] = this >> repeat object Pull: def continually[A](a: A): Pull[O, Nothing] = Output(a).repeat ``` - **部分评估 `take` 方法**： ```scala def take(n: Int): Pull[O, Option[R]] = if n <= 0 then Result(None) else step match case Left(r) => Result(Some(r)) case Right((hd, tl)) => Output(hd) >> tl.take(n - 1) def uncons: Pull[Nothing, Either[R, (O, Pull[O, R])]] = Pull.done >> Result(step) def take(n: Int): Pull[O, Option[R]] = if n <= 0 then Result(None) else uncons.flatMap: case Left(r) => Result(Some(r)) case Right((hd, tl)) => Output(hd) >> tl.take(n - 1) ``` #### 4. Pull 转换 - **转换输出元素 `mapOutput`**： ```scala def mapOutput[O2](f: O => O2): Pull[O2, R] = uncons.flatMap: case Left(r) => Result(r) case Right((hd, tl)) => Output(f(hd)) >> tl.mapOutput(f) ``` - **过滤输出元素 `filter`**： ```scala def filter(p: O => Boolean): Pull[O, R] = uncons.flatMap: case Left(r) => Result(r) case Right((hd, tl)) => ```