Hi Rafał, Could you please tell me how to achieve it? The code is listed in the initial message. When onPull procedure is called for the out1 Outlet, onPull for out0 is already executed and inlet is already in pulled state, so hasBeenPulled function returns true for it and it is not possible to pull the inlet.
Regards, Sergey среда, 11 января 2017 г., 18:56:00 UTC+2 пользователь Rafał Krzewski написал: > > The process should look like following: > > 1. Wait for both outlets to pull > 2. Pull from inlet > 2. Wait for the inlet to push an element, make the decision and push it to > the appropriate outlet > 3. Goto 1 > > This way you only ever pull inlet once, and once the element is available > you can always push it out, since both outlets are available. > > Cheers, > Rafał > > W dniu wtorek, 10 stycznia 2017 19:07:48 UTC+1 użytkownik Sergey Sopin > napisał: >> >> So, does it make sense? It seems it would be better to allow inlet be >> pullet once per outlet, not once at all. >> May be there is some existing solution for the problem above? >> >> - Sergey >> >> понедельник, 9 января 2017 г., 17:03:43 UTC+2 пользователь Sergey Sopin >> написал: >>> >>> Hi Endre, >>> >>> I am not sure if I need it. But it seems, that sometimes my inlet is >>> being pulled by outlet_0 when, according to conditions, message should be >>> pushed to outlet_1. In this case my flow get stuck, becase it waits until >>> outlet_1 will pull it and it doesn't happen (because port is already pulled >>> by outlet_0). >>> >>> My flow looks like following: >>> >>> >>> >>> <https://lh3.googleusercontent.com/-91i8qjm-MXM/WHOla3zvGeI/AAAAAAAABQA/t1sEPAi_q6oeMdPWsz4DIlsZdrB8s06QACLcB/s1600/Flow.png> >>> >>> >>> Custom logic of the element (FanOutShape2) can be found in the initial >>> message. Let me please know if you see something wrong there. >>> >>> Thanks! >>> >>> Regards, >>> Sergey >>> >>> понедельник, 9 января 2017 г., 16:10:14 UTC+2 пользователь drewhk >>> написал: >>>> >>>> >>>> >>>> On Mon, Jan 9, 2017 at 3:03 PM, Sergey Sopin <sopi...@gmail.com> wrote: >>>> >>>>> Hi again, >>>>> >>>>> In my case I have 1 inlet and 2 outlets. It seems that inlet port can >>>>> be pulled only once. >>>>> >>>> >>>> Yes, this is clearly documented here: >>>> http://doc.akka.io/docs/akka/2.4/scala/stream/stream-customize.html#Port_states__InHandler_and_OutHandler >>>> >>>> Please look at the state charts of the ports. >>>> >>>> >>>>> So when two outlets try to pull the same inlet one by one, like in >>>>> example above, second attempt will fail, because hasBeenPulled(inlet) >>>>> function will return true. Could you please help me to figure out how to >>>>> deal with it? >>>>> >>>> >>>> Why do you want to pull again if you have already pulled? >>>> >>>> -Endre >>>> >>>> >>>>> Thank you in advance! >>>>> >>>>> Cheers, >>>>> Sergey >>>>> >>>>> >>>>> понедельник, 9 января 2017 г., 14:55:22 UTC+2 пользователь Julian >>>>> Howarth написал: >>>>>> >>>>>> As per >>>>>> http://doc.akka.io/docs/akka/2.4.16/java/stream/stream-customize.html#Port_states__AbstractInHandler_and_AbstractOutHandler >>>>>> >>>>>> you can only push to a port if downstream has pulled (ie signalled that >>>>>> it >>>>>> is ready for data). So, in addition to checking isClosed(outx), you >>>>>> also need to check isAvailable(outx). >>>>>> >>>>>> >>>>>> Alternatively, you can use emit(outx)which takes care of this >>>>>> automatically. >>>>>> >>>>>> Julian >>>>>> >>>>>> >>>>>> On Monday, January 9, 2017 at 12:46:43 PM UTC, Sergey Sopin wrote: >>>>>>> >>>>>>> Hi, >>>>>>> >>>>>>> I have created FanOutShape2 shape with cusom logic: >>>>>>> >>>>>>> @Override >>>>>>> public GraphStageLogic createLogic(Attributes >>>>>>> inheritedAttributes) { >>>>>>> return new GraphStageLogic(shape) { >>>>>>> { >>>>>>> setHandler(in, new AbstractInHandler() { >>>>>>> @Override >>>>>>> public void onPush() throws Exception { >>>>>>> Object result = process(grab(in), >>>>>>> materializer()); >>>>>>> >>>>>>> if (result instanceof ProcessingResponse) { >>>>>>> >>>>>>> ProcessingResponse response = >>>>>>> (ProcessingResponse) result; >>>>>>> if (!isClosed(out1)) { >>>>>>> push(out1, response); >>>>>>> //This is FAndLShape.java:46 >>>>>>> } >>>>>>> } else if (result != null && result >>>>>>> instanceof FinderData) { >>>>>>> FinderData response = (FinderData) >>>>>>> result; >>>>>>> if (!isClosed(out0)) { >>>>>>> push(out0, response); >>>>>>> } >>>>>>> } >>>>>>> } >>>>>>> }); >>>>>>> >>>>>>> setHandler(out0, new AbstractOutHandler() { >>>>>>> @Override >>>>>>> public void onPull() throws Exception { >>>>>>> if ((!hasBeenPulled(in)) && (!isClosed(in))) >>>>>>> { >>>>>>> pull(in); >>>>>>> } >>>>>>> } >>>>>>> }); >>>>>>> >>>>>>> setHandler(out1, new AbstractOutHandler() { >>>>>>> @Override >>>>>>> public void onPull () throws Exception { >>>>>>> if ((!hasBeenPulled(in)) && (!isClosed(in))) >>>>>>> { >>>>>>> pull(in); >>>>>>> } >>>>>>> } >>>>>>> }); >>>>>>> } >>>>>>> >>>>>>> }; >>>>>>> } >>>>>>> >>>>>>> And sometimes I get following error message: >>>>>>> >>>>>>> [error] a.a.RepointableActorRef - Error in stage >>>>>>> [kernel.modeller.workers.streamFinder.finderShapes.FAndLShape@4efd96f7]: >>>>>>> >>>>>>> requirement failed: Cannot push port (out1) twice >>>>>>> java.lang.IllegalArgumentException: requirement failed: Cannot push >>>>>>> port (out1) twice >>>>>>> at scala.Predef$.require(Predef.scala:219) >>>>>>> at akka.stream.stage.GraphStageLogic.push(GraphStage.scala:439) >>>>>>> at >>>>>>> kernel.modeller.workers.streamFinder.finderShapes.FAndLShape$1$1.onPush(FAndLShape.java:46) >>>>>>> at >>>>>>> akka.stream.impl.fusing.GraphInterpreter.processElement$1(GraphInterpreter.scala:582) >>>>>>> at >>>>>>> akka.stream.impl.fusing.GraphInterpreter.processEvent(GraphInterpreter.scala:593) >>>>>>> at >>>>>>> akka.stream.impl.fusing.GraphInterpreter.execute(GraphInterpreter.scala:535) >>>>>>> at >>>>>>> akka.stream.impl.fusing.GraphInterpreterShell.runBatch(ActorGraphInterpreter.scala:443) >>>>>>> at >>>>>>> akka.stream.impl.fusing.GraphInterpreterShell.receive(ActorGraphInterpreter.scala:387) >>>>>>> at >>>>>>> akka.stream.impl.fusing.ActorGraphInterpreter$$anonfun$receive$1.applyOrElse(ActorGraphInterpreter.scala:547) >>>>>>> at akka.actor.Actor$class.aroundReceive(Actor.scala:484) >>>>>>> >>>>>>> Could you tell me what is wrong here? >>>>>>> Thanks! >>>>>>> >>>>>>> Cheers, >>>>>>> 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+...@googlegroups.com. >>>>> To post to this group, send email to akka...@googlegroups.com. >>>>> Visit this group at https://groups.google.com/group/akka-user. >>>>> For more options, visit https://groups.google.com/d/optout. >>>>> >>>> >>>> -- >>>>>>>>>> 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.
