如何从 akka Source[ByteString， _] 中提取 Future[String]？

Question

我正在尝试使用akka流传输一个文件，在将流的结果提取到Future[String]时遇到了一个小问题:

def streamMigrationFile(source: Source[ByteString, _]): Future[String] = {
  var fileString = ""
  val sink = Sink.foreach[ByteString](byteString => fileString = 
    fileString.concat(byteString.decodeString("US-ASCII")))
  source.runWith(sink)
}

我得到一个编译错误:

Expression of type Future[Done] does not conform to expected type Future[String]

任何人都可以帮助我了解我做错了什么，以及我需要做些什么来提取流的结果？

Answer 1

如果您查看Sink.foreach的定义，您会发现评估类型是Sink[T，Future[Done]]，这意味着流中元素的计算结果会发生什么并不重要。定义如下：

def foreach[T](f: T ⇒ Unit): Sink[T, Future[Done]]

另一方面，Sink的定义。折叠计算为未来[U]，即U0的类型。换句话说，您可以在处理结束时定义未来的类型。

以下是Sink.fold的定义（和实现）：

def fold[U, T](zero: U)(f: (U, T) ⇒ U): Sink[T, Future[U]] =
          Flow[T].fold(zero)(f).toMat(Sink.head)(Keep.right).named("foldSink")

根据上面的实现，您可以看到要在具体化中保存的类型是< code>Future[U]，因为< code>Keep.right意味着类似于这样的内容:“我不在乎传入的元素是< code>Ts(或者在您的情况下是< code > ByteString )I(流)将给您< code>Us(或者在您的情况下是< code>String)..当我完成时(在< code >未来)”

以下是替换＜code＞接收器的工作示例。foreach带有接收器。折叠，并将整个表达式求值为未来[String]

def streamMigrationFile(source: Source[ByteString, _]): Future[String] = {

  var fileString = ""

  //def foreach[T](f: T ⇒ Unit): Sink[T, Future[Done]]
  val sinkForEach: Sink[ByteString, Future[Done]] = Sink.foreach[ByteString](byteString => fileString =
    fileString.concat(byteString.decodeString("US-ASCII")))

  /*
    def fold[U, T](zero: U)(f: (U, T) ⇒ U): Sink[T, Future[U]] =
      Flow[T].fold(zero)(f).toMat(Sink.head)(Keep.right).named("foldSink")
   */
  val sinkFold: Sink[ByteString, Future[String]] = Sink.fold("") { case (acc, str) =>
    acc + str
  }

  val res1: Future[Done] = source.runWith(sinkForEach)
  val res2: Future[String] = source.runWith(sinkFold)

  res2

}

Answer 2

如果我的猜测是正确的，您希望将整个文件内容流式传输到字符串中。这最好通过Sink.fold来实现，而不是使用Sink.foreach。示例如下。

def streamMigrationFile(source: Source[ByteString, _]): Future[String] = {
  val sink = Sink.fold[String, ByteString]("") { case (acc, str) =>
    acc + str.decodeString("US-ASCII")
  }
  source.runWith(sink)
}

您可能已经意识到了这一点，但是您的文件需要放入内存中才能使程序正常运行。