BeamFnDataTimeBasedBufferingOutboundObserver extends BeamFnDataSizeBasedBufferingOutboundObserver<T> { @Override public void accept(T t) throws IOException { checkFlushThreadException(); super.accept(t); } BeamFnDataTimeBasedBufferingOutboundObserver(
      int sizeLimit,
      long timeLimit,
      LogicalEndpoint outputLocation,
      Coder<T> coder,
      StreamObserver<BeamFnApi.Elements> outboundObserver); @Override void close(); @Override synchronized void flush(); @Override void accept(T t);  }
@Test public void testConfiguredTimeLimit() throws Exception { List<Elements> values = new ArrayList<>(); PipelineOptions options = PipelineOptionsFactory.create(); options .as(ExperimentalOptions.class) .setExperiments(Arrays.asList("data_buffer_time_limit_ms=1")); final CountDownLatch waitForFlush = new CountDownLatch(1); CloseableFnDataReceiver<byte[]> consumer = BeamFnDataBufferingOutboundObserver.forLocation( options, endpoint, CODER, TestStreams.withOnNext( (Consumer<Elements>) e -> { values.add(e); waitForFlush.countDown(); }) .build()); consumer.accept(new byte[1]); waitForFlush.await(); assertEquals(messageWithData(new byte[1]), values.get(0)); }
@Test public void testConfiguredTimeLimit() throws Exception { Collection<Elements> values = new ArrayList<>(); PipelineOptions options = PipelineOptionsFactory.create(); options .as(ExperimentalOptions.class) .setExperiments(Arrays.asList("data_buffer_time_limit_ms=1")); final CountDownLatch waitForFlush = new CountDownLatch(1); CloseableFnDataReceiver<WindowedValue<byte[]>> consumer = BeamFnDataBufferingOutboundObserver.forLocation( options, OUTPUT_LOCATION, CODER, TestStreams.withOnNext( (Consumer<Elements>) e -> { values.add(e); waitForFlush.countDown(); }) .build()); consumer.accept(valueInGlobalWindow(new byte[1])); waitForFlush.await(); assertEquals(messageWithData(new byte[1]), Iterables.get(values, 0)); }