title: Akka Stream use HttpResponse in Flow
body: <p>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:</p>
  
  <pre><code>val httpClient = Http().superPool[User]()
  
  val cityRequest = Flow[User].map { user=&gt;
    (HttpRequest(uri=Uri(config.getString("cityRequestEndpoint"))), User)
  }
  
  val cityResponse = Flow[(Try[HttpResponse], User)].map {
    case (Failure(ex), user) =&gt; user
    case (Success(resp), user) =&gt; {
      // &lt;&lt; What to do here to get the value &gt;&gt; //
      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))
  </code></pre>
  
  <p>I have a problem to understand the mechanics and difference between
  the streaming nature and materialization / Futures inside the Flow.</p>
  
  <p>Following ideas did not explain it to me:</p>
  
  <ul>
  <li><a href="http://doc.akka.io/docs/akka-http/current/scala/http/implications-of-streaming-http-entity.html" rel="nofollow noreferrer">http://doc.akka.io/docs/akka-http/current/scala/http/implications-of-streaming-http-entity.html</a></li>
  <li><a href="https://stackoverflow.com/questions/32315789/akka-httpresponse-read-body-as-string-scala">akka HttpResponse read body as String scala</a></li>
  </ul>
  
  <p>How do i get the value from the response into the new user object,
  so i can handle that object in the following steps.</p>
  
  <p>Thanks for help.</p>
  
  <p><em>Update:</em></p>
  
  <p>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.</p>
  
  <p>I added .async to the end of the cityResponse parser at some time and the result output took longer, but was correct. </p>
  
  <p>What is the reason for that behaviour and how does it fit together with the accepted Answer?</p>
  
  <pre><code>val cityResponse = Flow[(Try[HttpResponse], User)].map {
    case (Failure(ex), member) =&gt; member
    case (Success(response), member) =&gt; {
      Unmarshal(response.entity).to[String] onComplete {
        case Success(s) =&gt;  member.city = Some(s)
        case Failure(ex) =&gt; member.city = None
      }
    }
    member
  }.async  // &lt;&lt;-- This changed the behavior to be correct, why?
  </code></pre>