This channel is available for all Akka enthusiasts—newbies as well as gurus—for the exchange of knowledge and the coordination of efforts around Akka. For more info see: http://bit.ly/akka-gitter
9 replies
Hi,
When I am trying to consume data on server into memory later created strict entity with it, but getting following error message. Can you please help me?
Substream Source cannot be materialized more than onceHere is the code,
if (!entity.isChunked()) {
entity.dataBytes.runFold(ByteString.empty)(_ ++ _).flatMap { encryptedBody =>
decrypt(encryptedBody).map { decrypted =>
HttpEntity(decrypted)
}
}
}AkkaHttp Version: “10.0.13”
7 replies
@bmaso Thanks. Here it goes:
I've a stream of job requests coming in and need to do some processing on them. The thing is that I would like to refresh some enrichment related data periodically that is needed for the processing of these jobs.
I've come up with two solutions.
- A cached stream that spits out the enrichment loaded from db repeatedly and via timeout on the stream I recursively reload the source. This works ok but it's a bit strange since the enrichment is rather part of a processing stage then the messages/jobs themselves so zipping the two looks just strange. The idea here is that the repeated source will always emit so it's safe to zip it with jobs stream to form a tuple for processing a job with some enrichment.
- I could create an actor that does the caching and receives a periodic message via a scheduled task to reload the enrichment. Then when processing the jobs, each job would be handled via a
flatMapConcatwhere I can ask the actor for the latest enrichment and create the graphStage according to that information for every message.
I think the second approach is better. I'm wondering whether there is any other way or recommendation to do such a thing. (I of course could reload the data via some crappy counter that reloads on every n hit but the timer would be better and I'm also looking for a rather 'pure' streaming/akka solution)
Http().singleRequest(...) does retries by default, but I do not know for which HTTP status and I would like to apply explicit rules, which kind of responses are actually retried how often and with delays based on the Retry-After Header if available. To me this sounds so basic, that it should not be that complicated to accomplish. Does anybody know of good code examples?
@hbergmey Yes, custom code probably would be the way to go. Akka's built-in retry i think can only be configured using max-retries, and reading https://github.com/akka/akka-http/blob/324f985b9d6a66518b789eaf0cccb9f6bf400079/docs/src/main/paradox/client-side/host-level.md#retrying-a-request, i think it's only meant to retry for which no response was received at all. So, retry on 503 you'd have to code yourself.
However, I don't think you'd need a Flow for that... just a simple recursive method tryRequest(rq: HttpRequest, retriesLeft: Int): Future[HttpResponse] could do it, which would call itself with retriesLeft - 1 with a delay on error.
It's still going to hit the http pool underneath, where queueing is implemented (according to max-open-requests / max-connections), so that'll give you some safety w.r.t. concurrency.
A side remark: if your downstream services are indeed yielding 503 because they're overloaded, you may want to consider a proper circuit breaker instead (akka has one). That'll make your downstream services a lot happier than a blind retry.