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.
