BoundedSourceRunner { public void runReadLoop(WindowedValue<InputT> value) throws Exception { try (Reader<OutputT> reader = value.getValue().createReader(pipelineOptions)) { if (!reader.start()) { return; } do { WindowedValue<OutputT> nextValue = WindowedValue.timestampedValueInGlobalWindow( reader.getCurrent(), reader.getCurrentTimestamp()); for (FnDataReceiver<WindowedValue<OutputT>> consumer : consumers) { consumer.accept(nextValue); } } while (reader.advance()); } } BoundedSourceRunner(
      PipelineOptions pipelineOptions,
      RunnerApi.FunctionSpec definition,
      Collection<FnDataReceiver<WindowedValue<OutputT>>> consumers); @Deprecated void start(); void runReadLoop(WindowedValue<InputT> value); @Override String toString();  }
@Test public void testRunReadLoopWithMultipleSources() throws Exception { List<WindowedValue<Long>> out1Values = new ArrayList<>(); List<WindowedValue<Long>> out2Values = new ArrayList<>(); Collection<FnDataReceiver<WindowedValue<Long>>> consumers = ImmutableList.of(out1Values::add, out2Values::add); BoundedSourceRunner<BoundedSource<Long>, Long> runner = new BoundedSourceRunner<>( PipelineOptionsFactory.create(), RunnerApi.FunctionSpec.getDefaultInstance(), consumers); runner.runReadLoop(valueInGlobalWindow(CountingSource.upTo(2))); runner.runReadLoop(valueInGlobalWindow(CountingSource.upTo(1))); assertThat( out1Values, contains(valueInGlobalWindow(0L), valueInGlobalWindow(1L), valueInGlobalWindow(0L))); assertThat( out2Values, contains(valueInGlobalWindow(0L), valueInGlobalWindow(1L), valueInGlobalWindow(0L))); }
@Test public void testRunReadLoopWithEmptySource() throws Exception { List<WindowedValue<Long>> outValues = new ArrayList<>(); Collection<FnDataReceiver<WindowedValue<Long>>> consumers = ImmutableList.of(outValues::add); BoundedSourceRunner<BoundedSource<Long>, Long> runner = new BoundedSourceRunner<>( PipelineOptionsFactory.create(), RunnerApi.FunctionSpec.getDefaultInstance(), consumers); runner.runReadLoop(valueInGlobalWindow(CountingSource.upTo(0))); assertThat(outValues, empty()); }