EventCorrelatingSpoutOutputCollector extends SpoutOutputCollector { @Override public void emitDirect(int taskId, String streamId, List<Object> tuple) { StreamlineEvent newEvent = injectCorrelationInformation(tuple); delegate.emitDirect(taskId, 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 emitDirect() throws Exception { setupExpectationsForTopologyContextEmit(); setupExpectationsForEventCorrelationInjector(); sut = new EventCorrelatingSpoutOutputCollector(mockedTopologyContext, mockedOutputCollector); Deencapsulation.setField(sut, "eventCorrelationInjector", mockEventCorrelationInjector); String testStreamId = "testStreamId"; final Values tuple = new Values(INPUT_STREAMLINE_EVENT); String messageId = "testMessageId"; new Expectations() {{ mockedOutputCollector.emitDirect(TASK_1, testStreamId, withAny(tuple)); }}; sut.emitDirect(TASK_1, testStreamId, tuple); new Verifications() {{ List<Object> capturedValues; mockedOutputCollector.emitDirect(TASK_1, 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.emitDirect(TASK_1, withAny(tuple)); }}; sut.emitDirect(TASK_1, tuple); new Verifications() {{ List<Object> capturedValues; mockedOutputCollector.emitDirect(TASK_1, 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.emitDirect(TASK_1, testStreamId, withAny(tuple), messageId); }}; sut.emitDirect(TASK_1, testStreamId, tuple, messageId); new Verifications() {{ List<Object> capturedValues; mockedOutputCollector.emitDirect(TASK_1, 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.emitDirect(TASK_1, withAny(tuple), messageId); }}; sut.emitDirect(TASK_1, tuple, messageId); new Verifications() {{ List<Object> capturedValues; mockedOutputCollector.emitDirect(TASK_1, 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()); }}; }