[Akka-streams] Cannot push port twice

[Sergey Sopin]

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

[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

[Sergey Sopin]

Hi Julian,

Thank you very much! 

понедельник, 9 января 2017 г., 14:55:22 UTC+2 пользователь Julian Howarth написал:

[Sergey Sopin]

Hi again,

In my case I have 1 inlet and 2 outlets. It seems that inlet port can be pulled only once. 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?

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).

[Endre Varga]

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

 

--
>>>>>>>>>> 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+unsubscribe@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.

[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: 

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 написал:

To unsubscribe from this group and stop receiving emails from it, send an email to akka-user+...@googlegroups.com.

[Sergey Sopin]

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 написал:

[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ł

[Sergey Sopin]

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 написал:

[Sergey Sopin]

Hi Rafał,

Yeah, you are right! Thank you! I understood.

Cheers,

Sergey

среда, 11 января 2017 г., 18:56:00 UTC+2 пользователь Rafał Krzewski написал:

The process should look like following:

[Sergey Sopin]

Hi again,

Rafał, could you please give me an example of how to implement waiting?

Thanks!

- Sergey

среда, 11 января 2017 г., 18:56:00 UTC+2 пользователь Rafał Krzewski написал:

The process should look like following:

[Rafał Krzewski]

Of course, you should wait explicitly. I only meant the order of events happening.

I coded up a minimal implementation https://gist.github.com/rkrzewski/f1d131405ddb8ce0d1a6fded55da8c23 (it's in Scala, I hope you don't mind)

A more practical one would require handling back-pressure on each outlet separately, so it would look more like the implementation of Partition stage: https://github.com/akka/akka/blob/master/akka-stream/src/main/scala/akka/stream/scaladsl/Graph.scala#L501 

Hope that helps,

Rafał

[Rafał Krzewski]

Hi,

here's more robust implementation of stream splitter, based on Partition stage: https://gist.github.com/rkrzewski/a0fc5d0b47d9a3e0b2c81435adef3fe7

cheers,

Rafał

[Sergey Sopin]

Hi Rafał,

Thank you very much!

Cheers,

Sergey

четверг, 12 января 2017 г., 12:51:23 UTC+2 пользователь Rafał Krzewski написал: