: fallback_picker_lb_config_{fallback_load_balancer_factory,

                                 std::move(fallback_load_balancer_config)},

      override_host_sources_(std::move(override_host_sources)),

      selected_host_key_(std::move(selected_host_key)) {}

OverrideHostLbConfig::OverrideSource::make(const OverrideHost::OverrideHostSource& config) {

  return OverrideHostLbConfig::OverrideSource{

      !config.header().empty()

          ? absl::optional<Http::LowerCaseString>(Http::LowerCaseString(config.header()))

          : absl::nullopt,

      config.has_metadata() ? absl::optional<Config::MetadataKey>(config.metadata())

                            : absl::nullopt};

}

    const Protobuf::RepeatedPtrField<OverrideHost::OverrideHostSource>& override_sources) {

  std::vector<OverrideSource> result;

  for (const OverrideHost::OverrideHostSource& source : override_sources) {

    result.push_back(OverrideSource::make(source));

    if (!result.back().header_name.has_value() && !result.back().metadata_key.has_value()) {

      return absl::InvalidArgumentError("Empty override source");

    }

    if (result.back().header_name.has_value() && result.back().metadata_key.has_value()) {

      return absl::InvalidArgumentError("Only one override source must be set");

    }

  }

  return result;

}

OverrideHostLbConfig::make(const OverrideHost& config, ServerFactoryContext& context) {

  absl::StatusOr<std::vector<OverrideSource>> override_host_sources =

      makeOverrideSources(config.override_host_sources());

  RETURN_IF_NOT_OK(override_host_sources.status());

  absl::optional<Config::MetadataKey> selected_host_key;

  if (config.has_selected_host_key()) {

    selected_host_key.emplace(config.selected_host_key());

  }

  ASSERT(config.has_fallback_policy());

  absl::InlinedVector<absl::string_view, 4> missing_policies;

  for (const auto& policy : config.fallback_policy().policies()) {

    TypedLoadBalancerFactory* factory =

        Envoy::Config::Utility::getAndCheckFactory<TypedLoadBalancerFactory>(

            policy.typed_extension_config(), /*is_optional=*/true);

    if (factory != nullptr) {

      auto proto_message = factory->createEmptyConfigProto();

      RETURN_IF_NOT_OK(Envoy::Config::Utility::translateOpaqueConfig(

          policy.typed_extension_config().typed_config(), context.messageValidationVisitor(),

          *proto_message));

      auto fallback_load_balancer_config = factory->loadConfig(context, *proto_message);

      RETURN_IF_NOT_OK_REF(fallback_load_balancer_config.status());

      return std::unique_ptr<OverrideHostLbConfig>(new OverrideHostLbConfig(

          std::move(override_host_sources).value(), std::move(selected_host_key), factory,

          std::move(fallback_load_balancer_config.value())));

    }

    missing_policies.push_back(policy.typed_extension_config().name());

  }

  return absl::InvalidArgumentError(

      absl::StrCat("dynamic forwarding LB: didn't find a registered fallback load balancer factory "

                   "with names from ",

                   absl::StrJoin(missing_policies, ", ")));

}

                                                                  TimeSource& time_source) const {

  return fallback_picker_lb_config_.load_balancer_factory->create(

      makeOptRefFromPtr<const LoadBalancerConfig>(

          fallback_picker_lb_config_.load_balancer_config.get()),

      cluster_info, priority_set, runtime, random, time_source);

}

absl::Status OverrideHostLoadBalancer::initialize() {

  ASSERT(fallback_picker_lb_ != nullptr); // Always needs a locality picker LB.

  return fallback_picker_lb_->initialize();

}

LoadBalancerFactorySharedPtr OverrideHostLoadBalancer::factory() {

  ASSERT_IS_MAIN_OR_TEST_THREAD();

  if (!factory_) {

    factory_ = std::make_shared<LoadBalancerFactoryImpl>(config_, fallback_picker_lb_->factory());

  }

  return factory_;

}

OverrideHostLoadBalancer::LoadBalancerImpl::peekAnotherHost(LoadBalancerContext* context) {

  return fallback_picker_lb_->peekAnotherHost(context);

}

OverrideHostLoadBalancer::LoadBalancerImpl::chooseHostInternal(LoadBalancerContext* context) {

  if (!context || !context->requestStreamInfo()) {

    return fallback_picker_lb_->chooseHost(context);

  }

  OverrideHostFilterState* override_host_state = nullptr;

  if (override_host_state =

          context->requestStreamInfo()->filterState()->getDataMutable<OverrideHostFilterState>(

              OverrideHostFilterState::kFilterStateKey);

      override_host_state == nullptr) {

    auto state_ptr = std::make_shared<OverrideHostFilterState>(getSelectedHosts(context));

    override_host_state = state_ptr.get();

    context->requestStreamInfo()->filterState()->setData(

        OverrideHostFilterState::kFilterStateKey, std::move(state_ptr),

        StreamInfo::FilterState::StateType::Mutable);

  }

  if (override_host_state->empty()) {

    ENVOY_LOG(trace, "No overridden hosts were found. Using fallback LB policy.");

    return fallback_picker_lb_->chooseHost(context);

  }

  if (HostConstSharedPtr host = getEndpoint(*override_host_state); host != nullptr) {

    return {host};

  }

  ENVOY_LOG(trace, "Failed to find any endpoints from metadata in the cluster. "

                   "Using fallback LB policy.");

  return fallback_picker_lb_->chooseHost(context);

}

