@Override
  public void process(Object elem) throws Exception {
    try (Closeable scope = context.enterProcess()) {
      checkStarted();
      Receiver receiver = receivers[0];
      if (receiver != null) {
        receiver.process(elem);
      }
    }
  }
@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();

    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);
  }