: factory_context_(factory_context), cluster_(cluster),

      config_(std::make_shared<Router::FilterConfig>(

          factory_context, http_context.asyncClientStatPrefix(), *stats_store.rootScope(), cm,

          factory_context.runtime(), factory_context.api().randomGenerator(),

          std::move(shadow_writer), true, false, false, false, false, false, false,

          Protobuf::RepeatedPtrField<std::string>{}, dispatcher.timeSource(), http_context,

          router_context)),

      dispatcher_(dispatcher), local_reply_(LocalReply::Factory::createDefault()) {}

AsyncClientImpl::~AsyncClientImpl() {

  while (!active_streams_.empty()) {

    active_streams_.front()->reset();

  }

}

template <typename T> T* AsyncClientImpl::internalStartRequest(T* async_request) {

  if (!async_request) {

  async_request->initialize();

  std::unique_ptr<AsyncStreamImpl> new_request{async_request};

  if (!new_request->remote_closed_) {

    LinkedList::moveIntoList(std::move(new_request), active_streams_);

    return async_request;

  } else {

    new_request->cleanup();

    return nullptr;

  }

}

                                            const AsyncClient::RequestOptions& options) {

  AsyncRequestImpl* async_request =

      AsyncRequestImpl::create(std::move(request), *this, callbacks, options);

  return internalStartRequest(async_request);

}

                              const AsyncClient::RequestOptions& options) {

  AsyncOngoingRequestImpl* async_request =

      AsyncOngoingRequestImpl::create(std::move(request_headers), *this, callbacks, options);

  return internalStartRequest(async_request);

}

                                            const AsyncClient::StreamOptions& options) {

  auto stream_or_error = AsyncStreamImpl::create(*this, callbacks, options);

  if (!stream_or_error.ok()) {

  LinkedList::moveIntoList(std::move(stream_or_error.value()), active_streams_);

  return active_streams_.front().get();

}

                  absl::Status& creation_status) {

  if (options.retry_policy.has_value()) {

    auto policy_or_error = Router::RetryPolicyImpl::create(

        options.retry_policy.value(), ProtobufMessage::getNullValidationVisitor(), context);

    creation_status = policy_or_error.status();

    return policy_or_error.status().ok() ? std::move(policy_or_error.value())

                                         : Router::RetryPolicyImpl::DefaultRetryPolicy;

  }

  return options.parsed_retry_policy != nullptr ? options.parsed_retry_policy

                                                : Router::RetryPolicyImpl::DefaultRetryPolicy;

}

    : parent_(parent),

      discard_response_body_(options.discard_response_body),

      new_async_client_retry_logic_(Runtime::runtimeFeatureEnabled(

          "envoy.reloadable_features.http_async_client_retry_respect_buffer_limits")),

      buffer_limit_(options.buffer_limit_), stream_callbacks_(callbacks),

      stream_id_(parent.config_->random_.random()),

      router_(options.filter_config_ ? options.filter_config_ : parent.config_,

              parent.config_->async_stats_),

      stream_info_(Protocol::Http11, parent.dispatcher().timeSource(), nullptr,

                   options.filter_state != nullptr

                       ? options.filter_state

                       : std::make_shared<StreamInfo::FilterStateImpl>(

                             StreamInfo::FilterState::LifeSpan::FilterChain)),

      tracing_config_(Tracing::EgressConfig::get()), local_reply_(*parent.local_reply_),

      account_(options.account_), send_xff_(options.send_xff),

      send_internal_(options.send_internal),

      upstream_override_host_(options.upstream_override_host_) {

  auto retry_policy = createRetryPolicy(options, parent.factory_context_, creation_status);

  if (!creation_status.ok()) {

    return;

  }

  const Router::MetadataMatchCriteria* metadata_matching_criteria = nullptr;

  if (options.parent_context.stream_info != nullptr) {

    stream_info_.setParentStreamInfo(*options.parent_context.stream_info);

    parent_route_ = options.parent_context.stream_info->route();

    if (parent_route_ != nullptr) {

      const auto* route_entry = parent_route_->routeEntry();

      if (route_entry != nullptr) {

        metadata_matching_criteria = route_entry->metadataMatchCriteria();

      }

    }

  }

  auto route_or_error = NullRouteImpl::create(

      parent_.cluster_->name(), std::move(retry_policy), parent_.factory_context_.regexEngine(),

      options.timeout, options.hash_policy, metadata_matching_criteria);

  SET_AND_RETURN_IF_NOT_OK(route_or_error.status(), creation_status);

  route_ = std::move(*route_or_error);

  stream_info_.dynamicMetadata().MergeFrom(options.metadata);

  stream_info_.setIsShadow(options.is_shadow);

  stream_info_.setUpstreamClusterInfo(parent_.cluster_);

  stream_info_.route_ = route_;

  if (options.buffer_body_for_retry) {

    buffered_body_ = std::make_unique<Buffer::OwnedImpl>(account_);

  }

  router_.setDecoderFilterCallbacks(*this);

}

                                     absl::string_view details) {

  stream_info_.setResponseCodeDetails(details);

  if (encoded_response_headers_) {

    resetStream();

    return;

  }

  Utility::sendLocalReply(

      remote_closed_,

      Utility::EncodeFunctions{

          [modify_headers](ResponseHeaderMap& headers) -> void {

            if (modify_headers != nullptr) {

              modify_headers(headers);

            }

          },

          [this](ResponseHeaderMap& response_headers, Code& code, std::string& body,

                 absl::string_view& content_type) -> void {

            local_reply_.rewrite(request_headers_, response_headers, stream_info_, code, body,

                                 content_type);

          },

          [this, &details](ResponseHeaderMapPtr&& headers, bool end_stream) -> void {

            encodeHeaders(std::move(headers), end_stream, details);

          },

          [this](Buffer::Instance& data, bool end_stream) -> void {

            encodeData(data, end_stream);

          }},

      Utility::LocalReplyData{is_grpc_request_, code, body, grpc_status, is_head_request_});

}

                                    absl::string_view) {

  ENVOY_LOG(debug, "async http request response headers (end_stream={}):\n{}", end_stream,

            *headers);

  ASSERT(!remote_closed_);

  encoded_response_headers_ = true;

  stream_callbacks_.onHeaders(std::move(headers), end_stream);

  closeRemote(end_stream);

  closeLocal(end_stream);

}

