scala - actors - akka microservices
Cómo convertir Source[ByteString, Any] a InputStream (3)
akka-http representa un archivo cargado utilizando la codificación multipart / form-data como Source[ByteString, Any]
. Necesito descomprimirlo utilizando la biblioteca de Java que espera un InputStream
.
¿Cómo se puede convertir Source[ByteString, Any]
en InputStream
?
A partir de la versión 2.x logras esto con el siguiente código:
import akka.stream.scaladsl.StreamConverters
...
val inputStream: InputStream = entity.dataBytes
.runWith(
StreamConverters.asInputStream(FiniteDuration(3, TimeUnit.SECONDS))
)
Nota: se rompió en la versión 2.0.2 y se corrigió en 2.4.2
Podría extraer un interator de ByteString y luego obtener InputStream. Algo así (pseudocódigo):
source.map { data: ByteString =>
data.iterator.asInputStream
}
Actualizar
Una muestra más elaborada que comienza con un Multipart.FormData
def isSourceFromFormData(formData: Multipart.FormData): Source[InputStream, Any] =
formData.parts.map { part =>
part.entity.dataBytes
.map(_.iterator.asInputStream)
}.flatten(FlattenStrategy.concat)
Puede intentar usar un OutputStreamSink
que escribe en un PipedOutputStream
y PipedOutputStream
a un PipedInputStream
que su otro código usa como su flujo de entrada. Es una idea un poco áspera pero podría funcionar. El código se vería así:
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
}
}