Sounds to me like you want a pooled set of streams for processing your events, with the restriction that events with the same ID must go into the same stream?
If that's the case, a FlexiRoute with a fixed set of downstreams should do the job. Then all you need is a way of consistently selecting a downstream based on event ID (e.g. hashcode modulo n) - this will be more extensible than storing a map with every event ID you encounter. On Monday, 20 July 2015 16:03:10 UTC+1, Gabriel Volpe wrote: > > I don't understand how the FlexiRoute could be the solution. Could you > expand the case further? > > Actually, I'm solving the problem using only Akka Actors and it's quite > easy. I have a Controller Actor which has a Map[Long, ActorRef] > corresponding to the Event ID and the Processor Actor (child). > > Whenever the Controller receives an event, it checks if the ID already > exists in the map and it sends a message to the Processor. If the ID does > not exists, it creates another processor actor only for this Event ID. The > only problem with this approach is that I have to control the back-pressure > manually, that's why I'm trying Akka Streams. > > Thanks, > Gabriel. > > El lunes, 20 de julio de 2015, 14:51:35 (UTC+1), Konrad Malawski escribió: >> >> I think you might want to use a FlexiRoute >> http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0/scala/stream-customize.html#Using_FlexiRoute >> And in it you'd decide "which way the element should go". >> You'd have 2 outputs from it, one being the async path, and the other one >> the other path. >> After done with the event (via one of these Flows), you'd add a Merge at >> the end of them and thus be able to still consume it using one sink. >> >> Maybe I missed something in your requirements? >> >> On Fri, Jul 17, 2015 at 4:57 PM, Gabriel Volpe <[email protected]> >> wrote: >> >>> Hi all, this is my case: >>> >>> I have a queue (Rabbit MQ) producing Json data with the following >>> format: >>> >>> { "id": 123, "message": "hello" } >>> { "id": 876, "message": "shutdown" } >>> { "id": 123, "message": "bye" } >>> >>> And I have a respective Consumer. What I want to do is to distribute the >>> load depending on the *id* field. I mean, if the id already appeared >>> before I want to respect the order, if not I want to process the event in >>> parallel. This would be the Flow: >>> >>> case class Event(id: Long, message: String) >>> >>> val queue = Source(rabbitQueueConnection).mapAsync(100)(d => Future( >>> fromJson(d.message.body.utf8String))) >>> val processor = Flow[Try[Option[Event]]] filter (_.isSuccess) map (_.get >>> ) filter (_.isDefined) map (_.get) >>> val out = Sink.ignore >>> >>> queue ~> processor ~> out >>> >>> I've been playing around >>> <https://github.com/gvolpe/events-processor-prototype/blob/master/src/main/scala/com/gvolpe/prototypes/processor/flows/ConsumerProcessorFlow.scala> >>> >>> but I don't know how to do this. >>> >>> Any ideas? >>> >>> Thanks, >>> Gabriel. >>> >>> -- >>> >>>>>>>>>> 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. >>> >> >> >> >> -- >> Cheers, >> Konrad 'ktoso' Malawski >> Akka <http://akka.io/> @ Typesafe <http://typesafe.com/> >> > -- >>>>>>>>>> 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.
