CompositeModuleInterpreterActor extends AbstractModuleInterpreterActor { private void submoduleOutPortHasSignal(int moduleId, int outPortId) { assert state.compareTo(State.RUNNING) >= 0; final RuntimeOutPort outPort = module.getModules().get(moduleId).getOutPorts().get(outPortId); if (submoduleOutPortsReceivedMessage[moduleId].get(outPortId)) { log.warning(String.format( "Ignoring redundant message that %s of submodule '%s' provided a value.", outPort, outPort.getModule().getSimpleName() )); } else if (isOutPortNeeded(outPort)) { submoduleOutPortsReceivedMessage[moduleId].set(outPortId); List<CompletableFuture<Void>> copyFutures = outPort.getOutConnections().stream() .filter(this::isConnectionToRecomputedNode) .map(connection -> connection.accept(transmitFromSourcePortVisitor, outPort)) .collect(Collectors.toList()); CompletableFuture<Object> messageFuture = Futures.collect(copyFutures) .thenApply(ignored -> new SubmoduleOutPortNoLongerNeeded(outPort)); pipeResultToSelf(messageFuture, "copying value of %s of submodule to out-connections", outPort, outPort.getSimpleName()); } else { log.warning(String.format( "Ignoring unexpected message that %s of submodule '%s' provided a value because it is not needed.", outPort, outPort.getSimpleName() )); } } private  CompositeModuleInterpreterActor(Factory factory); @Override void preStart(); @Override void postStop(); @Override void onReceive(Object message);  }
@Test public void submoduleOutPortHasSignal() { try ( ModuleInterpretation moduleInterpretation = testCaseBuilder("submoduleOutPortHasSignal", PROXY_CONTEXT) .build() ) { MapStagingArea stagingArea = moduleInterpretation.getStagingArea(); Map<ExecutionTrace, ObjectNode> stagingAreaMap = stagingArea.toUnmodifiableMap(); put(stagingArea, ":in:x", ":in:y"); Props props = moduleInterpretation.toCompositeModuleInterpreterProps(3, bitSetOf(), bitSetOf(0, 1)); ActorSystem actorSystem = moduleInterpretation.getActorSystem(); TestActorRef<CompositeModuleInterpreterActor> actorRef = TestActorRef.create( moduleInterpretation.getActorSystem(), props, moduleInterpretation.getSupervisor()); TestProbe akkaEventProbe = new TestProbe(actorSystem); actorSystem.eventStream().subscribe(akkaEventProbe.ref(), Object.class); CallingThreadExecutor asyncTaskExecutor = moduleInterpretation.getAsyncTaskExecutor(); asyncTaskExecutor.executeAll(); Assert.assertEquals( stagingAreaMap.keySet(), tracesSet(":in:x", ":in:y", "/a:in:ax", "/a:in:ay", "/b:in:bx", ":out:q") ); TestProbe aProbe = moduleInterpretation.getTestProbeForChild("a"); aProbe.expectMsgClass(InPortHasSignal.class); aProbe.expectMsgClass(InPortHasSignal.class); Assert.assertFalse(aProbe.msgAvailable()); int aId = submodule(moduleInterpretation, "a").getIndex(); put(stagingArea, "/a:out:ap"); actorRef.receive(new SubmoduleOutPortHasSignal(aId, 0)); Assert.assertFalse(akkaEventProbe.msgAvailable()); actorRef.receive(new SubmoduleOutPortHasSignal(aId, 0)); LogEvent redundantMsgLogEvent = akkaEventProbe.expectMsgClass(LogEvent.class); Assert.assertEquals(redundantMsgLogEvent.logSource(), actorRef.path().toString()); String redundantMsg = (String) redundantMsgLogEvent.message(); Assert.assertTrue(redundantMsg.contains("redundant")); RuntimeParentModule parentModule = (RuntimeParentModule) moduleInterpretation.getResolvedModule(); RuntimeModule submodule = Objects.requireNonNull( parentModule.getEnclosedElement(RuntimeModule.class, SimpleName.identifier("a"))); Assert.assertTrue(redundantMsg.contains(submodule.getOutPorts().get(0).toString())); Assert.assertFalse(akkaEventProbe.msgAvailable()); actorRef.receive(new SubmoduleOutPortHasSignal(aId, 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(submodule.getOutPorts().get(1).toString())); asyncTaskExecutor.executeAll(); Assert.assertEquals( stagingAreaMap.keySet(), tracesSet(":in:x", ":in:y", "/b:in:bx", "/b:in:by", ":out:q") ); } }