Scala Akka RestartSource onFailuresWithBackoff not working complaining dead letters encountered

Viewed 12

I have created a Flow like below

private val httpFlow: Flow[HttpRequest, HttpResponse, Any] =
    Http().outgoingConnection(host, port)

And then processing flow like

val request = Post(Uri(path), some_json_payload)(
      stringMarshaller(MediaTypes.`application/json`),
      system.executionContext
    )
httpClient.processRequest(request, httpFlow) map (...)

My http client looks like below, added a RestartSettings and restarting source on onFailuresWithBackoff

class HttpClient(implicit val system: ActorSystem[_]) {
  implicit val materializer: Materializer = Materializer(system)

  private def checkSuccess(response: HttpResponse, path: String): Either[Throwable, HttpResponse] =
    Either.cond(
      response.status == StatusCodes.OK,
      response,
      new Throwable(s"Could not GET $path, Service returned ${response.status.toString()}"),
    )

  val settings: RestartSettings = RestartSettings(
    minBackoff = 3.seconds,
    maxBackoff = 30.seconds,
    randomFactor = 0.2
  ).withMaxRestarts(20, 5.minutes)


  private def httpRequest(
     request: HttpRequest,
     flow: Flow[HttpRequest, HttpResponse, Any],
     ): Future[Either[Throwable, HttpResponse]] =
        RestartSource.onFailuresWithBackoff(settings) { () =>
          Source.single(request)
        }.via(flow)
         .runWith(Sink.head)
         .map(checkSuccess(_, request.uri.path.toString()))

  private def unmarshallString(response: Either[Throwable, HttpResponse]): Future[Either[Throwable, String]] =
    response.fold(
      err => Future(err.asLeft[String]),
      r => Unmarshal(r.entity).to[String].map(_.asRight[Throwable]),
    )

  def processRequest(request: HttpRequest, flow: Flow[HttpRequest, HttpResponse, Any]): Future[Enclosed[Json]] =
    httpRequest(request, flow) flatMap unmarshallString map (_.flatMap(parse))
}

I am pretty new especially in akka , what I am missing here .

0 Answers
Related