Akka-http 异常处理 (Scala)

Question

我正在使用 Akka-hhtp (scala) 将多个 http 批处理请求异步发送到 API，并且想知道当响应代码不是 200 OK 时处理异常的正确方法是什么。

下面是一些伪代码来证明我的观点。

/* Using For comprehension here because the server API has restriction on the amount of data we can send and the time it takes them to process each request. So they require us to send multiple mini requests instead. If one of those fails, then our entire job should fail.*/

val eventuallyResponses = for {
    batches <- postBatch(payload)
} yield batches

val eventualResponses = Future.sequence(eventuallyResponses)

/* Do I need to recover here? If I don't, will the actor system terminate? */
eventualResponses.recover { case es =>
   log.warn("some message")
   List()
}

/* As I said I need to wait for all mini batch requests to complete. If one response is different than 200, then the entire job should fail. */
val result = Await.result(eventualResponses, 10.minutes)


actorSystem.terminate().oncomplete{
  case Success(_) =>
      if (result.isEmpty) =>
          /* This doesn't seem to interrupt the program */
          throw new RuntimeException("POST failed")
      } else {
          log.info("POST Successful")
      }
   case Failure(ex) =>
      log.error("error message $ex")
      throw ex
}

def postBatch(payload) = {
    val responseFuture: Future[HttpResponse] = httpClient.post(payload)

     responseFuture.flatMap{ res =>
       res.status match {
         case StatusCodes.OK => Future.successful(res)
         case _ => Future.failed(new RuntimeException("error message"))
       }
      }
}

当我们收到与 OK 不同的 StatusCodes 时，上面的代码会抛出异常。它确实通过了result.isEmpty true 的分支，但它似乎并没有停止/中断程序的执行。我需要它来做到这一点，因为这被安排为 Autosys 作业，如果至少有一个批处理请求返回的响应与 200 OK 不同，我需要使作业失败。

如果我不recover 并让异常被抛出（当我们收到非 200 状态码时），Actor System 会正确终止吗？

您知道执行上述操作的好方法吗？

谢谢:)

Answer 1

据我了解您的问题，如果某些响应的状态不是 200，您需要从 main 正文中抛出异常。

def postBatch(payload: HttpRequest)(implicit system: ActorSystem, ec: ExecutionContext): Future[HttpResponse] = {
    Http().singleRequest(payload).flatMap(response => response.status match {
        case StatusCodes.OK => Future.successful(response)
        case _ => Future.failed(new RuntimeException("error message"))
    })
}

val reuests: List[HttpRequest] = List(...)
/*
You don't need for comprehension here because
val eventuallyResponses = for {
  batches <- postBatch(payload)
} yield batches

is equal to
val eventuallyResponses = postBatch(payload)

For comprehension doesn't process recursive sending. If you need it you should write it yourself by flatMap on each request future.
*/
val eventualResponses: Future[List[HttpResponse]] =
    Future.sequence(reuests.map(postBatch)) //also, its better to add some throttling logic here

//as far as i understand you need to wait for all responses and stop the actor system after that
Await.ready(eventualResponses, 10 minutes) //wait for all responses
Await.ready(actorSystem.terminate(), Duration.Inf) //wait for actor system termination

//because of Await.ready(eventualResponses, 10 minutes) we can match on the future value and expect that it should be completed  
eventualResponses.value match {
    case Some(Success(responses)) =>
        log.info("All requests completed")
    case Some(Failure(exception)) =>
        log.error("Some request failed")
        throw exception //rethrow this exception
    case None =>
        log.error("Requests timed out")
        throw RuntimeException("Requests timed out")
}