HarnessStreamObserverFactories { public static OutboundObserverFactory fromOptions(PipelineOptions options) { List<String> experiments = options.as(ExperimentalOptions.class).getExperiments(); if (experiments != null && experiments.contains("beam_fn_api_buffered_stream")) { int bufferSize = getBufferSize(experiments); if (bufferSize > 0) { return OutboundObserverFactory.clientBuffered( options.as(GcsOptions.class).getExecutorService(), bufferSize); } return OutboundObserverFactory.clientBuffered( options.as(GcsOptions.class).getExecutorService()); } return OutboundObserverFactory.clientDirect(); }  static OutboundObserverFactory fromOptions(PipelineOptions options);  }
@Test public void testDefaultInstantiation() { StreamObserver<String> observer = HarnessStreamObserverFactories.fromOptions(PipelineOptionsFactory.create()) .outboundObserverFor(this::fakeFactory, mockRequestObserver); assertThat(observer, instanceOf(DirectStreamObserver.class)); }
@Test public void testBufferedStreamInstantiation() { StreamObserver<String> observer = HarnessStreamObserverFactories.fromOptions( PipelineOptionsFactory.fromArgs( new String[] {"--experiments=beam_fn_api_buffered_stream"}) .create()) .outboundObserverFor(this::fakeFactory, mockRequestObserver); assertThat(observer, instanceOf(BufferingStreamObserver.class)); }