On Thu, Aug 11, 2016 at 12:32 AM, Mike Bryant <mikesn...@gmail.com> wrote:
> 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?). > The prefixAndTail with zero prefix is a nifty trick, and I actually contemplated to have it exposed as a method called "lift", but in our experience, source-of-sources type of streams very easily end up in deadlocks, leaked resources, etc. Just think about akka-http and unconsumed entity bytes. Hence, we don't really advertise it, but if someone really needs it and there is no alternative, then there is the prefixAndTail-zero trick. -Endre > > 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/c >>> urrent/additional/faq.html >>> >>>>>>>>>> Search the archives: https://groups.google.com/grou >>> p/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. >>> To post to this group, send email to akka...@googlegroups.com. >>> 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. > -- >>>>>>>>>> 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.
