scalaz-stream的膨胀的用法示例

Question

在以下用法示例中scalaz-stream(取自文档),如果输入和/或输出是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用来读取压缩文件.你需要一个生成ByteVectors而不是Strings 的进程来管理它们inflate.(您可以使用io.fileChunkR它.)下一步是将未压缩数据解码为Strings(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