Hi!
On Wed, Jul 16, 2014 at 10:49 AM, schoefts <thomas.schoeft...@gmail.com>
wrote:
> Hi everyone,
>
> we are currently integrating Akka-Streams in an application.
>
> Looking at the current Flow DSL/API, I can see that Transformers
> (Flow#transform) allow to *synchronously* produce a number of
> output-messages from 1 message.
> abstract class Transformer[-T, +U] {
> * /***
> * * Invoked for each element to produce a (possibly empty) sequence of*
> * * output elements.*
> * */*
> def onNext(element: T): *immutable.Seq[U]*
> Semantically, the Flow remains a Flow[U] - not a Flow[Seq[U]] - despite
> multiplying the incoming message. (Great!)
>
> Is there also a way to achieve the same thing *asynchronously *in a Flow
> (i.e. transform a single message into multiples)?
> From what I've seen, the FlowDSL supports mapFuture which allows async
> computation of a single output message:
>
Yes, this is correct, Transformer encapsulates synchronous computations.
There is a ticket to change that (not necessarily exactly what you want):
https://github.com/akka/akka/issues/15099
> * /***
> * * Transform this stream by applying the given function to each of the
> elements*
> * * as they pass through this processing step. The function returns a
> `Future` of the*
> * * element that will be emitted downstream. As many futures as
> requested elements by*
> * * downstream may run in parallel and may complete in any order, but
> the elements that*
> * * are emitted downstream are in the same order as from upstream.*
> * */*
> def mapFuture[U](f: T ⇒ Future[U]): Flow[U]
>
> Again, the Flow remains a Flow[U] - despite the async computation of the
> output element.
> Naturally mapFuture follows the semantics of the map operation and
> converts 1 input msg into 1 output message.
>
> Using MapFuture to produce a Seq[U] would alter the Flow from Flow[U] to
> Flow[Seq[U]].
> Following processors in that Flow would have to deal with Sequences
> instead of single messages.
>
You have several options currently to achieve what you want. One is to use
mapFuture to get a Flow[Seq[U]] and then use a mapSeq (or a custom
Transformer) to unroll the Seq and get a Flow[U]. The other one is to use a
map element that returns the result of the asynchronous computation as a
Producer (myAsyncComputation: T => Producer[U]), so you get a
Flow[Producer[U]], and then use the flatten(FlattenStrategy.Concat)
operation to get the unrolled sequence.
> This somewhat defies the purpose of backpressure.
>
> Maybe I am missing another function of the Flow DSL, but I haven't found
> it...
>
Streams are a preview right now and the combinators are limited. We are
actively researching the minimal usable set of combinators without blowing
up and having a method for every single case. We plan to have a cookbook
doc page, too, to show how certain elements can be combined.
-Endre
>
> Thanks for any help,
> -Thomas
>
> --
> >>>>>>>>>> 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 http://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 http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
