FlattenOperation extends ReceivingOperation { @Override public void process(Object elem) throws Exception { try (Closeable scope = context.enterProcess()) { checkStarted(); Receiver receiver = receivers[0]; if (receiver != null) { receiver.process(elem); } } } FlattenOperation(OutputReceiver[] receivers, OperationContext context); @VisibleForTesting  FlattenOperation(OutputReceiver outputReceiver, OperationContext context); @Override void process(Object elem); @Override boolean supportsRestart();  }
@Test public void testRunFlattenOperation() throws Exception { TestOutputReceiver receiver = new TestOutputReceiver( counterSet, NameContext.create("test", "receiver", "receiver", "receiver")); OperationContext context = TestOperationContext.create(counterSet, nameContext); FlattenOperation flattenOperation = new FlattenOperation(receiver, context); flattenOperation.start(); flattenOperation.process("hi"); flattenOperation.process("there"); flattenOperation.process(""); flattenOperation.process("bob"); flattenOperation.finish(); Assert.assertThat( receiver.outputElems, CoreMatchers.<Object>hasItems("hi", "there", "", "bob")); CounterUpdateExtractor<?> updateExtractor = Mockito.mock(CounterUpdateExtractor.class); counterSet.extractUpdates(false, updateExtractor); verify(updateExtractor).longSum(getObjectCounterName("test_receiver_out"), false, 4L); verify(updateExtractor) .longMean( getMeanByteCounterName("test_receiver_out"), false, LongCounterMean.ZERO.addValue(14L, 4)); verifyNoMoreInteractions(updateExtractor); }
@Test public void testFlattenOperationContext() throws Exception { OperationContext context = Mockito.mock(OperationContext.class); OutputReceiver receiver = Mockito.mock(OutputReceiver.class); FlattenOperation operation = new FlattenOperation(receiver, context); Closeable startCloseable = Mockito.mock(Closeable.class); Closeable processCloseable = Mockito.mock(Closeable.class); Closeable finishCloseable = Mockito.mock(Closeable.class); when(context.enterStart()).thenReturn(startCloseable); when(context.enterProcess()).thenReturn(processCloseable); when(context.enterFinish()).thenReturn(finishCloseable); operation.start(); operation.process("hello"); operation.finish(); InOrder inOrder = Mockito.inOrder(receiver, context, startCloseable, processCloseable, finishCloseable); inOrder.verify(context).enterProcess(); inOrder.verify(receiver).process("hello"); inOrder.verify(processCloseable).close(); }