如何在Akka Streams中的Akka客户端WebSocket中使用Flow.map

编程问答 2022-08-15

问题描述

我有一个WebSocket服务器，我需要通过已建立的WebSocket连接向其发送和接收消息，因为Akka客户端WebSocket不能像Java中那样为我们提供常规的websocket.send()功能

所以我找到了这个解决方案How to use Akka-HTTP client websocket send message

效果很好，但是我是Akka流中的初学者，我正在尝试实现以下方案

在我的案例中，WebSocket服务器在ws：//0.0.0.0：8188上运行的示例

首先，我将向服务器发送一条消息以启动sessionID 

request# 1

{
        "janus" : "create","transaction" : "<random alphanumeric string>"
}

服务器将使用会话ID进行响应

response #1 
{
   "janus": "success","session_id": 2630959283560140,"transaction": "asqeasd4as3d4asdasddas","data": {
        "id": 4574061985075210
   }
}

然后基于ID 4574061985075210，我将发送另一条消息并接收更多信息

    request # 02 {
 "janus": "attach","session_id":${sessionId},"plugin":"janus.plugin.echotest","transaction":"asqeasd4as3d4asdasddas"     
}

response # 02 {
}
----

到目前为止，我可以显示request #1的响应，但我不知道如何使用从请求＃1获得的sessionID发起请求＃2并显示其响应

这是我的代码

    def main(args: Array[String]): Unit = {
        val url = "ws://0.0.0.0:8188"
        val req = WebSocketRequest(url,Nil,Option("janus-protocol"))
    
      

    implicit val materializer = ActorMaterializer()

    import system.dispatcher

    val webSocketFlow = Http().webSocketClientFlow(req)

    val messageSource: Source[Message,ActorRef] =
      Source.actorRef[TextMessage.Strict](bufferSize = 10,OverflowStrategy.fail)

    val messageSink: Sink[Message,NotUsed] =
      Flow[Message]
        .map{message =>
          println(s"Received text message: [$message]")
          val strjson = message.toString
          val jsonResponse = strjson.parseJson
          val jsonObj = jsonResponse.asJsObject
          val janus = jsonObj.fields("janus").convertTo[String]
          val data = jsonObj.fields("data").asJsObject
          val sessionID = data.fields("id")
          // i need to take this SessionId and send to the websocket established connection and receive its response
        }
        .to(Sink.ignore)

    val ((ws,upgradeResponse),closed) =
      messageSource
        .viaMat(webSocketFlow)(Keep.both)
        .toMat(messageSink)(Keep.both)
        .run()

    val connected = upgradeResponse.flatMap { upgrade =>
      if (upgrade.response.status == StatusCodes.SwitchingProtocols) {
        Future.successful(Done)
      } else {
        throw new RuntimeException(s"Connection Failed: ${upgrade.response.status}")
      }
    }

    val source =
      """{ "janus": "create","transaction":"d1403sa54a5s3d4as3das"}"""
    val jsonAst = source.parseJson


    ws ! TextMessage.Strict(jsonAst.toString())
   
  }

任何帮助将不胜感激

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）

akkaakka-streamscalawebsocket