ConsumeQueueBehavior.scala

def consumeQueueBehavior(queue: Queue[Future], 
                         msgSink: ActorRef[Received]): Behavior[Try[String]] =
  Behaviors.setup { ctx =>
    import ctx.executionContext

    ctx.log.info("[queue] receiving message")
    queue.read().andThen { case result => ctx.self ! result }

    Behaviors
      .receiveMessage[Try[String]] {
        case Success(msg) =>
          msgSink ! Received(msg)
          consumeQueueBehavior(queue, msgSink)

        case Failure(e) =>
          ctx.log.info(s"[queue] failure: ${e.getMessage}")
          throw e
      }
      .receiveSignal {
        case (_, PostStop) =>
          ctx.log.info("[queue-stop] closing")
          Await.result(queue.close(), 1.minute)
          ctx.log.info("[queue-stop] closed")
          Behaviors.same
      }
  }