EventCorrelatingOutputCollector extends OutputCollector { @Override public void ack(Tuple tuple) { delegate.ack(tuple); } EventCorrelatingOutputCollector(TopologyContext topologyContext, OutputCollector delegate); @Override List<Integer> emit(String streamId, Tuple anchor, List<Object> tuple); @Override List<Integer> emit(Tuple anchor, List<Object> tuple); @Override List<Integer> emit(String streamId, List<Object> tuple); @Override List<Integer> emit(List<Object> tuple); @Override List<Integer> emit(Collection<Tuple> anchors, List<Object> tuple); @Override List<Integer> emit(String streamId, Collection<Tuple> anchors, List<Object> tuple); @Override void emitDirect(int taskId, String streamId, Tuple anchor, List<Object> tuple); @Override void emitDirect(int taskId, String streamId, List<Object> tuple); @Override void emitDirect(int taskId, Collection<Tuple> anchors, List<Object> tuple); @Override void emitDirect(int taskId, Tuple anchor, List<Object> tuple); @Override void emitDirect(int taskId, List<Object> tuple); @Override void emitDirect(int taskId, String streamId, Collection<Tuple> anchors, List<Object> tuple); @Override void ack(Tuple tuple); @Override void fail(Tuple tuple); @Override void resetTimeout(Tuple tuple); @Override void reportError(Throwable throwable);  }
@Test public void testAck() throws Exception { setupExpectationsForTuple(); setupExpectationsForTopologyContextNoEmit(); EventCorrelatingOutputCollector sut = getSystemUnderTest(); Tuple anchor = new TupleImpl(mockedTopologyContext, new Values(PARENT_STREAMLINE_EVENT), TASK_0, Utils.DEFAULT_STREAM_ID); sut.ack(anchor); new Verifications() {{ mockedOutputCollector.ack(anchor); times = 1; }}; }