ConnectToQueueBehavior.scala

def connectToQueueBehavior(connector: QueueConnector[Future], 
                           msgSink: ActorRef[Received]): Behavior[Nothing] = {
  Behaviors.setup[Try[Queue[Future]]] { ctx =>
    import ctx.executionContext

    ctx.log.info("[queue-start] connecting")
    connector.connect.andThen { case result => ctx.self ! result }

    Behaviors.receiveMessage {
      case Success(queue) =>
        ctx.log.info("[queue-start] connected")

        val consumeActor = ctx.spawn(consumeQueueBehavior(queue, msgSink), 
                                     "consume-queue")
        ctx.watch(consumeActor)

        // we can either not handle Terminated, which will cause 
        // DeathPactException to be thrown and propagated or rethrow the 
        // original exception
        Behaviors.receiveSignal {
          case (_, t @ Terminated(_)) =>
            t.failure.foreach(throw _)
            Behaviors.empty
        }
      case Failure(e) =>
        ctx.log.info("[queue-start] failure")
        throw e
    }
  }
}.narrow[Nothing]