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] > <javascript:>> 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] <javascript:>. >> To post to this group, send email to [email protected] >> <javascript:>. >> 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.
