const std::string& rds_type, RouteConfigProviderManager& route_config_provider_manager) {

  absl::Status creation_status = absl::OkStatus();

  auto ret = std::unique_ptr<RdsRouteConfigSubscription>(new RdsRouteConfigSubscription(

      std::move(config_update), std::move(resource_decoder), config_source, route_config_name,

      manager_identifier, factory_context, stat_prefix, rds_type, route_config_provider_manager,

      creation_status));

  RETURN_IF_NOT_OK(creation_status);

  return ret;

}

    : route_config_name_(route_config_name),

      scope_(factory_context.scope().createScope(stat_prefix + route_config_name_ + ".")),

      factory_context_(factory_context),

      parent_init_target_(

          fmt::format("RdsRouteConfigSubscription {} init {}", rds_type, route_config_name_),

          [this]() { local_init_manager_.initialize(local_init_watcher_); }),

      local_init_watcher_(fmt::format("{} local-init-watcher {}", rds_type, route_config_name_),

                          [this]() { parent_init_target_.ready(); }),

      local_init_target_(fmt::format("RdsRouteConfigSubscription {} local-init-target {}", rds_type,

                                     route_config_name_),

                         [this]() { subscription_->start({route_config_name_}); }),

      local_init_manager_(fmt::format("{} local-init-manager {}", rds_type, route_config_name_)),

      stat_prefix_(stat_prefix), rds_type_(rds_type),

      stats_({ALL_RDS_STATS(POOL_COUNTER(*scope_), POOL_GAUGE(*scope_))}),

      route_config_provider_manager_(route_config_provider_manager),

      manager_identifier_(manager_identifier), config_update_info_(std::move(config_update)),

      resource_decoder_(std::move(resource_decoder)) {

  const auto resource_type = route_config_provider_manager_.protoTraits().resourceType();

  auto subscription_or_error =

      Runtime::runtimeFeatureEnabled(

          "envoy.reloadable_features.xdstp_based_config_singleton_subscriptions")

          ? factory_context.xdsManager().subscribeToSingletonResource(

                route_config_name_, config_source, Envoy::Grpc::Common::typeUrl(resource_type),

                *scope_, *this, resource_decoder_, {})

          : factory_context.clusterManager().subscriptionFactory().subscriptionFromConfigSource(

                config_source, Envoy::Grpc::Common::typeUrl(resource_type), *scope_, *this,

                resource_decoder_, {});

  SET_AND_RETURN_IF_NOT_OK(subscription_or_error.status(), creation_status);

  subscription_ = std::move(*subscription_or_error);

  local_init_manager_.add(local_init_target_);

}

RdsRouteConfigSubscription::~RdsRouteConfigSubscription() {

  local_init_target_.ready();

  route_config_provider_manager_.eraseDynamicProvider(manager_identifier_);

}

    const std::string& version_info) {

  if (resources.empty()) {

    ENVOY_LOG(debug, "Missing {} RouteConfiguration for {} in onConfigUpdate()", rds_type_,

              route_config_name_);

    stats_.update_empty_.inc();

    local_init_target_.ready();

    return absl::OkStatus();

  }

  if (resources.size() != 1) {

    return absl::InvalidArgumentError(

        fmt::format("Unexpected {} resource length: {}", rds_type_, resources.size()));

  }

  const auto& route_config = resources[0].get().resource();

  Protobuf::ReflectableMessage reflectable_config = createReflectableMessage(route_config);

  if (reflectable_config->GetDescriptor()->full_name() !=

      route_config_provider_manager_.protoTraits().resourceType()) {

    return absl::InvalidArgumentError(

        fmt::format("Unexpected {} configuration type (expecting {}): {}", rds_type_,

                    route_config_provider_manager_.protoTraits().resourceType(),

                    reflectable_config->GetDescriptor()->full_name()));

  }

  if (resourceName(route_config_provider_manager_.protoTraits(), route_config) !=

      route_config_name_) {

    return absl::InvalidArgumentError(

        fmt::format("Unexpected {} configuration (expecting {}): {}", rds_type_, route_config_name_,

                    resourceName(route_config_provider_manager_.protoTraits(), route_config)));

  }

  std::unique_ptr<Init::ManagerImpl> noop_init_manager;

  std::unique_ptr<Cleanup> resume_rds;

  if (config_update_info_->onRdsUpdate(route_config, version_info)) {

    stats_.config_reload_.inc();

    stats_.config_reload_time_ms_.set(DateUtil::nowToMilliseconds(factory_context_.timeSource()));

    RETURN_IF_NOT_OK(beforeProviderUpdate(noop_init_manager, resume_rds));

    ENVOY_LOG(debug, "rds: loading new configuration: config_name={} hash={}", route_config_name_,

              config_update_info_->configHash());

    if (route_config_provider_ != nullptr) {

      RETURN_IF_NOT_OK(route_config_provider_->onConfigUpdate());

    }

    RETURN_IF_NOT_OK(afterProviderUpdate());

  }

  local_init_target_.ready();

  return absl::OkStatus();

}

    const Protobuf::RepeatedPtrField<std::string>& removed_resources, const std::string&) {

  if (!removed_resources.empty()) {

    ENVOY_LOG(trace,

              "Server sent a delta {} update attempting to remove a resource (name: {}). Ignoring.",

              rds_type_, removed_resources[0]);

  }

  if (!added_resources.empty()) {

    return onConfigUpdate(added_resources, added_resources[0].get().version());

  }

  return absl::OkStatus();

}

    Envoy::Config::ConfigUpdateFailureReason reason, const EnvoyException*) {

  ASSERT(Envoy::Config::ConfigUpdateFailureReason::ConnectionFailure != reason);

  local_init_target_.ready();

}