Akka-Stream + Akka Http - Get Antrag auf Fehler

Question

Ich habe folgenden Strom, der ziemlich gut funktioniert:

source 
    .map(x => HttpRequest(uri = x.rawRequest)) 
    .via(Http().outgoingConnection(host, port)) 
    .to(Sink.actorRef(myActor, IsDone)) 
    .run() 

und einen einfachen Schauspieler den Antwortstatus und die letzte Meldung zu behandeln, wenn der Strom vervollständigt:

/** 
    * A simple actor to count how many rows have been processed 
    * in the complete process given a http status 
    * 
    * It also finish the main thread upon a message of type [[IsDone]] is received 
    */ 
class MyActor extends Actor with ActorLogging { 

    var totalProcessed = 0 

    def receive = LoggingReceive { 

    case response: HttpResponse => 

     if(response.status.isSuccess()) { 
     totalProcessed = totalProcessed + 1 
     } else if(response.status.isFailure()) { 
     log.error(s"Http response error: ${response.status.intValue()} - ${response.status.reason()}") 
     } else { 
     log.error(s"Error: ${response.status.intValue()} - ${response.status.reason()}") 
     } 

    case IsDone => 
     println(s"total processed: $totalProcessed") 
     sys.exit() 
    } 
} 

case object IsDone 

Ich weiß nicht, ob dies der beste Ansatz ist, um Dinge zu zählen und auch um Antwortstatus zu behandeln, aber es funktioniert so weit.

Die Frage ist, wie die ursprüngliche Anfrage an den Aktor in einer Weise übergeben werden, die ich weiß, welche Anfrage einen bestimmten Fehler verursacht hat.

Mein Schauspieler könnte erwarten, dass die folgenden statt:

case (request: String, response: HttpResponse) => 

Aber wie diese Informationen zu übergeben, die ich am Anfang meiner Pipeline?

Ich war von so zu map denken:

source 
    .map(x => (HttpRequest(uri = x.rawRequest), x.rawRequest)) 

Aber ich habe keine Ahnung, wie die Http Fluss zu feuern.

Irgendwelche Vorschläge?

Answer 1

Mit @cmbaxter Hilfe konnte ich mein Problem mit dem folgenden Stück Code lösen:

val poolClientFlow = Http().cachedHostConnectionPool[String](host, port) 

source 
    .map(url => HttpRequest(uri = url) -> url) 
    .via(poolClientFlow) 
    .to(Sink.actorRef(myActor, IsDone)) 
    .run() 

Nun meine Schauspieler ist in der Lage, dies zu erhalten:

case (respTry: Try[HttpResponse], request: String) =>