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 <[email protected] > <javascript:>> 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 [email protected] <javascript:>. >> To post to this group, send email to [email protected] >> <javascript:>. >> 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 [email protected]. To post to this group, send email to [email protected]. Visit this group at https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
