OperatorImplGraph { public Collection<InputOperatorImpl> getAllInputOperators() { return Collections.unmodifiableCollection(this.inputOperators.values()); } OperatorImplGraph(OperatorSpecGraph specGraph, Context context, Clock clock); InputOperatorImpl getInputOperator(SystemStream systemStream); void close(); Collection<InputOperatorImpl> getAllInputOperators();  }
@Test public void testEmptyChain() { StreamApplicationDescriptorImpl graphSpec = new StreamApplicationDescriptorImpl(appDesc -> { }, mock(Config.class)); OperatorImplGraph opGraph = new OperatorImplGraph(graphSpec.getOperatorSpecGraph(), context, mock(Clock.class)); assertEquals(0, opGraph.getAllInputOperators().size()); }
@Test public void testMergeChain() { String inputStreamId = "input"; String inputSystem = "input-system"; StreamApplicationDescriptorImpl graphSpec = new StreamApplicationDescriptorImpl(appDesc -> { GenericSystemDescriptor sd = new GenericSystemDescriptor(inputSystem, "mockFactoryClass"); GenericInputDescriptor inputDescriptor = sd.getInputDescriptor(inputStreamId, mock(Serde.class)); MessageStream<Object> inputStream = appDesc.getInputStream(inputDescriptor); MessageStream<Object> stream1 = inputStream.filter(mock(FilterFunction.class)); MessageStream<Object> stream2 = inputStream.map(mock(MapFunction.class)); stream1.merge(Collections.singleton(stream2)) .map(new TestMapFunction<Object, Object>("test-map-1", (Function & Serializable) m -> m)); }, getConfig()); TaskName mockTaskName = mock(TaskName.class); TaskModel taskModel = mock(TaskModel.class); when(taskModel.getTaskName()).thenReturn(mockTaskName); when(this.context.getTaskContext().getTaskModel()).thenReturn(taskModel); OperatorImplGraph opImplGraph = new OperatorImplGraph(graphSpec.getOperatorSpecGraph(), this.context, mock(Clock.class)); Set<OperatorImpl> opSet = opImplGraph.getAllInputOperators().stream().collect(HashSet::new, (s, op) -> addOperatorRecursively(s, op), HashSet::addAll); Object[] mergeOps = opSet.stream().filter(op -> op.getOperatorSpec().getOpCode() == OpCode.MERGE).toArray(); assertEquals(1, mergeOps.length); assertEquals(1, ((OperatorImpl) mergeOps[0]).registeredOperators.size()); OperatorImpl mapOp = (OperatorImpl) ((OperatorImpl) mergeOps[0]).registeredOperators.iterator().next(); assertEquals(mapOp.getOperatorSpec().getOpCode(), OpCode.MAP); assertEquals(TestMapFunction.getInstanceByTaskName(mockTaskName, "test-map-1").numInitCalled, 1); }