You'd preserve the order with a statically partitioned routing table using consistent hashing as well.
On Wed, Jul 22, 2015 at 11:43 AM, Gabriel Volpe <[email protected]> wrote: > So, if this is the case, maybe Akka streams is not the solution to my > problem. It's working well with actors. I just wanted to try it to see > whether it fit better or not. > > Yes, as I explained before, I create actors dynamically depending on the > incoming data. Using this approach I can preserve the order given the > natural implementation of the actor's mailbox (queue). > > El martes, 21 de julio de 2015, 13:46:35 (UTC+1), Joe Edwards escribió: >> >> AFAIK there is currently no way of dynamically changing the size of the >> pool. What you can do is assign a fixed number of 'slots' which get >> dynamically turned off/on >> >> You could look at >> https://github.com/akka/akka/blob/releasing-akka-stream-and-http-experimental-1.0/akka-http-core/src/main/scala/akka/http/impl/engine/client/PoolFlow.scala >> to see how akka do it with their HTTP Connection Pool. >> >> Do you actually need it to be dynamically created? It would certainly be >> more straightforward to have a fixed size pool that is materialized to >> start with. >> >> On Tuesday, 21 July 2015 12:57:01 UTC+1, Gabriel Volpe wrote: >>> >>> 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? >>> >>> >>> Yes, that's exactly what I wanna do because I need to preserve the order >>> of the events with the same ID. >>> >>> >>>> 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. >>>> >>> >>> I will take a look to FlexiRoute. Can I create downstreams dinamically >>> with this solution? That's what I'm doing with actors. If it is the first >>> time I received the ID 1234, then I create an actor for this event. >>> Otherwise I just send the event to the correspondent actor. >>> >>> Thanks for your time! >>> Gabriel. >>> >>> >>>> >>>> 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. > -- Cheers, √ -- >>>>>>>>>> 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.
