Sergey, I spent a few minutes looking at your code, and I see two problems: -inletNumberToPullFrom is not scoped properly. It should be local to GraphLogic implementation. In the current form, it is shared between all ManagementFlow instances that get materialized that is definitely not what you want - you expect that elements will be pushed to the inlets before findInelt() method will be first called. To the best of my knowledge this is not how Akka Streams work. If you don't see the exception in the default case of switch(inletNumbetToPullFrom) thrown, that would mean that the demand token never reaches ManagementFlow inlet 0, possibly because it is not propagated correctly in FL graph.
My recommendation is still to test FL, S, C and ManagementFlow in isolation (ie connecting them only to TestSource, TestSink, Source.single, Sink.ignore etc) and only then proceed to wiring them together. BTW, have you read the blog posts [1] and [2]? I find them quite informative. Cheers, Rafał [1] http://blog.akka.io/streams/2016/07/30/mastering-graph-stage-part-1 [2] http://blog.akka.io/integrations/2016/08/25/simple-sink-source-with-graphstage W dniu poniedziałek, 24 października 2016 23:38:54 UTC+2 użytkownik Sergey Sopin napisał: > > Hi Rafał, > > - sink requests data from you >> - OutletHandler.onPull is invoked on the outlet where the sink is >> connected to >> - you propagate demand outstream by calling pull on any (or all) of your >> Inlets, depending on your logic >> - eventually data becomes available upstream >> - InletHandler.onPush is invoked on the inlet you pulled previously, with >> the incoming element >> > > I tried to add logging into onPull functions, but it didn't help. I see > only messages from the beggining of flow ("Message_1"), but not from my > custom shape. > > OK, but the actual number of workers should not be greater than the number >> of available CPUs, because otherwise Akka will interleave their execution >> anyway. Spawning 1000s of worker flows will only waste memory. Of course I >> understand that the input of fixed collection of data is artif >> > > I tried to remove balancer at all and work with the single worker. It > doesn't help. > > I usually prefer to debug single problem at a time than a number of >> possibly interrelated problems at once. >> > > Me too and I will try, but I am sure that it will give me nothing in this > case. > The problem is in the shape itself. > > Regards, > Sergey > -- >>>>>>>>>> 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 akka-user+unsubscr...@googlegroups.com. To post to this group, send email to akka-user@googlegroups.com. Visit this group at https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
