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