Thanks Endre - that's perfect. The prefixAndTail with zero prefix was the missing ingredient I needed to transform the whole source within a separate flow (maybe there's a case for adding that as another specific "mapSource"-type method?).
Best, ~Mike On Wednesday, 10 August 2016 22:20:58 UTC+1, drewhk wrote: > > > Using Akka HTTP and streams I can take a Source[ByteString, _] and using >> it as the body entity for a POST request to some HTTP endpoint, yielding >> some response. I'd like to abstract this into an Akka streams *Sink* I >> can run with said Source and materialise to, for example, the HTTP status >> code (or the response body.) The goal is reusability - to be able to write >> to, say, stdout (via StreamConverters.fromOutputStream), or HTTP, or both >> (the same way I can pipe to Curl.) >> >> Having become a bit familiar with the streams API I'm a quite stumped as >> to how to do this, if it it's even possible at present. If it is possible, >> would anyone be able to offer some advice or a general outline on how to >> achieve it? >> > > Just a quick attempt (didn't try any of these): > > bytesSrc // Source[ByteString, NotUsed] > .prefixAndTail(0).map(_._2) // Source[Source[ByteString, NotUsed], > NotUsed] > .map(bytes => HttpRequest(..., body = bytes)) // Source[HttpRequest, > NotUsed] > .via(HttpConnection) // Source[HttpResponse, NotUsed] > .to(Sink.head) // RunnableGraph[Future[HttpResponse]] > > You can make it a Sink if you don't start from a Source but a Flow instead: > > val postSink = Flow[ByteString] > .prefixAndTail(0).map(_._2) > .map(bytes => HttpRequest(..., body = bytes)) > .via(HttpConnection) > .to(Sink.head) // Sink[Future[HttpResponse]] > > Now you can do: > > val responseFuture = myBytes.runWith(postSink) > val another = responseFuture.entityBytes.runWith(postSink) > // Don't forget to consume the response body of the HttpResponse!!!! If > you don't need it, then probably better consume it in the Sink itself and > only expose status code etc. > > Done. > > -Endre > > >> -- >> >>>>>>>>>> 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 akka-user+...@googlegroups.com <javascript:>. >> To post to this group, send email to akka...@googlegroups.com >> <javascript:>. >> Visit this group at https://groups.google.com/group/akka-user. >> For more options, visit https://groups.google.com/d/optout. >> > > -- >>>>>>>>>> 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 akka-user+unsubscr...@googlegroups.com. To post to this group, send email to akka-user@googlegroups.com. Visit this group at https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
