Websocket с графическим DSL - PullRequest
Новая
       8

Websocket с графическим DSL

0 голосов
/ 04 сентября 2018

Я пытаюсь реализовать поток входа в Websocket с помощью Akka Flow. Я получаю множество неприятных исключений во время выполнения проблем с входами, выходами и подключением. Мой последний: 

java.lang.IllegalStateException: Illegal GraphDSL usage. Inlets [Map.in] were not returned in the resulting shape and not connected.

Отрывок: 

object Login {

    def graph(system: ActorSystem, future: Future[LoginCommand.UserData], socketUrl: String) =
        Source.fromGraph(GraphDSL.create() { implicit builder: GraphDSL.Builder[NotUsed] =>
            import GraphDSL.Implicits._

            val in = Source.fromFuture(future)
            in.named("LoginData")

            val fanIn = Zip[LoginResponse, LoginCommand.UserData]
            val exasolLogin = builder.add(Http(system).webSocketClientFlow(WebSocketRequest(socketUrl)))
            val encryptLoginData = FlowShape(exasolLogin.in, fanIn.out)

            val exasolAnnounce = Http(system).webSocketClientFlow(WebSocketRequest(socketUrl))

            val announceLogin = Source.single(LoginCommand)

            in -> fanIn
            announceLogin -> exasolAnnounce -> fanIn

            fanIn -> encryptLoginData -> exasolLogin

            SourceShape(exasolLogin.out)
    })
}

Возможно, я неправильно использовал DSL, так как я еще не нашел ни одной записи, в которой подробно объясняются графики, формы, потоки, материализованные значения. Может кто-то указать, что я делаю неправильно или, возможно, как это должно быть написано?

РЕДАКТ. 1 :

Теперь заменили -> на ~> и получили неприятные ошибки компиляции: 

object Login {

    def graph(system: ActorSystem, future: Future[LoginCommand.UserData], socketUrl: String) =
        Source.fromGraph(GraphDSL.create() { implicit builder: GraphDSL.Builder[NotUsed] =>
            import GraphDSL.Implicits._

            val in = Source.fromFuture(future)
            in.named("LoginData")

            val fanIn = builder.add(Zip[LoginResponse, LoginCommand.UserData])
            val exasolLogin = Http(system).webSocketClientFlow(WebSocketRequest(socketUrl))
            val encryptLoginData = Flow[(LoginResponse, LoginCommand.UserData)].map(data => data._1)
            val loginDataMessage = Flow[LoginCommand.UserData].map(data => TextMessage("bar"))

            val exasolAnnounce = Http(system).webSocketClientFlow(WebSocketRequest(socketUrl))
            val announceResponse = Flow[Message].map(data => LoginResponse("key", "mod", "exp"))

            val loginMessage = Flow[LoginCommand].map(data => TextMessage("foo"))
            val session = builder.add(Flow[Message].map(data => LoginCommand.SessionData(0, 1, "2", "db", "w", 59, 546, 45, "q", "TZ", "TZB")))

            in ~> fanIn.in1
            Source.single(LoginCommand) ~> loginMessage ~> exasolAnnounce ~> announceResponse ~> fanIn.in0
            fanIn.out ~> encryptLoginData ~> loginDataMessage ~> exasolLogin ~> session

            SourceShape(session.out)
    })
}

, что приводит к 

exasol-client/LoginGraph.scala:42: error: overloaded method value ~> with alternatives:
  (to: akka.stream.SinkShape[exasol.LoginCommand.type])(implicit b: akka.stream.scaladsl.GraphDSL.Builder[_])Unit <and>
  (to: akka.stream.Graph[akka.stream.SinkShape[exasol.LoginCommand.type], _])(implicit b: akka.stream.scaladsl.GraphDSL.Builder[_])Unit <and>
  [Out](flow: akka.stream.FlowShape[exasol.LoginCommand.type,Out])(implicit b: akka.stream.scaladsl.GraphDSL.Builder[_])akka.stream.scaladsl.GraphDSL.Implicits.PortOps[Out] <and>
  [Out](junction: akka.stream.UniformFanOutShape[exasol.LoginCommand.type,Out])(implicit b: akka.stream.scaladsl.GraphDSL.Builder[_])akka.stream.scaladsl.GraphDSL.Implicits.PortOps[Out] <and>
  [Out](junction: akka.stream.UniformFanInShape[exasol.LoginCommand.type,Out])(implicit b: akka.stream.scaladsl.GraphDSL.Builder[_])akka.stream.scaladsl.GraphDSL.Implicits.PortOps[Out] <and>
  [Out](via: akka.stream.Graph[akka.stream.FlowShape[exasol.LoginCommand.type,Out],Any])(implicit b: akka.stream.scaladsl.GraphDSL.Builder[_])akka.stream.scaladsl.GraphDSL.Implicits.PortOps[Out] <and>
  [U >: exasol.LoginCommand.type](to: akka.stream.Inlet[U])(implicit b: akka.stream.scaladsl.GraphDSL.Builder[_])Unit
 cannot be applied to (akka.stream.FlowShape[exasol.LoginCommand,akka.http.scaladsl.model.ws.TextMessage.Strict])
            Source.single(LoginCommand) ~> loginMessage ~> exasolAnnounce ~> announceResponse ~> fanIn.in0

1 Ответ

0 голосов
/ 04 сентября 2018

Вам нужно что-то вроде этого: 

object Login {
    def graph(system: ActorSystem, future: Future[LoginCommand.UserData], socketUrl: String): Source[Message, NotUsed] =
        Source.fromGraph(GraphDSL.create() { implicit builder: GraphDSL.Builder[NotUsed] =>
            import GraphDSL.Implicits._

            val in = Source.fromFuture(future)
            in.named("LoginData")

            val fanIn = builder.add(Zip[LoginResponse, LoginCommand.UserData])
            val exasolLogin = builder.add(Http(system).webSocketClientFlow(WebSocketRequest(socketUrl)))
            val encryptLoginData = Flow[(LoginResponse, LoginCommand.UserData)].map(data => TextMessage(data.toString)) //stub
            val encryptAnnounceData = Flow[LoginCommand].map(data => TextMessage(data.toString)) //stub
            val decryptAnnounceData = Flow[Message].map(message => LoginResponse(message)) //stub
            val exasolAnnounce = Http(system).webSocketClientFlow(WebSocketRequest(socketUrl))
            val announceLogin = Source.single(LoginCommand)

            in ~> fanIn.in1
            announceLogin ~> encryptAnnounceData ~> exasolAnnounce ~> decryptAnnounceData ~> fanIn.in0
            fanIn.out ~> encryptLoginData ~> exasolLogin

            SourceShape(exasolLogin.out)
        })
}

Имейте в виду, что -> и ~> - разные операторы (вы должны использовать ~>). И вам нужно добавлять фигуру в конструктор, только если вы собираетесь вручную соединять входы и выходы фигур.

...