void AsyncStreamImpl::encodeData(Buffer::Instance& data, bool end_stream) {

  ENVOY_LOG(trace, "async http request response data (length={} end_stream={})", data.length(),

            end_stream);

  ASSERT(!remote_closed_);

  stream_callbacks_.onData(data, end_stream);

  closeRemote(end_stream);

  closeLocal(end_stream);

}

void AsyncStreamImpl::encodeTrailers(ResponseTrailerMapPtr&& trailers) {

  ENVOY_LOG(debug, "async http request response trailers:\n{}", *trailers);

  ASSERT(!remote_closed_);

  stream_callbacks_.onTrailers(std::move(trailers));

  closeRemote(true);

  closeLocal(true);

}

void AsyncStreamImpl::sendHeaders(RequestHeaderMap& headers, bool end_stream) {

  request_headers_ = &headers;

  if (Http::Headers::get().MethodValues.Head == headers.getMethodValue()) {

    is_head_request_ = true;

  }

  is_grpc_request_ = Grpc::Common::isGrpcRequestHeaders(headers);

  if (send_internal_) {

    headers.setReferenceEnvoyInternalRequest(Headers::get().EnvoyInternalRequestValues.True);

  }

  if (send_xff_) {

    Utility::appendXff(headers, *parent_.config_->factory_context_.localInfo().address());

  }

  router_.decodeHeaders(headers, end_stream);

  closeLocal(end_stream);

}

void AsyncStreamImpl::sendData(Buffer::Instance& data, bool end_stream) {

  ASSERT(dispatcher().isThreadSafe());

  if (local_closed_) {

    return;

  }

  if (!new_async_client_retry_logic_) {

    if (buffered_body_ != nullptr) {

      if (buffered_body_->length() + data.length() > kDefaultDecoderBufferLimit) {

        ENVOY_LOG_EVERY_POW_2(

            warn, "the buffer size limit (64KB) for async client retries has been exceeded.");

      } else {

        buffered_body_->add(data);

      }

    }

  }

  if (router_.awaitingHost()) {

    ENVOY_LOG_EVERY_POW_2(warn, "the buffer limit for the async client has been exceeded "

                                "due to async host selection");

    reset();

    return;

  }

  router_.decodeData(data, end_stream);

  closeLocal(end_stream);

}

void AsyncStreamImpl::sendTrailers(RequestTrailerMap& trailers) {

  request_trailers_ = &trailers;

  ASSERT(dispatcher().isThreadSafe());

  if (local_closed_) {

    return;

  }

  router_.decodeTrailers(trailers);

  closeLocal(true);

}

void AsyncStreamImpl::closeLocal(bool end_stream) {

  if (local_closed_) {

    return;

  }

  local_closed_ = end_stream;

  if (complete()) {

    stream_callbacks_.onComplete();

    cleanup();

  }

}

void AsyncStreamImpl::closeRemote(bool end_stream) {

  if (remote_closed_) {

    return;

  }

  remote_closed_ = end_stream;

  if (complete()) {

    stream_callbacks_.onComplete();

    cleanup();

  }

}

void AsyncStreamImpl::reset() {

  routerDestroy();

  resetStream();

}

void AsyncStreamImpl::routerDestroy() {

  if (!router_destroyed_) {

    router_destroyed_ = true;

    router_.onDestroy();

  }

}

