FilterBolt extends BaseBasicBolt { @Override public void execute(Tuple input, BasicOutputCollector collector) { if (!filter(input, toFilter)) { collector.emit(new Values(input.getValue(1))); } } FilterBolt(T toFilter); @Override void execute(Tuple input, BasicOutputCollector collector); @Override void declareOutputFields(OutputFieldsDeclarer declarer); static boolean filter(Tuple input, T toFilter); static final String FIELDS; }
@Test public void shouldNotEmitIfFiltered() { when(tuple.getValue(0)).thenReturn(TO_FILTER); bolt.execute(tuple, collector); verify(tuple, never()).getValue(1); verifyZeroInteractions(collector); }
@Test public void shouldEmitSecondFieldIfNotFiltered() { when(tuple.getValue(0)).thenReturn(NON_FILTER); bolt.execute(tuple, collector); verify(tuple, times(1)).getValue(1); verify(collector, times(1)).emit(any(Values.class)); }