Hi Luciano,
It looks like you are making too many concurrent requests. By adding the
following failure handler to the second worker's receive function:
case akka.actor.Status.Failure(failure) => log.error(failure, "Failed")
You should then see the this error message and accompanying stack trace:
10:44:43.858 [Akka-Http-Client-akka.actor.default-dispatcher-6] ERROR
SecondWorker - Failed
akka.stream.BufferOverflowException: Exceeded configured max-open-requests
value of [32]
Regards,
Mark
On Friday, 17 June 2016 22:20:19 UTC+1, Luciano Molinari wrote:
>
> Hi all,
>
> I'm trying to create a scenario where I have two sets of workers (behind a
> Router) where the first set of workers is responsible for sending an HTTP
> request (using Akka HTTP Client) and the second is responsible for handling
> the response. Here's the code:
>
> import akka.actor.{Actor, ActorLogging, ActorRef, ActorSystem, Props}
> import akka.http.scaladsl.Http
> import akka.http.scaladsl.model.{HttpRequest, HttpResponse, StatusCodes}
> import akka.pattern.pipe
> import akka.routing.RoundRobinPool
> import akka.stream.javadsl.Sink
> import akka.stream.{ActorMaterializer, ActorMaterializerSettings}
>
> import scala.concurrent.ExecutionContext.Implicits.global
> import scala.io.StdIn
>
> case class Request(id: Long)
>
> object RouterWithWorkersHttpClient {
>
> def main(args: Array[String]) = {
>
> implicit val system = ActorSystem("Akka-Http-Client")
> implicit val materializer = ActorMaterializer()
>
> val secondWorkers =
> system.actorOf(Props[SecondWorker].withRouter(RoundRobinPool(5)), name =
> "SecondWorkers")
> val firstWorkers =
> system.actorOf(FirstWorker.props(secondWorkers).withRouter(RoundRobinPool(5)),
> name = "FirstWorkers")
>
> (1 to 50).foreach(i => firstWorkers ! Request(i))
>
> StdIn.readLine
> Http().shutdownAllConnectionPools() andThen { case _ =>
> system.terminate() }
> println("System completed")
> }
>
> }
>
> object FirstWorker {
> def props(secondWorker: ActorRef): Props = Props(new
> FirstWorker(secondWorker))
> }
>
> class FirstWorker(secondWorker: ActorRef) extends Actor with ActorLogging {
> implicit val materializer: ActorMaterializer =
> ActorMaterializer(ActorMaterializerSettings(context.system))
>
> def receive = {
> case Request(id) => {
> log.info(s"Processing request $id at ${context.self}")
> Http(context.system).singleRequest(HttpRequest(uri =
> "http://localhost:8080/customer/2";)) pipeTo secondWorker
> log.info(s"Http sent for ${id}")
> }
> }
>
> }
>
> class SecondWorker extends Actor with ActorLogging {
> implicit val materializer: ActorMaterializer =
> ActorMaterializer(ActorMaterializerSettings(context.system))
>
> def receive = {
> case HttpResponse(StatusCodes.OK, headers, entity, _) =>
> log.info(s"Response received")
> entity.dataBytes.runWith(Sink.ignore)
>
> case HttpResponse(code, _, _, _) =>
> log.info(s"Not good..got reply $code")
> }
>
> }
>
>
>
> The issue I'm facing is that while I can see 50 messages on the log for
> "Processing request" and "Http sent for", I can only see 32 for "Response
> received". It looks like I'm not getting all the responses.
>
> Am I missing something? Is this a good design for such scenario?
>
> Thanks.
>
> Regards,
> Luciano
>
>
>
>
>
>
--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ:
>>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
