EventCorrelatingSpoutOutputCollector extends SpoutOutputCollector { @Override public List<Integer> emit(String streamId, List<Object> tuple) { StreamlineEvent newEvent = injectCorrelationInformation(tuple); return delegate.emit(streamId, new Values(newEvent)); } EventCorrelatingSpoutOutputCollector(TopologyContext topologyContext, SpoutOutputCollector delegate); @Override List<Integer> emit(String streamId, List<Object> tuple); @Override List<Integer> emit(List<Object> tuple); @Override List<Integer> emit(String streamId, List<Object> tuple, Object messageId); @Override List<Integer> emit(List<Object> tuple, Object messageId); @Override void emitDirect(int taskId, String streamId, List<Object> tuple); @Override void emitDirect(int taskId, List<Object> tuple); @Override void emitDirect(int taskId, String streamId, List<Object> tuple, Object messageId); @Override void emitDirect(int taskId, List<Object> tuple, Object messageId); @Override void reportError(Throwable throwable); @Override long getPendingCount();  }
@Test public void emit() throws Exception { String testStreamId = "testStreamId"; final Values tuple = new Values(INPUT_STREAMLINE_EVENT); String messageId = "testMessageId"; List<Integer> expectedTasks = Lists.newArrayList(TASK_1, TASK_2); setupExpectationsForTopologyContextEmit(); setupExpectationsForEventCorrelationInjector(); sut = new EventCorrelatingSpoutOutputCollector(mockedTopologyContext, mockedOutputCollector); Deencapsulation.setField(sut, "eventCorrelationInjector", mockEventCorrelationInjector); new Expectations() {{ mockedOutputCollector.emit(testStreamId, withAny(tuple)); this.result = expectedTasks; }}; List<Integer> tasks = sut.emit(testStreamId, tuple); assertEquals(expectedTasks, tasks); new Verifications() {{ List<Object> capturedValues; mockedOutputCollector.emit(testStreamId, capturedValues = withCapture()); StreamlineEventTestUtil.assertEventIsProperlyCopied((StreamlineEvent) capturedValues.get(0), INPUT_STREAMLINE_EVENT); List<Tuple> capturedParents; mockEventCorrelationInjector.injectCorrelationInformation(tuple, capturedParents = withCapture(), TEST_COMPONENT_NAME_FOR_STORM); assertEquals(0, capturedParents.size()); }}; new Expectations() {{ mockedOutputCollector.emit(withAny(tuple)); result = expectedTasks; }}; tasks = sut.emit(tuple); assertEquals(expectedTasks, tasks); new Verifications() {{ List<Object> capturedValues; mockedOutputCollector.emit(capturedValues = withCapture()); StreamlineEventTestUtil.assertEventIsProperlyCopied((StreamlineEvent) capturedValues.get(0), INPUT_STREAMLINE_EVENT); List<Tuple> capturedParents; mockEventCorrelationInjector.injectCorrelationInformation(tuple, capturedParents = withCapture(), TEST_COMPONENT_NAME_FOR_STORM); assertEquals(0, capturedParents.size()); }}; new Expectations() {{ mockedOutputCollector.emit(testStreamId, withAny(tuple), messageId); result = expectedTasks; }}; tasks = sut.emit(testStreamId, tuple, messageId); assertEquals(expectedTasks, tasks); new Verifications() {{ List<Object> capturedValues; mockedOutputCollector.emit(testStreamId, capturedValues = withCapture(), messageId); StreamlineEventTestUtil.assertEventIsProperlyCopied((StreamlineEvent) capturedValues.get(0), INPUT_STREAMLINE_EVENT); List<Tuple> capturedParents; mockEventCorrelationInjector.injectCorrelationInformation(tuple, capturedParents = withCapture(), TEST_COMPONENT_NAME_FOR_STORM); assertEquals(0, capturedParents.size()); }}; new Expectations() {{ mockedOutputCollector.emit(withAny(tuple), messageId); result = expectedTasks; }}; tasks = sut.emit(tuple, messageId); assertEquals(expectedTasks, tasks); new Verifications() {{ List<Object> capturedValues; mockedOutputCollector.emit(capturedValues = withCapture(), messageId); StreamlineEventTestUtil.assertEventIsProperlyCopied((StreamlineEvent) capturedValues.get(0), INPUT_STREAMLINE_EVENT); List<Tuple> capturedParents; mockEventCorrelationInjector.injectCorrelationInformation(tuple, capturedParents = withCapture(), TEST_COMPONENT_NAME_FOR_STORM); assertEquals(0, capturedParents.size()); }}; }