OverrideHostLoadBalancer::LoadBalancerImpl::chooseHost(LoadBalancerContext* context) {

  auto response = chooseHostInternal(context);

  addSelectedHostKey(context, response);

  return response;

}

    LoadBalancerContext* context, HostSelectionResponse& response) {

  if (!config_.selectedHostKey().has_value()) {

    return;

  }

  if (response.host == nullptr) {

  const std::string selected_endpoint = response.host->address()->asString();

  const Config::MetadataKey& metadata_key = config_.selectedHostKey().value();

  if (metadata_key.path_.size() < 1) {

  Protobuf::Struct updated_metadata;

  Protobuf::Struct* updated_metadata_ptr = &updated_metadata;

  for (size_t i = 0; i + 1 < metadata_key.path_.size(); i++) {

  (*updated_metadata_ptr->mutable_fields())[metadata_key.path_.back()].set_string_value(

      selected_endpoint);

  context->requestStreamInfo()->setDynamicMetadata(metadata_key.key_, updated_metadata);

}

    const ::envoy::config::core::v3::Metadata& metadata, const Config::MetadataKey& metadata_key) {

  const Protobuf::Value& metadata_value = Config::Metadata::metadataValue(&metadata, metadata_key);

  if (metadata_value.has_string_value()) {

    return absl::string_view{metadata_value.string_value()};

  }

  return absl::nullopt;

}

    const Envoy::Http::RequestHeaderMap* header_map, const Http::LowerCaseString& header_name) {

  if (!header_map) {

    return absl::nullopt;

  }

  HeaderMap::GetResult result = header_map->get(header_name);

  if (result.empty()) {

    return absl::nullopt;

  }

  return result[0]->value().getStringView();

}

OverrideHostLoadBalancer::LoadBalancerImpl::getSelectedHosts(LoadBalancerContext* context) {

  ASSERT(!config_.overrideHostSources().empty());

  std::vector<std::string> selected_hosts;

  selected_hosts.reserve(4);

  for (const auto& override_source : config_.overrideHostSources()) {

    ASSERT(override_source.header_name.has_value() != override_source.metadata_key.has_value());

    absl::optional<absl::string_view> hosts;

    if (override_source.header_name.has_value()) {

      hosts = getSelectedHostsFromHeader(context->downstreamHeaders(),

                                         override_source.header_name.value());

    } else if (override_source.metadata_key.has_value()) {

      hosts = getSelectedHostsFromMetadata(context->requestStreamInfo()->dynamicMetadata(),

                                           override_source.metadata_key.value());

    }

    if (!hosts.has_value()) {

      continue;

    }

    for (absl::string_view host : absl::StrSplit(hosts.value(), ',', absl::SkipWhitespace())) {

      selected_hosts.push_back(std::string(absl::StripAsciiWhitespace(host)));

    }

  }

  ENVOY_LOG(trace, "Selected endpoints: {}", absl::StrJoin(selected_hosts, ","));

  return selected_hosts;

}

OverrideHostLoadBalancer::LoadBalancerImpl::findHost(absl::string_view endpoint) {

  HostMapConstSharedPtr hosts = priority_set_.crossPriorityHostMap();

  if (hosts == nullptr) {

    return nullptr;

  }

  ENVOY_LOG(trace, "Looking up {} in {}", endpoint,

            absl::StrJoin(*hosts, ", ",

                          [](std::string* out, Envoy::Upstream::HostMap::const_reference entry) {

                            absl::StrAppend(out, entry.first);

                          }));

  if (const auto host_iterator = hosts->find(endpoint); host_iterator != hosts->end()) {

    return host_iterator->second;

  }

  return nullptr;

}

    OverrideHostFilterState& override_host_state) {

  absl::string_view override_host = override_host_state.consumeNextHost();

  while (!override_host.empty()) {

    if (HostConstSharedPtr host = findHost(override_host); host != nullptr) {

      ENVOY_LOG(trace, "Selected endpoint: {}", override_host);

      return host;

    }

    override_host = override_host_state.consumeNextHost();

  }

  ENVOY_LOG(trace, "Number of attempts has exceeded the number of override hosts.");

  return nullptr;

}

OverrideHostLoadBalancer::LoadBalancerFactoryImpl::create(LoadBalancerParams params) {

  LoadBalancerPtr fallback_picker_lb = fallback_picker_lb_factory_->create(params);

  ASSERT(fallback_picker_lb != nullptr); // Factory can not create null LB.

  return std::make_unique<LoadBalancerImpl>(config_, std::move(fallback_picker_lb),

                                            params.priority_set);

}