如何将 Source[ByteString, Any] 转换为 InputStream

Question

akka-http 表示使用 multipart/form-data 编码为 Source[ByteString, Any] 上传的文件。我需要使用需要 InputStream 的 Java 库对其进行解组。

如何将 Source[ByteString, Any] 变成 InputStream？

Answer 1

您可以尝试使用写入 PipedOutputStream 的 OutputStreamSink 并将其馈送到其他代码用作其输入流的 PipedInputStream 中。这是一个有点粗糙的想法，但它可以工作。代码如下所示：

import akka.util.ByteString
import akka.stream.scaladsl.Source
import java.io.PipedInputStream
import java.io.PipedOutputStream
import akka.stream.io.OutputStreamSink
import java.io.BufferedReader
import java.io.InputStreamReader
import akka.actor.ActorSystem
import akka.stream.ActorFlowMaterializer

object PipedStream extends App{
  implicit val system = ActorSystem("flowtest")
  implicit val mater = ActorFlowMaterializer()

  val lines = for(i <- 1 to 100) yield ByteString(s"This is line $i\n")
  val source = Source(lines)

  val pipedIn = new PipedInputStream()
  val pipedOut = new PipedOutputStream(pipedIn)      
  val flow = source.to(OutputStreamSink(() => pipedOut))
  flow.run()

  val reader = new BufferedReader(new InputStreamReader(pipedIn))
  var line:String = reader.readLine
  while(line != null){
    println(s"Reader received line: $line")
    line = reader.readLine
  }           
}

Answer 2

您可以从 ByteString 中提取一个 interator，然后获取 InputStream。像这样（伪代码）：

source.map { data: ByteString =>
  data.iterator.asInputStream
}

更新

以 Multipart.FormData

开头的更详细的示例

def isSourceFromFormData(formData: Multipart.FormData): Source[InputStream, Any] = 
 formData.parts.map { part => 
   part.entity.dataBytes
   .map(_.iterator.asInputStream)
}.flatten(FlattenStrategy.concat)

Answer 3

从版本 2.x 开始，您可以使用以下代码实现此目的：

import akka.stream.scaladsl.StreamConverters
...
val inputStream: InputStream = entity.dataBytes
        .runWith(
           StreamConverters.asInputStream(FiniteDuration(3, TimeUnit.SECONDS))
        )

参见：http://doc.akka.io/docs/akka-stream-and-http-experimental/2.0.1/scala/migration-guide-1.0-2.x-scala.html

注意：在 2.0.2 版本中被破坏并在 2.4.2 中得到修复

如何将 Source[ByteString, Any] 转换为 InputStream

How to convert Source[ByteString, Any] to InputStream

scala

akka

akka-stream