CompositeOperator extends Operator<Self> { @Override public ElementarySopremoModule asElementaryOperators() { final SopremoModule module = new SopremoModule(this.getNumInputs(), this.getNumOutputs()); module.setName(this.getName()); this.addImplementation(module); if (this.getDegreeOfParallelism() != Operator.STANDARD_DEGREE_OF_PARALLELISM) for (final Operator<?> moduleOperator : module.getReachableNodes()) if (moduleOperator.getDegreeOfParallelism() == Operator.STANDARD_DEGREE_OF_PARALLELISM) moduleOperator.setDegreeOfParallelism(this.getDegreeOfParallelism()); module.validate(); return module.asElementary(); } CompositeOperator();  CompositeOperator(final int numberOfInputs, final int numberOfOutputs);  CompositeOperator(final int minInputs, final int maxInputs, final int minOutputs, final int maxOutputs); abstract void addImplementation(SopremoModule module); @Override ElementarySopremoModule asElementaryOperators();  }
@Test public void testAsElementaryOperators() throws Exception { final Operator<?> input1 = new Source("file: final Operator<?> input2 = new Source("file: final Operator<?> input3 = new Source("file: final CompositeOperator<?> fixture = new CompositeOperatorImpl(1); fixture.setInputs(input1, input2, input3); final ElementarySopremoModule module = fixture.asElementaryOperators(); assertNotNull(module); final List<Level<Operator<?>>> reachableNodes = GraphLevelPartitioner.getLevels( module.getAllOutputs(), OperatorNavigator.INSTANCE); assertEquals(3, reachableNodes.get(0).getLevelNodes().size()); assertEquals(1, reachableNodes.get(1).getLevelNodes().size()); assertEquals(1, reachableNodes.get(2).getLevelNodes().size()); assertEquals(1, reachableNodes.get(3).getLevelNodes().size()); for (int index = 0; index < 3; index++) assertSame(Source.class, reachableNodes.get(0).getLevelNodes().get(index).getClass()); assertSame(ElementaryOperatorImpl.class, reachableNodes.get(1) .getLevelNodes().get(0).getClass()); assertSame(ElementaryOperatorImpl.class, reachableNodes.get(2) .getLevelNodes().get(0).getClass()); assertSame(Sink.class, reachableNodes.get(3).getLevelNodes().get(0).getClass()); }