CODEWITHSUNDEEP
scalaakkaakka-streamakka-http
abergmeier
abergmeier

Reputation: 14042

Websocket with Graph DSL

I am trying to implement a Websocket Login flow with Akka Flow. I get a myriad of nasty runtime exceptions around Inlets, Outlets and Connection issues. My latest is:

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

Snippet:

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)
    })
}

I might be using the DSL totally wrong as I have not yet found a single writeup which explains Graphs, Shapes, Flows, Materialized values in depth. Could someone point out what I am doing wrong or perhaps how this is supposed to be written?

EDIT 1:

Have now replaced -> with ~> and get nasty compile errors:

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)
    })
}

which leads to

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

Upvotes: 1

Views: 274

Answers (1)

Aleksey Isachenkov
Aleksey Isachenkov

Reputation: 1240

You need something like this:

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)
        })
}

Keep in mind, that -> and ~> are different operators (you should use ~>). And you need to add a shape to the builder only if you are going to manually connect the shape inlets and outlets.

Upvotes: 0

Related Questions