void AsyncStreamImpl::cleanup() {

  ASSERT(dispatcher().isThreadSafe());

  local_closed_ = remote_closed_ = true;

  if (inserted()) {

    routerDestroy();

    dispatcher().deferredDelete(removeFromList(parent_.active_streams_));

  }

}

void AsyncStreamImpl::resetStream(Http::StreamResetReason, absl::string_view) {

  stream_callbacks_.onReset();

  cleanup();

}

    : AsyncStreamImpl(parent, *this, options, creation_status), callbacks_(callbacks),

      response_buffer_limit_(parent.config_->runtime_.snapshot().getInteger(

          AsyncClientImpl::ResponseBufferLimit, kBufferLimitForResponse)) {

  if (!creation_status.ok()) {

  if (options.parent_span_ != nullptr) {

    const std::string child_span_name =

        options.child_span_name_.empty()

            ? absl::StrCat("async ", parent.cluster_->name(), " egress")

            : options.child_span_name_;

    child_span_ = options.parent_span_->spawnChild(Tracing::EgressConfig::get(), child_span_name,

                                                   parent.dispatcher().timeSource().systemTime());

  } else {

    child_span_ = std::make_unique<Tracing::NullSpan>();

  }

  if (options.sampled_.has_value()) {

    child_span_->setSampled(options.sampled_.value());

  }

}

void AsyncRequestImpl::initialize() {

  Tracing::HttpTraceContext trace_context(request_->headers());

  Tracing::UpstreamContext upstream_context(nullptr,                    // host_

                                            parent_.cluster_.get(),     // cluster_

                                            Tracing::ServiceType::Http, // service_type_

                                            true                        // async_client_span_

  );

  child_span_->injectContext(trace_context, upstream_context);

  sendHeaders(request_->headers(), request_->body().length() == 0);

  if (request_->body().length() != 0) {

    sendData(request_->body(), true);

  }

}

void AsyncOngoingRequestImpl::initialize() {

  Tracing::HttpTraceContext trace_context(*request_headers_);

  Tracing::UpstreamContext upstream_context(nullptr,                    // host_

                                            parent_.cluster_.get(),     // cluster_

                                            Tracing::ServiceType::Http, // service_type_

                                            true                        // async_client_span_

  );

  child_span_->injectContext(trace_context, upstream_context);

  sendHeaders(*request_headers_, false);

}

void AsyncRequestSharedImpl::onComplete() {

  complete_ = true;

  callbacks_.onBeforeFinalizeUpstreamSpan(*child_span_, &response_->headers());

  Tracing::HttpTracerUtility::finalizeUpstreamSpan(*child_span_, streamInfo(),

                                                   Tracing::EgressConfig::get());

  callbacks_.onSuccess(*this, std::move(response_));

}

void AsyncRequestSharedImpl::onHeaders(ResponseHeaderMapPtr&& headers, bool) {

  const uint64_t response_code = Http::Utility::getResponseStatus(*headers);

  streamInfo().setResponseCode(response_code);

  response_ = std::make_unique<ResponseMessageImpl>(std::move(headers));

}

void AsyncRequestSharedImpl::onData(Buffer::Instance& data, bool) {

  if (discard_response_body_) {

    data.drain(data.length());

    return;

  }

  if (response_->body().length() + data.length() > response_buffer_limit_) {

    ENVOY_LOG_EVERY_POW_2(warn, "the buffer size limit for async client response body "

                                "has been exceeded, draining data");

    data.drain(data.length());

    response_buffer_overlimit_ = true;

    reset();

  } else {

    response_->body().move(data);

  }

}

void AsyncRequestSharedImpl::onTrailers(ResponseTrailerMapPtr&& trailers) {

  response_->trailers(std::move(trailers));

}

void AsyncRequestSharedImpl::onReset() {

  if (complete_) {

    return;

  }

  if (!cancelled_) {

    child_span_->setTag(Tracing::Tags::get().ErrorReason, "Reset");

  }

  callbacks_.onBeforeFinalizeUpstreamSpan(*child_span_,

                                          remoteClosed() ? &response_->headers() : nullptr);

  Tracing::HttpTracerUtility::finalizeUpstreamSpan(*child_span_, streamInfo(),

                                                   Tracing::EgressConfig::get());

  if (!cancelled_) {

    if (response_buffer_overlimit_) {

      callbacks_.onFailure(*this, AsyncClient::FailureReason::ExceedResponseBufferLimit);

    } else {

      callbacks_.onFailure(*this, AsyncClient::FailureReason::Reset);

    }

  }

}

void AsyncRequestSharedImpl::cancel() {

  cancelled_ = true;

  child_span_->setTag(Tracing::Tags::get().Canceled, Tracing::Tags::get().True);

  reset();

}