class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,                              public Network::ReadFilter,                              public ServerConnectionCallbacks,                              public Network::ConnectionCallbacks,                              public Http::ApiListener {public:  ConnectionManagerImpl(ConnectionManagerConfig& config,                         const Network::DrainDecision& drain_close,                        Random::RandomGenerator& random_generator,                         Http::Context& http_context,                        Runtime::Loader& runtime,                         const LocalInfo::LocalInfo& local_info,                        Upstream::ClusterManager& cluster_manager,                        Server::OverloadManager& overload_manager,                         TimeSource& time_system);  ~ConnectionManagerImpl() override;  // Network::ReadFilter  Network::FilterStatus onData(Buffer::Instance& data, bool end_stream) override;  Network::FilterStatus onNewConnection() override;  void initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callbacks) override;  // Http::ServerConnectionCallbacks  RequestDecoder& newStream(ResponseEncoder& response_encoder,                            bool is_internally_created = false) override;  // Network::ConnectionCallbacks  void onEvent(Network::ConnectionEvent event) override;  void onAboveWriteBufferHighWatermark() override;  void onBelowWriteBufferLowWatermark() override;};

// HTTP 数据处理流程Network::FilterStatus ConnectionManagerImpl::onData(Buffer::Instance& data, bool end_stream) {  // 延迟创建编解码器，直到有数据到达  if(!codec_) {    createCodec(data);  }  // 将数据传递给编解码器处理  codec_->dispatch(data);  // 继续处理下一个网络过滤器（如果有）  return Network::FilterStatus::Continue;}// 编解码器创建流程void ConnectionManagerImpl::createCodec(Buffer::Instance& data) {  // 根据数据判断协议版本  if(isHttp2(data)) {    codec_ = createHttp2Codec();  } else if (isHttp1(data)) {    codec_ = createHttp1Codec();  } else {    handleCodecError("Unsupported protocol");    return;  }  codec_->initialize();}

class CodecClient : protected Logger::Loggable<Logger::Id::client>,                    public Http::ConnectionCallbacks,                    public Network::ConnectionCallbacks,                    public Event::DeferredDeletable {public:  CodecClient(CodecType type, Network::ClientConnectionPtr&& connection,              Upstream::HostDescriptionConstSharedPtr host, Event::Dispatcher& dispatcher);  voidconnect();  RequestEncoder& newStream(ResponseDecoder& response_decoder);  voidclose();  Protocol protocol(){ return codec_->protocol(); }  size_tnumActiveRequests(){ return active_requests_.size(); }  // Http::ConnectionCallbacks  voidonGoAway(GoAwayErrorCode error_code)override;  voidonSettings(ReceivedSettings& settings)override;  voidonMaxStreamsChanged(uint32_t num_streams)override;  // Network::ConnectionCallbacks  voidonEvent(Network::ConnectionEvent event)override;  voidonAboveWriteBufferHighWatermark()override;  voidonBelowWriteBufferLowWatermark()override;};

// HTTP 流创建流程RequestEncoder& CodecClient::newStream(ResponseDecoder& response_decoder){  auto active_request = std::make_unique<ActiveRequest>(*this, response_decoder);  RequestEncoder& encoder = codec_->newStream(response_decoder);  active_request->encoder_ = &encoder;  active_requests_.push_back(std::move(active_request));  return encoder;}// 网络事件处理voidCodecClient::onEvent(Network::ConnectionEvent event){  switch (event) {    case Network::ConnectionEvent::Connected:      codec_->initialize();      break;    case Network::ConnectionEvent::RemoteClose:    case Network::ConnectionEvent::LocalClose:      remote_closed_ = true;      resetAllStreams();      break;  }}

// ConnectionManagerImpl 中的定时器使用void ConnectionManagerImpl::initializeTimers() {  // 连接空闲超时定时器  connection_idle_timer_ = dispatcher_.createTimer([this]() { this->onIdleTimeout(); });  connection_idle_timer_->enableTimer(config_.idle_timeout());  // 连接持续时间定时器  if (config_.connection_duration_timeout().has_value()) {    connection_duration_timer_ = dispatcher_.createTimer(        [this]() { this->onConnectionDurationTimeout(); });    connection_duration_timer_->enableTimer(config_.connection_duration_timeout().value());  }}// ActiveStream 中的定时器使用ActiveStream::ActiveStream(ConnectionManagerImpl& connection_manager, uint32_t buffer_limit)    : connection_manager_(connection_manager) {  // 请求超时定时器  request_timer_ = connection_manager_.dispatcher_.createTimer(      [this]() { this->onRequestTimeout(); });  request_timer_->enableTimer(config_.request_timeout());  // 请求头超时定时器  request_header_timer_ = connection_manager_.dispatcher_.createTimer(      [this]() { this->onRequestHeaderTimeout(); });  request_header_timer_->enableTimer(config_.request_header_timeout());  // 流空闲超时定时器  stream_idle_timer_ = connection_manager_.dispatcher_.createTimer(      [this]() { this->onIdleTimeout(); });  stream_idle_timer_->enableTimer(config_.stream_idle_timeout());}

// CodecClient 中的连接事件处理void CodecClient::onEvent(Network::ConnectionEvent event) {  switch(event) {    case Network::ConnectionEvent::Connected:      connected_ = true;      codec_->initialize();      break;    case Network::ConnectionEvent::RemoteClose:      remote_closed_ = true;      resetAllStreams(StreamResetReason::RemoteReset,                       "Connection closed by remote");      break;    case Network::ConnectionEvent::LocalClose:      resetAllStreams(StreamResetReason::LocalReset,                       "Connection closed locally");      break;    case Network::ConnectionEvent::ConnectError:      handleConnectionError();      break;  }}

http1_settings:  enable_trailing_headers: true  allow_chunked_length: true  enable_keep_alives: true  max_concurrent_streams: 100  initial_window_size: 65536

quic_options:  connection_timeout: 5s  max_concurrent_connections: 1000  idle_timeout: 300s  max_datagram_size: 1452

quic_options:  connection_timeout: 5s  max_concurrent_connections: 1000  idle_timeout: 300s  max_datagram_size: 1452

class ConnectionPoolBase : public std::enable_shared_from_this<ConnectionPoolBase> {public:  virtual ClientConnectionPtr connect()= 0;  virtualvoidaddConnectionCallback(ConnectionCallback& callback)= 0;  virtualvoidremoveConnectionCallback(ConnectionCallback& callback)= 0;};class ConnectionPoolImpl : public ConnectionPoolBase {public:  ClientConnectionPtr connect()override{    // 从连接池中获取或创建新连接    if (!idle_connections_.empty()) {      auto connection = std::move(idle_connections_.front());      idle_connections_.pop_front();      return connection;    }    if (active_connections_.size() < max_connections_) {      return createNewConnection();    }    // 连接池已满，等待空闲连接    return nullptr;  }private:  std::list<ClientConnectionPtr> idle_connections_;  std::unordered_set<ClientConnectionPtr> active_connections_;  size_t max_connections_;};

struct ActiveStream final : LinkedObject<ActiveStream>,                            public Event::DeferredDeletable,                            public StreamCallbacks,                            public RequestDecoder {  voidcompleteRequest(){    // 标记流完成    state_.codec_saw_local_complete_ = true;    // 清理定时器    disarmRequestTimeout();    resetIdleTimer();    // 统计信息更新    updateStats();    // 延迟删除流，避免在回调中删除对象    connection_manager_.doDeferredStreamDestroy(*this);  }  voidonIdleTimeout(){    resetStream(StreamResetReason::IdleTimeout);  }};

class FilterManager : public FilterManagerCallbacks {public:  FilterManager(ActiveStream& stream, ConnectionManagerImpl& connection_manager,                const ConnectionManagerConfig& config);  // 应用请求过滤器链  voidapplyRequestFilters(RequestHeaderMapPtr&& headers);  // 应用响应过滤器链  voidapplyResponseFilters(ResponseHeaderMapPtr&& headers);private:  std::vector<RequestDecoderFilterPtr> request_filters_;  std::vector<ResponseEncoderFilterPtr> response_filters_;  ActiveStream& stream_;  ConnectionManagerImpl& connection_manager_;};

http_filters:  # 认证过滤器  - name: envoy.filters.http.jwt_authn    typed_config:      "@type": type.googleapis.com/envoy.extensions.filters.http.jwt_authn.v3.JwtAuthentication      providers:        my_provider:          issuer: example.com          audiences:          - api.example.com          remote_jwks:            http_uri:              uri: https://example.com/.well-known/jwks.json              timeout: 5s            cache_duration: 300s  # 速率限制过滤器  - name: envoy.filters.http.rate_limit    typed_config:      "@type": type.googleapis.com/envoy.extensions.filters.http.rate_limit.v3.RateLimit      domain: example_domain      stage: 0  # 路由过滤器（必须最后）  - name: envoy.filters.http.router    typed_config:      "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router

static_resources:  listeners:  - name: http_listener    address:      socket_address: { address: 0.0.0.0, port_value: 8080 }    filter_chains:    - filters:      - name: envoy.filters.network.http_connection_manager        typed_config:          "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager          stat_prefix: ingress_http          codec_type: AUTO          route_config:            name: local_route            virtual_hosts:            - name: local_service              domains: ["*"]              routes:              - match: { prefix: "/" }                route: { cluster: "example_cluster" }          http_filters:          - name: envoy.filters.http.router  clusters:  - name: example_cluster    connect_timeout: 0.25s    type: STRICT_DNS    lb_policy: ROUND_ROBIN    http2_protocol_options: {}    load_assignment:      cluster_name: example_cluster      endpoints:      - lb_endpoints:        - endpoint:            address:              socket_address: { address: 127.0.0.1, port_value: 80 }

// ConnectionManagerImpl 构造函数ConnectionManagerImpl::ConnectionManagerImpl(ConnectionManagerConfig& config,                                             const Network::DrainDecision& drain_close,                                             Random::RandomGenerator& random_generator,                                             Http::Context& http_context, Runtime::Loader& runtime,                                             const LocalInfo::LocalInfo& local_info,                                             Upstream::ClusterManager& cluster_manager,                                             Server::OverloadManager& overload_manager,                                             TimeSource& time_source)    : config_(config), stats_(config_.stats()),      conn_length_(new Stats::HistogramCompletableTimespanImpl(          stats_.named_.downstream_cx_length_ms_, time_source)),      drain_close_(drain_close), user_agent_(http_context.userAgentContext()),      random_generator_(random_generator), http_context_(http_context), runtime_(runtime),      local_info_(local_info), cluster_manager_(cluster_manager),      listener_stats_(config_.listenerStats()),      overload_state_(overload_manager.getThreadLocalOverloadState()),      time_source_(time_source) {}// 初始化读取过滤器回调void ConnectionManagerImpl::initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callbacks) {  read_callbacks_ = &callbacks;  stats_.named_.downstream_cx_total_.inc();  stats_.named_.downstream_cx_active_.inc();  // 创建连接超时定时器  if (config_.idleTimeout()) {    connection_idle_timer_ = read_callbacks_->connection().dispatcher().createScaledTimer(        Event::ScaledTimerType::HttpDownstreamIdleConnectionTimeout,        [this]() -> void { onIdleTimeout(); });    connection_idle_timer_->enableTimer(config_.idleTimeout().value());  }}

RequestDecoder& ConnectionManagerImpl::newStream(ResponseEncoder& response_encoder,                                                 bool is_internally_created) {  if (connection_idle_timer_) {    connection_idle_timer_->disableTimer();  }  // 创建 ActiveStream 实例  auto& buffer_factory = read_callbacks_->connection().dispatcher().getWatermarkFactory();  Buffer::BufferMemoryAccountSharedPtr downstream_stream_account =      buffer_factory.createAccount(response_encoder.getStream());  response_encoder.getStream().setAccount(downstream_stream_account);  ActiveStreamPtr new_stream(new ActiveStream(*this, response_encoder.getStream().bufferLimit(),                                              std::move(downstream_stream_account)));  accumulated_requests_++;  if (config_.maxRequestsPerConnection() > 0 &&      accumulated_requests_ >= config_.maxRequestsPerConnection()) {    if (codec_->protocol() < Protocol::Http2) {      new_stream->state_.saw_connection_close_ = true;      drain_state_ = DrainState::Closing;    }  }  new_stream->response_encoder_ = &response_encoder;  new_stream->response_encoder_->getStream().addCallbacks(*new_stream);  LinkedList::moveIntoList(std::move(new_stream), streams_);  return **streams_.begin();}

void ConnectionManagerImpl::ActiveStream::decodeHeaders(RequestHeaderMapPtr&& headers,                                                        bool end_stream) {  request_headers_ = std::move(headers);  filter_manager_.requestHeadersInitialized();  // 路径规范化处理  const auto action =      ConnectionManagerUtility::maybeNormalizePath(*request_headers_, connection_manager_.config_);  if (action == ConnectionManagerUtility::NormalizePathAction::Reject) {    sendLocalReply(Code::BadRequest, "", nullptr, absl::nullopt,                   StreamInfo::ResponseCodeDetails::get().PathNormalizationFailed);    return;  }  // 应用过滤器链  filter_manager_.decodeHeaders(*request_headers_, end_stream);}

void ConnectionManagerImpl::ActiveStream::encodeHeaders(ResponseHeaderMap& headers,                                                        bool end_stream) {  // 设置 Date 头部  if (!headers.Date()) {    connection_manager_.config_.dateProvider().setDateHeader(headers);  }  // 设置 Server 头部  const auto transformation = connection_manager_.config_.serverHeaderTransformation();  if (transformation == ConnectionManagerConfig::HttpConnectionManagerProto::OVERWRITE ||      (transformation == ConnectionManagerConfig::HttpConnectionManagerProto::APPEND_IF_ABSENT &&       headers.Server() == nullptr)) {    headers.setReferenceServer(connection_manager_.config_.serverName());  }  chargeStats(headers);  response_encoder_->encodeHeaders(headers, end_stream);}void ConnectionManagerImpl::ActiveStream::encodeData(Buffer::Instance& data, bool end_stream) {  filter_manager_.streamInfo().addBytesSent(data.length());  response_encoder_->encodeData(data, end_stream);}

http_connection_manager:  # 连接管理配置  idle_timeout: 300s  max_connection_duration: 3600s  max_requests_per_connection: 100  delayed_close_timeout: 15s  # 超时配置  stream_idle_timeout: 60s  request_timeout: 30s  request_headers_timeout: 5s  max_stream_duration: 120s  # 协议配置  codec_type: AUTO  http1_settings:    accept_http_10: true    default_host_for_http_10: "example.com"  http2_settings:    initial_stream_window_size: 65536    initial_connection_window_size: 1048576    max_concurrent_streams: 256