I thought I needed to create dynamic flows at one point too. Then I realized if I just "tag" my incoming data with what I need and make a smart enough detached stage, I can control the pressure and order as necessary. Also, at one point I did need to fork my data (into a predefined number of slots) and join it later and I just used broadcast/merge instead of "flexi" pieces.
On Tuesday, July 21, 2015 at 7:56:07 AM UTC-5, √ wrote: > > In my opinion, there is little value in creating the slots dynamically as > you most likely will either: > > A) have more slots than CPUs, which means you won't get any performance > improvement for doing it dynamically. > or B) less slots than CPUs, in which case you can statically create them > and use consistent hashing on a key/keys > > > > On Tue, Jul 21, 2015 at 2:46 PM, Joe Edwards <[email protected] > <javascript:>> wrote: > >> 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] <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, > √ > -- >>>>>>>>>> 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.
