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(ExecutorOptions.class).getScheduledExecutorService(), bufferSize);
      }
      return OutboundObserverFactory.clientBuffered(
          options.as(ExecutorOptions.class).getScheduledExecutorService());
    }
    return OutboundObserverFactory.clientDirect();
  }
@Test
  public void testDefaultInstantiation() {
    StreamObserver<String> observer =
        HarnessStreamObserverFactories.fromOptions(PipelineOptionsFactory.create())
            .outboundObserverFor(this::fakeFactory, mockRequestObserver);
    assertThat(observer, instanceOf(DirectStreamObserver.class));
  }