CompositeModuleInterpreterActor extends AbstractModuleInterpreterActor { @Override void inPortHasSignal(int inPortId) { if (state != State.REPLAY && inPortReceivedMessage.get(inPortId)) { log.warning(String.format( "Ignoring redundant message that %s received value.", module.getInPorts().get(inPortId) )); } else if (!recomputedInPorts.get(inPortId)) { log.warning(String.format( "Ignoring unexpected message that %s received a value. " + "In-ports that are expected to receive a value: %s", module.getInPorts().get(inPortId), recomputedInPorts.stream() .mapToObj(module.getInPorts()::get) .map(RuntimeInPort::getSimpleName) .collect(Collectors.toList()) )); } else { inPortReceivedMessage.set(inPortId); if (state != State.STARTING) { RuntimeInPort inPort = module.getInPorts().get(inPortId); transmitValueFromInPort(inPort); endAsynchronousAction(inPort); } } } private  CompositeModuleInterpreterActor(Factory factory); @Override void preStart(); @Override void postStop(); @Override void onReceive(Object message);  }
@Test public void inPortHasSignal() { try (ModuleInterpretation moduleInterpretation = testCaseBuilder("inPortHasSignal", PROXY_CONTEXT).build()) { MapStagingArea stagingArea = moduleInterpretation.getStagingArea(); Map<ExecutionTrace, ObjectNode> stagingAreaMap = stagingArea.toUnmodifiableMap(); put(stagingArea, ":in:x", ":in:y"); int moduleId = 3; Props props = moduleInterpretation.toCompositeModuleInterpreterProps(moduleId, bitSetOf(0), bitSetOf(0, 1)); ActorSystem actorSystem = moduleInterpretation.getActorSystem(); TestActorRef<CompositeModuleInterpreterActor> actorRef = TestActorRef.create(actorSystem, props, moduleInterpretation.getSupervisor()); TestProbe akkaEventProbe = new TestProbe(actorSystem); actorSystem.eventStream().subscribe(akkaEventProbe.ref(), Object.class); actorRef.receive(new InPortHasSignal(0)); actorRef.receive(new InPortHasSignal(0)); LogEvent redundantMsgLogEvent = akkaEventProbe.expectMsgClass(LogEvent.class); Assert.assertEquals(redundantMsgLogEvent.logSource(), actorRef.path().toString()); String redundantMsg = (String) redundantMsgLogEvent.message(); Assert.assertTrue(redundantMsg.contains("redundant")); Assert.assertTrue(redundantMsg.contains(PROXY_CONTEXT.getModule().getInPorts().get(0).toString())); CallingThreadExecutor asyncTaskExecutor = moduleInterpretation.getAsyncTaskExecutor(); asyncTaskExecutor.executeAll(); TestProbe parentProbe = moduleInterpretation.getParentProbe(); Assert.assertEquals( parentProbe.expectMsgClass(Object.class), new SubmoduleOutPortHasSignal(moduleId, 1) ); Assert.assertFalse(parentProbe.msgAvailable()); Assert.assertFalse(akkaEventProbe.msgAvailable()); Assert.assertEquals( stagingAreaMap.keySet(), tracesSet(":in:x", ":in:y", "/a:in:ax", "/a:in:ay", "/b:in:bx", ":out:q") ); Assert.assertEquals( IntStream.range(0, 2) .mapToObj( id -> moduleInterpretation.getTestProbeForChild("a").expectMsgClass(InPortHasSignal.class) ) .mapToInt(InPortHasSignal::getInPortId) .collect(BitSet::new, BitSet::set, BitSet::or), bitSetOf(0, 1) ); Assert.assertFalse(moduleInterpretation.getTestProbeForChild("a").msgAvailable()); moduleInterpretation.getTestProbeForChild("b").expectMsg(new InPortHasSignal(0)); Assert.assertFalse(moduleInterpretation.getTestProbeForChild("b").msgAvailable()); Assert.assertFalse(moduleInterpretation.getTestProbeForChild("c").msgAvailable()); actorRef.receive(new InPortHasSignal(1)); LogEvent unexpectedMsgLogEvent = akkaEventProbe.expectMsgClass(LogEvent.class); Assert.assertEquals(unexpectedMsgLogEvent.logSource(), actorRef.path().toString()); String unexpectedMsg = (String) unexpectedMsgLogEvent.message(); Assert.assertTrue(unexpectedMsg.contains("unexpected")); Assert.assertTrue(unexpectedMsg.contains(PROXY_CONTEXT.getModule().getInPorts().get(1).toString())); } }