Akka Stream在Flow中使用HttpResponse(Akka Stream use HttpResponse in Flow)

 我想利用一个简单的流程从http服务中收集一些额外的数据，并用结果增强我的数据对象。 以下说明了这个想法： 
 
val httpClient = Http().superPool[User]()

val cityRequest = Flow[User].map { user=>
  (HttpRequest(uri=Uri(config.getString("cityRequestEndpoint"))), User)
}

val cityResponse = Flow[(Try[HttpResponse], User)].map {
  case (Failure(ex), user) => user
  case (Success(resp), user) => {
    // << What to do here to get the value >> //
    val responseData = processResponseSomehowToGetAValue?
    val enhancedUser = new EnhancedUser(user.data, responseData)
    enhancedUser
  }
}

val processEnhancedUser = Flow[EnhancedUser].map {
  // e.g.: Asynchronously save user to a database
}

val useEnhancementGraph = userSource
  .via(getRequest)
  .via(httpClient)
  .via(getResponse)
  .via(processEnhancedUser)
  .to(Sink.foreach(println))
 
 我有一个问题需要理解Flow中流媒体性质和物化/期货之间的机制和差异。 
 
 以下想法没有向我解释： 
 
 
 
 http://doc.akka.io/docs/akka-http/current/scala/http/implications-of-streaming-http-entity.html 
 
 
 akka HttpResponse将body读取为String scala 
 
 
 如何从响应中获取值到新用户对象，因此我可以在以下步骤中处理该对象。 
 
 感谢帮助。 
 
 更新： 
 
 我正在使用远程akka http服务器评估代码，使用下面的代码在解析之前立即和10秒之间回答请求。 这导致一些“EnhancedUser”实例最终出现，但那些花了太长时间才回答的人错过了他们的价值观。 
 
 我在某个时间将.async添加到cityResponse解析器的末尾，结果输出花费的时间更长，但是正确。 
 
 这种行为的原因是什么？它与接受的答案如何契合？ 
 
val cityResponse = Flow[(Try[HttpResponse], User)].map {
  case (Failure(ex), member) => member
  case (Success(response), member) => {
    Unmarshal(response.entity).to[String] onComplete {
      case Success(s) =>  member.city = Some(s)
      case Failure(ex) => member.city = None
    }
  }
  member
}.async  // <<-- This changed the behavior to be correct, why?

I would like to utilize a simple Flow to gather some extra data from a http service and enhance my data object with the results. The following illustrates the Idea:
 
val httpClient = Http().superPool[User]()

val cityRequest = Flow[User].map { user=>
  (HttpRequest(uri=Uri(config.getString("cityRequestEndpoint"))), User)
}

val cityResponse = Flow[(Try[HttpResponse], User)].map {
  case (Failure(ex), user) => user
  case (Success(resp), user) => {
    // << What to do here to get the value >> //
    val responseData = processResponseSomehowToGetAValue?
    val enhancedUser = new EnhancedUser(user.data, responseData)
    enhancedUser
  }
}

val processEnhancedUser = Flow[EnhancedUser].map {
  // e.g.: Asynchronously save user to a database
}

val useEnhancementGraph = userSource
  .via(getRequest)
  .via(httpClient)
  .via(getResponse)
  .via(processEnhancedUser)
  .to(Sink.foreach(println))
 
I have a problem to understand the mechanics and difference between the streaming nature and materialization / Futures inside the Flow.
 
Following ideas did not explain it to me:
 
 
 
http://doc.akka.io/docs/akka-http/current/scala/http/implications-of-streaming-http-entity.html
 
 
akka HttpResponse read body as String scala
 
 
How do i get the value from the response into the new user object, so i can handle that object in the following steps.
 
Thanks for help.
 
Update:
 
I was evaluating the code with a remote akka http server answering to requests between immediately and 10 seconds using the code below for parsing. This led to the effect that some "EnhancedUser" Instances showed up at the end, but the ones who took too long to answer were missing their values.
 
I added .async to the end of the cityResponse parser at some time and the result output took longer, but was correct. 
 
What is the reason for that behaviour and how does it fit together with the accepted Answer?
 
val cityResponse = Flow[(Try[HttpResponse], User)].map {
  case (Failure(ex), member) => member
  case (Success(response), member) => {
    Unmarshal(response.entity).to[String] onComplete {
      case Success(s) =>  member.city = Some(s)
      case Failure(ex) => member.city = None
    }
  }
  member
}.async  // <<-- This changed the behavior to be correct, why?

原文：https://stackoverflow.com/questions/43236367