scala - scalaz-stream 的 inflate 使用示例

Question

在以下 scalaz-stream 的使用示例中(取自 documentation )，如果输入和/或输出是 gzip 文件，我需要更改什么？换句话说，我如何使用 compress ？

import scalaz.stream._
import scalaz.concurrent.Task

val converter: Task[Unit] =
  io.linesR("testdata/fahrenheit.txt")
    .filter(s => !s.trim.isEmpty && !s.startsWith("//"))
    .map(line => fahrenheitToCelsius(line.toDouble).toString)
    .intersperse("\n")
    .pipe(text.utf8Encode)
    .to(io.fileChunkW("testdata/celsius.txt"))
    .run

// at the end of the universe...
val u: Unit = converter.run

Answer 1

压缩输出很容易。由于 compress.deflate() 是一个 Process1[ByteVector, ByteVector]，因此您需要将其插入您要发出 ByteVector 的管道中(就在 text.utf8Encode 之后，这是一个 Process1[String, ByteVector]):

val converter: Task[Unit] =
  io.linesR("testdata/fahrenheit.txt")
    .filter(s => !s.trim.isEmpty && !s.startsWith("//"))
    .map(line => fahrenheitToCelsius(line.toDouble).toString)
    .intersperse("\n")
    .pipe(text.utf8Encode)
    .pipe(compress.deflate())
    .to(io.fileChunkW("testdata/celsius.zip"))
    .run

对于inflate，您不能使用io.linesR 来读取压缩文件。您需要一个生成 ByteVector 而不是 String 的进程，以便将它们通过管道传输到 inflate 中。 (您可以为此使用 io.fileChunkR。)下一步是将未压缩的数据解码为 String(使用 text.utf8Decode示例)，然后使用 text.lines() 逐行发出文本。像这样的东西应该可以解决问题:

val converter: Task[Unit] =
  Process.constant(4096).toSource
    .through(io.fileChunkR("testdata/fahrenheit.zip"))
    .pipe(compress.inflate())
    .pipe(text.utf8Decode)
    .pipe(text.lines())
    .filter(s => !s.trim.isEmpty && !s.startsWith("//"))
    .map(line => fahrenheitToCelsius(line.toDouble).toString)
    .intersperse("\n")
    .pipe(text.utf8Encode)
    .to(io.fileChunkW("testdata/celsius.txt"))
    .run