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 .