使用akka-http websockets上传和处理文件

Question

我正在使用一些示例Scala代码来使服务器通过websocket接收文件，临时存储该文件，在其上运行bash脚本，然后通过TextMessage返回stdout。

示例代码取自this github project。

我在echoService中稍微编辑了代码，以便它运行另一个处理临时文件的函数。

object WebServer { 
    def main(args: Array[String]) { 

    implicit val actorSystem = ActorSystem("akka-system") 
    implicit val flowMaterializer = ActorMaterializer() 

    val interface = "localhost" 
    val port = 3000 

    import Directives._ 

    val route = get { 
     pathEndOrSingleSlash { 
     complete("Welcome to websocket server") 
     } 
    } ~ 
     path("upload") { 
     handleWebSocketMessages(echoService) 
     } 

     val binding = Http().bindAndHandle(route, interface, port) 
     println(s"Server is now online at http://$interface:$port\nPress RETURN to stop...") 
     StdIn.readLine() 

     binding.flatMap(_.unbind()).onComplete(_ => actorSystem.shutdown()) 
     println("Server is down...") 

    } 

    implicit val actorSystem = ActorSystem("akka-system") 
    implicit val flowMaterializer = ActorMaterializer() 


    val echoService: Flow[Message, Message, _] = Flow[Message].mapConcat { 

     case BinaryMessage.Strict(msg) => { 
     val decoded: Array[Byte] = msg.toArray 
     val imgOutFile = new File("/tmp/" + "filename") 
     val fileOuputStream = new FileOutputStream(imgOutFile) 
     fileOuputStream.write(decoded) 
     fileOuputStream.close() 
     TextMessage(analyze(imgOutFile)) 
     } 

     case BinaryMessage.Streamed(stream) => { 

     stream 
      .limit(Int.MaxValue) // Max frames we are willing to wait for 
      .completionTimeout(50 seconds) // Max time until last frame 
      .runFold(ByteString(""))(_ ++ _) // Merges the frames 
      .flatMap { (msg: ByteString) => 

      val decoded: Array[Byte] = msg.toArray 
      val imgOutFile = new File("/tmp/" + "filename") 
      val fileOuputStream = new FileOutputStream(imgOutFile) 
      fileOuputStream.write(decoded) 
      fileOuputStream.close() 
      Future(Source.single("")) 
     } 
     TextMessage(analyze(imgOutFile)) 
     } 


     private def analyze(imgfile: File): String = { 
     val p = Runtime.getRuntime.exec(Array("./run-vision.sh", imgfile.toString)) 
     val br = new BufferedReader(new InputStreamReader(p.getInputStream, StandardCharsets.UTF_8)) 
     try { 
      val result = Stream 
      .continually(br.readLine()) 
      .takeWhile(_ ne null) 
      .mkString 
      result 

     } finally { 
      br.close() 
     } 
     } 
    } 




} 

在测试过程中使用的WebSocket黑暗终端，case BinaryMessage.Strict工作正常。

问题：然而，case BinaryMessage.Streaming没有完成运行analyze功能，导致来自服务器的空白响应之前写入文件。

我正试图围绕Akka-HTTP中的Flow如何使用期货，但我没有尝试通过所有官方文档获得很多运气。

目前，.mapAsync看起来很有前途，或者基本上找到了连接未来的方法。

我真的很感谢一些见解。

Answer 1

是的，mapAsync会在这个场合帮助你。它是一个组合器，可以在您的流中执行Future（可能并行），并在输出端显示它们的结果。

在您的情况下，为了使事物变得均匀并使类型检查程序开心，您需要将Strict案例的结果包装为Future.successful。

要快速解决您的代码可能是：

val echoService: Flow[Message, Message, _] = Flow[Message].mapAsync(parallelism = 5) { 

    case BinaryMessage.Strict(msg) => { 
     val decoded: Array[Byte] = msg.toArray 
     val imgOutFile = new File("/tmp/" + "filename") 
     val fileOuputStream = new FileOutputStream(imgOutFile) 
     fileOuputStream.write(decoded) 
     fileOuputStream.close() 
     Future.successful(TextMessage(analyze(imgOutFile))) 
    } 

    case BinaryMessage.Streamed(stream) => 

     stream 
     .limit(Int.MaxValue) // Max frames we are willing to wait for 
     .completionTimeout(50 seconds) // Max time until last frame 
     .runFold(ByteString(""))(_ ++ _) // Merges the frames 
     .flatMap { (msg: ByteString) => 

     val decoded: Array[Byte] = msg.toArray 
     val imgOutFile = new File("/tmp/" + "filename") 
     val fileOuputStream = new FileOutputStream(imgOutFile) 
     fileOuputStream.write(decoded) 
     fileOuputStream.close() 
     Future.successful(TextMessage(analyze(imgOutFile))) 
     } 
    }