CrunchExecutor { public void run() { runWithSink(this.sink); } CrunchExecutor(MRecordSource source, EvaluationPipeline evaluationPipeline, EventSink sink);  CrunchExecutor(MRecordSource source, EvaluationPipeline evaluationPipeline); void run(); void run(EventSink sink);  }
@Test public void run() { Function<Long, UntypedValues> values = VALUES_FACTORY; MRecordSource source = MRecordSources.of(values.apply(123L), values.apply(124L)); EvaluationPipeline pipeline = createPipeline(); EventSink sink = Mockito.mock(EventSink.class); CrunchExecutor crunchExecutor = new CrunchExecutor(source, pipeline, sink); crunchExecutor.run(); Mockito.verify(sink, times(1)).handle(any()); }
@Test public void runWithTwoSubstreams() { UntypedValues values = UntypedValues.builder() .source("test") .prefix("") .timestamp(123L) .values(Collections.singletonMap("test", "test")) .build(); MRecordSource source = MRecordSources.of(VALUES_FACTORY.apply(123L), VALUES_FACTORY.apply(124L), VALUES_FACTORY.apply(125L)); EvaluationPipeline pipeline = createPipelineWithTwoSubstreams(); EventSink sink = Mockito.mock(EventSink.class); CrunchExecutor crunchExecutor = new CrunchExecutor(source, pipeline, sink); crunchExecutor.run(); Mockito.verify(sink, times(4)).handle(any()); }
@Test public void resultHandlerSendsBeforeFilter() { UntypedValues values = UntypedValues.builder() .source("test") .prefix("") .timestamp(123L) .values(Collections.singletonMap("test", "test")) .build(); MRecordSource source = MRecordSources.of(VALUES_FACTORY.apply(123L), VALUES_FACTORY.apply(124L), VALUES_FACTORY.apply(125L)); final TestRecordHandler handler = new TestRecordHandler(); EvaluationPipeline pipeline = createPipelineWithRecordHandler(handler); CrunchExecutor crunchExecutor = new CrunchExecutor(source, pipeline, NoOpSink.INSTANCE); crunchExecutor.run(); assertEquals(3, TestRecordHandler.records.size()); }