class 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();
  }

}