ApexStreamImpl implements ApexStream<T> { @Override @SuppressWarnings("unchecked") public <O, STREAM extends ApexStream<O>> STREAM addOperator(Operator op, Operator.InputPort<T> inputPort, Operator.OutputPort<O> outputPort, Option... opts) { checkArguments(op, inputPort, outputPort); DagMeta.NodeMeta nm = null; if (lastBrick == null) { nm = graph.addNode(op, null, null, inputPort, opts); } else { nm = graph.addNode(op, lastBrick.nodeMeta, lastBrick.lastOutput, inputPort, opts); } Brick<O> newBrick = new Brick<>(); newBrick.nodeMeta = nm; newBrick.setLastOutput(outputPort); if (lastBrick != null) { newBrick.lastStream = Pair.<Operator.OutputPort, Operator.InputPort>of(lastBrick.lastOutput, inputPort); } if (this.getClass() == ApexStreamImpl.class || this.getClass() == ApexWindowedStreamImpl.class) { return (STREAM)newStream(this.graph, newBrick); } else { try { return (STREAM)this.getClass().getConstructor(ApexStreamImpl.class).newInstance(newStream(this.graph, newBrick)); } catch (Exception e) { throw new RuntimeException("You have to override the default constructor with ApexStreamImpl as default parameter", e); } } } ApexStreamImpl();  ApexStreamImpl(ApexStreamImpl<T> apexStream);  ApexStreamImpl(DagMeta graph);  ApexStreamImpl(DagMeta graph, Brick<T> lastBrick); Brick<T> getLastBrick(); void setLastBrick(Brick<T> lastBrick); @Override STREAM map(Function.MapFunction<T, O> mf, Option... opts); @Override STREAM flatMap(FlatMapFunction<T, O> flatten, Option... opts); @Override @SuppressWarnings("unchecked") STREAM filter(final Function.FilterFunction<T> filter, Option... opts); STREAM countByElement(); @Override STREAM endWith(Operator op, Operator.InputPort<T> inputPort, Option... opts); @Override @SuppressWarnings("unchecked") STREAM addOperator(Operator op, Operator.InputPort<T> inputPort, Operator.OutputPort<O> outputPort, Option... opts); @Override OUTSTREAM addCompositeStreams(CompositeStreamTransform<INSTREAM, OUTSTREAM> compositeStreamTransform); @Override STREAM union(ApexStream<T>... others); @Override @SuppressWarnings("unchecked") ApexStreamImpl<T> print(Option... opts); @Override @SuppressWarnings("unchecked") ApexStreamImpl<T> print(); @Override @SuppressWarnings("unchecked") ApexStream<T> printErr(); @Override @SuppressWarnings("unchecked") ApexStream<T> with(Attribute attribute, Object value); @Override @SuppressWarnings("unchecked") ApexStream<T> setGlobalAttribute(Attribute attribute, Object value); @Override @SuppressWarnings("unchecked") ApexStream<T> with(DAG.Locality locality); @Override @SuppressWarnings("unchecked") ApexStream<T> with(String propName, Object value); @Override DAG createDag(); @Override void populateDag(DAG dag); @Override void runEmbedded(boolean async, long duration, Callable<Boolean> exitCondition); @Override void run(); @Override WindowedStream<T> window(WindowOption option); @Override WindowedStream<T> window(WindowOption windowOption, TriggerOption triggerOption); @Override WindowedStream<T> window(WindowOption windowOption, TriggerOption triggerOption, Duration allowLateness);  }
@Test public void testAddOperator() { LogicalPlan dag = new LogicalPlan(); TestOperator<String, Integer> firstOperator = new TestOperator<>(); TestOperator<Integer, Date> secondOperator = new TestOperator<>(); new ApexStreamImpl<String>().addOperator(firstOperator, null, firstOperator.output, name("first")) .endWith(secondOperator, secondOperator.input, name("second")) .with(DAG.Locality.THREAD_LOCAL) .with(Context.OperatorContext.AUTO_RECORD, true) .with("prop", "TestProp").populateDag(dag); Assert.assertTrue(dag.getAllOperators().size() == 2); Set<String> opNames = new HashSet<>(); opNames.add("first"); opNames.add("second"); for (LogicalPlan.OperatorMeta operatorMeta : dag.getAllOperators()) { Assert.assertTrue(operatorMeta.getOperator() instanceof TestOperator); Assert.assertTrue(opNames.contains(operatorMeta.getName())); if (operatorMeta.getName().equals("second")) { Assert.assertTrue(operatorMeta.getAttributes().get(Context.OperatorContext.AUTO_RECORD)); Assert.assertEquals("TestProp", ((TestOperator)operatorMeta.getOperator()).prop); } else { Assert.assertNull(operatorMeta.getAttributes().get(Context.OperatorContext.AUTO_RECORD)); Assert.assertNull(((TestOperator)operatorMeta.getOperator()).prop); } } Collection<LogicalPlan.StreamMeta> streams = dag.getAllStreams(); Assert.assertTrue(streams.size() == 1); for (LogicalPlan.StreamMeta stream : streams) { Assert.assertEquals("first", stream.getSource().getOperatorMeta().getName()); Collection<InputPortMeta> portMetaCollection = stream.getSinks(); Assert.assertTrue(1 == portMetaCollection.size()); for (InputPortMeta inputPortMeta : portMetaCollection) { Assert.assertEquals("second", inputPortMeta.getOperatorMeta().getName()); } Assert.assertTrue(stream.getLocality() == DAG.Locality.THREAD_LOCAL); } }