AbstractSource extends ListenableSource<E> { public final void processEvent(final E event) { try { if (!eventFilter.apply(event)) { log.debug("Event filtered from source {}. Skipping. event={}", name, event); return; } notifyEvent(event); final Stopwatch stopwatch = Stopwatch.createStarted(); metrics.eventReceived(event); log.debug("Received event from source {}. event={}", name, event); notifyMutations(mutationMapper.map(event)); stopwatch.stop(); final long time = stopwatch.elapsed(TimeUnit.MILLISECONDS); metrics.processEventTime(event, time); } catch (Exception ex) { if (!isStarted()) { return; } final String errorMessage = String.format("Failed to process event from source %s", name); log.error(errorMessage, ex); metrics.eventFailure(ex); notifyError(ex); throw new SourceException(errorMessage, ex); } }  @Override final void open(); @Override final void close(); @Override final void checkpoint(Mutation<?> mutation); @Override void clear(); final void processEvent(final E event); final void addEventValidator(@NonNull final Validator<E> validator); final void addMutationValidator(
      @NonNull final Validator<M> validator);  }
@Test public void testProcessEvent() throws Exception { List mutations = Collections.singletonList(mock(Mutation.class)); when(mapper.map(event)).thenReturn(mutations); when(filter.apply(event)).thenReturn(false); source.processEvent(event); verifyZeroInteractions(metrics); verify(listener, times(0)).onMutation(mutations); when(filter.apply(event)).thenReturn(true); source.processEvent(event); verify(listener, times(1)).onMutation(mutations); when(mapper.map(event)).thenReturn(Collections.emptyList()); source.processEvent(event); verify(listener, times(2)).onEvent(event); verify(metrics, times(2)).eventReceived(event); verify(listener, times(1)).onMutation(mutations); }
@Test(expected = SourceException.class) public void testProcessEventFailure() throws Exception { RuntimeException testException = new RuntimeException(); when(filter.apply(event)).thenThrow(testException); source.setStarted(false); source.processEvent(event); verify(listener, times(0)).onMutation(any()); verify(metrics, times(0)).eventFailure(testException); verify(listener, times(0)).onError(testException); source.setStarted(true); source.processEvent(event); verify(listener, times(0)).onMutation(any()); verify(metrics, times(1)).eventFailure(testException); verify(listener, times(1)).onError(testException); }