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