CODEWITHSUNDEEP
scalaakkaakka-stream
softshipper
softshipper

Reputation: 34061

Why the Failure case never reached?

I have a websocket client, that looks as following: 

object Main extends App {

  private val sapServer = "127.0.0.1:8080"

  implicit val system = ActorSystem("WsSystem")
  implicit val materializer = ActorMaterializer()
  implicit val dispatcher = system.dispatcher

  RestartSource.withBackoff(
    minBackoff = 3.seconds,
    maxBackoff = 30.seconds,
    randomFactor = 0.2
  ) { () =>

    // Consider if the server supports websocket or not
    val (supported, source) = Source.tick(1.seconds, 15.seconds, TextMessage.Strict(Monoid.empty[String]))
      .viaMat(Http().webSocketClientFlow(WebSocketRequest(s"ws://$sapServer")))(Keep.right)
      .preMaterialize()

    supported.flatMap { upgrade =>
      //Switching from HTTP to WS protocol
      if (upgrade.response.status == StatusCodes.SwitchingProtocols)
        Future.successful(Done)
      else
        throw new RuntimeException(s"Connection failed: ${upgrade.response.status}")

    }
    source
  }.runWith(Sink.foreach(println))
    .onComplete {
      case Success(_) =>
        //Do nothing
        Unit
      case Failure(_) =>
        println("Probably server is down.")
    }


}

I haven't started the websocket server yet, however, I expect that it shows Probably server is down. on the console. But instead I've got: 

[WARN] [06/09/2019 15:06:25.823] [WsSystem-akka.actor.default-dispatcher-14] [RestartWithBackoffSource(akka://WsSystem)] Restarting graph due to failure. stack_trace:  (akka.stream.StreamTcpException: Tcp command [Connect(127.0.0.1:8080,None,List(),Some(10 seconds),true)] failed because of java.net.ConnectException: Connection refused)
[WARN] [06/09/2019 15:06:32.674] [WsSystem-akka.actor.default-dispatcher-13] [RestartWithBackoffSource(akka://WsSystem)] Restarting graph due to failure. stack_trace:  (akka.stream.StreamTcpException: Tcp command [Connect(127.0.0.1:8080,None,List(),Some(10 seconds),true)] failed because of java.net.ConnectException: Connection refused)
[WARN] [06/09/2019 15:06:45.367] [WsSystem-akka.actor.default-dispatcher-3] [RestartWithBackoffSource(akka://WsSystem)] Restarting graph due to failure. stack_trace:  (akka.stream.StreamTcpException: Tcp command [Connect(127.0.0.1:8080,None,List(),Some(10 seconds),true)] failed because of java.net.ConnectException: Connection refused)
[WARN] [06/09/2019 15:07:09.828] [WsSystem-akka.actor.default-dispatcher-13] [RestartWithBackoffSource(akka://WsSystem)] Restarting graph due to failure. stack_trace:  (akka.stream.StreamTcpException: Tcp command [Connect(127.0.0.1:8080,None,List(),Some(10 seconds),true)] failed because of java.net.ConnectException: Connection refused)

The question is, how to catch the StreamTcpException exception?

Upvotes: 2

Views: 355

Answers (1)

Matt Fowler
Matt Fowler

Reputation: 2723

The supported future is the one that is failing with the connection exception, not the future that RestartSource.withBackoff is creating. Doing something like:

supported.flatMap {...}.onComplete {
  case Failure(_) =>
    println("Probably server is down.")
}

Will print the message you expect.

Upvotes: 4

Related Questions