Hi,
I am receiving messages and piping them to mapAsyncUnordered. I know that
for some messages the futures will take much more time than for others (I
can check that with a predicate).
What I would like to do is to split the messages into a “fast lane” and a
“slow lane”, so that the entire stream does not stall (i.e. backpressure)
when mapAsyncUnordered is filled with slow messages only.
My first approach was something like this:
val source = Source.actorRef[Message](400, OverflowStrategy.dropHead)
source.groupBy(isSlow).map {
case (true, slowLane) => slowLane.buffer(300, OverflowStrategy.dropHead).
mapAsyncUnordered(2)(process).to(sink).run()
case (false, fastLane) => slowLane.buffer(100, OverflowStrategy.dropHead).
mapAsyncUnordered(6)(process).to(sink).run()
}.toMat(Sink.ignore)(Keep.left).run()
def process(m: Message) = source ! m
However since my source is a Source.actorRef, I can just do this instead:
val fastLane = Source.actorRef[Message](300, OverflowStrategy.dropHead).
mapAsyncUnordered(2)(process).to(sink).run()
val slowLane = Source.actorRef[Message](100, OverflowStrategy.dropHead).
mapAsyncUnordered(4)(process).to(sink).run()
def process(m: Message) = if (isSlow(m)) slowLane ! m else fastLane ! m
I am not sure I fully understand the pros and cons of each approach, and
maybe there are even better ones. So I would really appreciate any thoughts.
Kind regards,
Nick
--
>>>>>>>>>> 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 [email protected].
To post to this group, send email to [email protected].
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
