ConfigSubscriptionFactory::SubscriptionData& data) {

  const envoy::config::core::v3::ApiConfigSource& api_config_source =

      data.config_.api_config_source();

  CustomConfigValidatorsPtr custom_config_validators = std::make_unique<CustomConfigValidatorsImpl>(

      data.validation_visitor_, data.server_, api_config_source.config_validators());

  auto strategy_or_error = Utility::prepareJitteredExponentialBackOffStrategy(

      api_config_source, data.api_.randomGenerator(), SubscriptionFactory::RetryInitialDelayMs,

      SubscriptionFactory::RetryMaxDelayMs);

  THROW_IF_NOT_OK_REF(strategy_or_error.status());

  JitteredExponentialBackOffStrategyPtr backoff_strategy = std::move(strategy_or_error.value());

  auto factory_primary_or_error = Config::Utility::factoryForGrpcApiConfigSource(

      data.cm_.grpcAsyncClientManager(), api_config_source, data.scope_, true, 0, false);

  THROW_IF_NOT_OK_REF(factory_primary_or_error.status());

  absl::StatusOr<RateLimitSettings> rate_limit_settings_or_error =

      Utility::parseRateLimitSettings(api_config_source);

  THROW_IF_NOT_OK_REF(rate_limit_settings_or_error.status());

  absl::StatusOr<Grpc::RawAsyncClientSharedPtr> primary_client_or_error =

      factory_primary_or_error.value()->createUncachedRawAsyncClient();

  THROW_IF_NOT_OK_REF(primary_client_or_error.status());

  Grpc::RawAsyncClientSharedPtr primary_client = std::move(*primary_client_or_error);

  std::function<std::unique_ptr<Upstream::LoadStatsReporter>()> lrs_factory =

      [&, data, primary_client]() -> std::unique_ptr<Upstream::LoadStatsReporter> {

  GrpcMuxContext grpc_mux_context{

      /*primary_async_client_=*/std::move(primary_client),

      /*failover_async_client_=*/nullptr,

      /*dispatcher_=*/data.dispatcher_,

      /*service_method_=*/deltaGrpcMethod(data.type_url_),

      /*local_info_=*/data.local_info_,

      /*rate_limit_settings_=*/rate_limit_settings_or_error.value(),

      /*scope_=*/data.scope_,

      /*config_validators_=*/std::move(custom_config_validators),

      /*xds_resources_delegate_=*/{},

      /*xds_config_tracker_=*/data.xds_config_tracker_,

      /*backoff_strategy_=*/std::move(backoff_strategy),

      /*target_xds_authority_=*/"",

      /*eds_resources_cache_=*/nullptr, // No EDS resources cache needed from collections.

      /*skip_subsequent_node_=*/api_config_source.set_node_on_first_message_only(),

      /*load_stats_reporter_factory_=*/lrs_factory};

  return std::make_unique<GrpcCollectionSubscriptionImpl>(

      data.collection_locator_.value(), std::make_shared<Config::NewGrpcMuxImpl>(grpc_mux_context),

      data.callbacks_, data.resource_decoder_, data.stats_, data.dispatcher_,

      Utility::configSourceInitialFetchTimeout(data.config_), /*is_aggregated=*/false,

      data.options_);

}

    ConfigSubscriptionFactory::SubscriptionData& data) {

  return std::make_unique<GrpcCollectionSubscriptionImpl>(

      data.collection_locator_.value(), data.cm_.adsMux(), data.callbacks_, data.resource_decoder_,

      data.stats_, data.dispatcher_, Utility::configSourceInitialFetchTimeout(data.config_),

      /*is_aggregated=*/true, data.options_);

}

AdsCollectionConfigSubscriptionFactory::create(ConfigSubscriptionFactory::SubscriptionData& data) {

  return std::make_unique<GrpcCollectionSubscriptionImpl>(

      data.collection_locator_.value(), data.cm_.adsMux(), data.callbacks_, data.resource_decoder_,

      data.stats_, data.dispatcher_, Utility::configSourceInitialFetchTimeout(data.config_), true,

      data.options_);

}