// 文件位置: envoy/network/client_connection_factory.hclassClientConnectionFactory{public:  virtual ~ClientConnectionFactory() = default;  virtual Network::ConnectionHandlePtr  createConnection(Event::Dispatcher& dispatcher,                    const Address::InstanceConstSharedPtr& address,                   Address::InstanceConstSharedPtr& source_address,                   TransportSocketPtr&& transport_socket,                   const StreamInfo::StreamInfo& stream_info,                   Network::TransportSocketOptionsConstSharedPtr transport_socket_options) PURE;};

// 文件位置: envoy/registry/registry.htemplate <class Base> class FactoryRegistry {public:  staticvoidregisterFactory(Base& factory, absl::string_view name);  static Base* getFactory(absl::string_view name);  static absl::flat_hash_map<std::string, Base*>& factories();private:  static absl::flat_hash_map<std::string, Base*>* factories_;};// 自动注册模板类template <class T, class Base> class RegisterFactory {public:  RegisterFactory() {    FactoryRegistry<Base>::registerFactory(instance_, instance_.name());  }private:  T instance_{};};

// 1. 定义工厂基类class NamedNetworkFilterConfigFactory : public Config::TypedFactory {public:  ~NamedNetworkFilterConfigFactory() override = default;  std::string category()constoverride{ return "envoy.network.filters"; }};// 2. 实现具体工厂class MyFilterFactory : public NamedNetworkFilterConfigFactory {public:  std::string name()constoverride{ return "my_filter"; }  Network::FilterFactoryCb createFilterFactoryFromProto(      const Protobuf::Message& proto_config,      Server::Configuration::FactoryContext& context) override {    return [context](Network::FilterManager& filter_manager) {      filter_manager.addFilter(std::make_shared<MyFilter>(context));    };  }  ProtobufTypes::MessagePtr createEmptyConfigProto()override{    return std::make_unique<MyFilterConfig>();  }};// 3. 静态注册（编译时自动注册）REGISTER_FACTORY(MyFilterFactory, NamedNetworkFilterConfigFactory);

编译时: REGISTER_FACTORY 宏创建静态对象   ↓静态初始化: RegisterFactory 构造函数调用 registerFactory   ↓运行时: 配置加载时调用 getFactory(name) 获取工厂   ↓使用: 工厂创建具体的过滤器实例

// 文件位置: envoy/http/filter.husing FilterFactoryCb = std::function<void(FilterChainFactoryCallbacks& callbacks)>;class FilterChainFactory {public:  virtual ~FilterChainFactory() = default;  virtualvoidcreateFilterChain(FilterChainManager& manager) PURE;  virtualboolcreateUpgradeFilterChain(absl::string_view upgrade,                                        const UpgradeMap* per_route_upgrade_map,                                        FilterChainManager& manager) PURE;};

// 文件位置: envoy/singleton/manager.hclass Manager {public:  virtual ~Manager() = default;  template <class T>  std::shared_ptr<T> getTyped(const std::string& name, SingletonFactoryCb cb){    return std::dynamic_pointer_cast<T>(get(name, cb));  }  virtual InstanceSharedPtr get(const std::string& name, SingletonFactoryCb cb) PURE;};// 单例注册宏#define SINGLETON_MANAGER_REGISTRATION(NAME)                                                       \  static constexpr char NAME##_singleton_name[] = #NAME "_singleton";                              \  static Envoy::Registry::RegisterInternalFactory<                                                 \      Envoy::Singleton::RegistrationImpl<NAME##_singleton_name>, Envoy::Singleton::Registration>   \      NAME##_singleton_registered_;

// 1. 注册单例类型SINGLETON_MANAGER_REGISTRATION(my_singleton);// 2. 获取单例实例auto singleton = singleton_manager.getTyped<MySingleton>(    SINGLETON_MANAGER_REGISTERED_NAME(my_singleton),    []() { return std::make_shared<MySingleton>(); });

Server 启动   ↓SingletonManager 初始化   ↓首次 getTyped() 调用 → 执行工厂回调创建实例   ↓后续调用返回缓存的实例   ↓Server 关闭 → 单例随 Server 销毁

// 示例：监听器建造者class ListenerBuilder {public:  ListenerBuilder& withName(absl::string_view name);  ListenerBuilder& withAddress(Address::InstanceConstSharedPtr address);  ListenerBuilder& withFilterChain(FilterChainPtr filter_chain);  ListenerBuilder& withSocketOptions(SocketOptions options);  ListenerBuilder& withConnectionLimit(uint32_t limit);  ListenerPtr build();};// 使用示例auto listener = ListenerBuilder()    .withName("http_listener")    .withAddress(address)    .withFilterChain(http_filter_chain)    .withConnectionLimit(10000)    .build();

// 将 HTTP 过滤器适配为网络过滤器class NetworkFilterAdapter : public Network::ReadFilter {public:  explicitNetworkFilterAdapter(Http::StreamFilterSharedPtr http_filter)      : http_filter_(http_filter) {}  Network::FilterStatus onData(Buffer::Instance& data, bool end_stream)override{    // 将网络数据适配为 HTTP 流数据    http_filter_->decodeData(data, end_stream);    return Network::FilterStatus::Continue;  }  Network::FilterStatus onNewConnection()override{    // 适配为 HTTP 流开始    return Network::FilterStatus::Continue;  }private:  Http::StreamFilterSharedPtr http_filter_;};

// 每个过滤器装饰处理逻辑class BufferFilter : public Http::StreamDecoderFilter {public:  FilterHeadersStatus decodeHeaders(RequestHeaderMap& headers, bool end_stream)override{    // 装饰：添加缓冲逻辑    return FilterHeadersStatus::Continue;  }  FilterDataStatus decodeData(Buffer::Instance& data, bool end_stream)override{    // 装饰：缓冲数据而不是立即转发    buffer_.add(data);    if (buffer_.length() >= min_buffer_size_ || end_stream) {      return FilterDataStatus::Continue;    }    return FilterDataStatus::StopIterationAndBuffer;  }private:  Buffer::OwnedImpl buffer_;  uint32_t min_buffer_size_;};

// 上游请求代理class UpstreamRequest : public Http::StreamCallbacks {public:  voidencodeHeaders(RequestHeaderMap& headers, bool end_stream)override{    // 代理：将请求头转发到上游连接    if (upstream_stream_) {      upstream_stream_->encodeHeaders(headers, end_stream);    }  }  voidencodeData(Buffer::Instance& data, bool end_stream)override{    // 代理：将数据转发到上游    if (upstream_stream_) {      upstream_stream_->encodeData(data, end_stream);    }  }private:  Http::StreamEncoder* upstream_stream_{nullptr};};

// 抽象：传输 Socket 接口class TransportSocket {public:  virtual ~TransportSocket() = default;  virtualvoidsetTransportSocketCallbacks(TransportSocketCallbacks& callbacks) PURE;  virtualvoiddoRead(Buffer::Instance& read_buffer) PURE;  virtualvoiddoWrite(Buffer::Instance& write_buffer, bool end_stream) PURE;  virtualvoidclose() PURE;};// 实现 A：普通 TCP Socketclass TcpSocket : public TransportSocket {public:  voiddoRead(Buffer::Instance& read_buffer) override {    // TCP 实现：从 socket fd 读取    ssize_t len = recv(fd_, buffer, size, 0);    read_buffer.add(buffer, len);  }};// 实现 B：TLS Socketclass SslSocket : public TransportSocket {public:  voiddoRead(Buffer::Instance& read_buffer) override {    // TLS 实现：通过 SSL_read 解密读取    int ret = SSL_read(ssl_, buffer, size);    read_buffer.add(buffer, ret);  }};

class FilterManager {public:  voidaddDecoderFilter(StreamDecoderFilterSharedPtr filter){    decoder_filters_.push_back(std::make_shared<ActiveDecoderFilter>(filter));  }  voidaddEncoderFilter(StreamEncoderFilterSharedPtr filter){    encoder_filters_.push_back(std::make_shared<ActiveEncoderFilter>(filter));  }  // 组合操作：递归调用所有过滤器  FilterHeadersStatus decodeHeaders(RequestHeaderMap& headers, bool end_stream){    for (auto& filter : decoder_filters_) {      auto status = filter->decodeHeaders(headers, end_stream);      if (status != FilterHeadersStatus::Continue) {        return status;      }    }    return FilterHeadersStatus::Continue;  }private:  std::vector<ActiveDecoderFilterPtr> decoder_filters_;  std::vector<ActiveEncoderFilterPtr> encoder_filters_;};

// 文件位置: envoy/server/lifecycle_notifier.hclass ServerLifecycleNotifier {public:  enum class Stage {     Startup,      // 服务启动阶段    PostInit,     // 初始化后阶段    ShutdownExit  // 关闭退出阶段  };  using StageCallback = std::function<void()>;  using HandlePtr = std::unique_ptr<Handle>;  virtual HandlePtr registerCallback(Stage stage, StageCallback callback) PURE;};// 连接事件观察者class ConnectionCallbacks {public:  virtual ~ConnectionCallbacks() = default;  virtualvoidonEvent(ConnectionEvent event) PURE;  virtualvoidonAboveWriteBufferHighWatermark() PURE;  virtualvoidonBelowWriteBufferLowWatermark() PURE;};

// 注册服务生命周期观察者class MyComponent {public:  voidregisterLifecycleNotifications(ServerLifecycleNotifier& notifier){    startup_handle_ = notifier.registerCallback(        ServerLifecycleNotifier::Stage::Startup,        [this]() { this->onServerStartup(); });    shutdown_handle_ = notifier.registerCallback(        ServerLifecycleNotifier::Stage::ShutdownExit,        [this]() { this->onServerShutdown(); });  }private:  voidonServerStartup(){ /* 启动逻辑 */ }  voidonServerShutdown(){ /* 清理逻辑 */ }  ServerLifecycleNotifier::HandlePtr startup_handle_;  ServerLifecycleNotifier::HandlePtr shutdown_handle_;};

// 文件位置: envoy/common/backoff_strategy.hclass BackOffStrategy {public:  virtual ~BackOffStrategy() = default;  virtualuint64_tnextBackOffMs() PURE;  virtualvoidreset() PURE;};// 具体策略 1：指数退避class ExponentialBackOffStrategy : public BackOffStrategy {public:  ExponentialBackOffStrategy(uint64_t base_interval_ms, uint64_t max_interval_ms)      : base_interval_ms_(base_interval_ms),        max_interval_ms_(max_interval_ms),        current_interval_ms_(base_interval_ms) {}  uint64_tnextBackOffMs()override{    uint64_t backoff = current_interval_ms_;    current_interval_ms_ = std::min(current_interval_ms_ * 2, max_interval_ms_);    return backoff;  }  voidreset()override{    current_interval_ms_ = base_interval_ms_;  }private:  const uint64_t base_interval_ms_;  const uint64_t max_interval_ms_;  uint64_t current_interval_ms_;};// 具体策略 2：固定间隔退避class FixedBackOffStrategy : public BackOffStrategy {public:  explicitFixedBackOffStrategy(uint64_t interval_ms) : interval_ms_(interval_ms) {}  uint64_tnextBackOffMs()override{ return interval_ms_; }  voidreset()override{}private:  const uint64_t interval_ms_;};

// 负载均衡策略接口class LoadBalancer {public:  virtual HostConstSharedPtr chooseHost(const HostSet& hosts) PURE;};// 轮询策略class RoundRobinLoadBalancer : public LoadBalancer { /* ... */ };// 最少连接策略class LeastRequestLoadBalancer : public LoadBalancer { /* ... */ };// 一致性哈希策略class ConsistentHashLoadBalancer : public LoadBalancer { /* ... */ };

// 文件位置: envoy/http/filter.henum class FilterHeadersStatus {  Continue,                    // 继续执行链中的下一个过滤器  StopIteration,               // 停止执行，不再传递给后续过滤器  StopAllIterationAndBuffer,   // 停止并缓冲数据};class StreamDecoderFilter {public:  virtual FilterHeadersStatus decodeHeaders(RequestHeaderMap& headers, bool end_stream) PURE;  virtual FilterDataStatus decodeData(Buffer::Instance& data, bool end_stream) PURE;  virtual FilterTrailersStatus decodeTrailers(RequestTrailerMap& trailers) PURE;};// 过滤器链处理class FilterChain {public:  voiddecodeHeaders(RequestHeaderMap& headers, bool end_stream){    for (auto& filter : filters_) {      auto status = filter->decodeHeaders(headers, end_stream);      switch (status) {        case FilterHeadersStatus::Continue:          continue;  // 继续传递给下一个过滤器        case FilterHeadersStatus::StopIteration:          return;    // 停止链的执行        case FilterHeadersStatus::StopAllIterationAndBuffer:          buffering_ = true;          return;      }    }  }private:  std::vector<StreamDecoderFilterPtr> filters_;  bool buffering_{false};};

请求 → Filter1.decodeHeaders() → Filter2.decodeHeaders() → ... → FilterN.decodeHeaders()         │                            │         ├─ Continue ─────────────────┤         ├─ StopIteration → 停止转发         └─ StopAndBuffer → 缓冲后异步处理

// 命令接口using PostCb = std::function<void()>;class Dispatcher {public:  virtualvoidpost(PostCb callback) PURE;  virtualvoiddefer(PostCb callback) PURE;};// 使用示例class AsyncOperation {public:  voidexecute(Dispatcher& dispatcher){    dispatcher.post([this]() {      // 在目标线程执行的操作      this->doWork();    });  }private:  voiddoWork(){ /* 实际工作 */ }};

// 连接状态机classConnectionImpl{public:  enumclassState{    Connecting,   // 连接中    Connected,    // 已连接    Draining,     // 排空中    Closed        // 已关闭  };  void onConnected() {    if(state_ == State::Connecting) {      state_ = State::Connected;      notifyConnected();    }  }  void onRemoteClose() {    if(state_ == State::Connected || state_ == State::Draining) {      state_ = State::Closed;      notifyRemoteClosed();    }  }  void close() {    if(state_ == State::Connecting || state_ == State::Connected) {      state_ = State::Draining;      initiateGracefulClose();    }  }private:  State state_{State::Connecting};};

// 基类定义模板方法class ConnectionImplBase {public:  // 模板方法：定义了关闭连接的算法骨架  voidclose(ConnectionCloseType type){    // 1. 前置处理    onPreClose();    // 2. 执行具体关闭（由子类实现）    doClose(type);    // 3. 清理资源    releaseResources();    // 4. 后置处理    onPostClose();  }protected:  virtualvoidonPreClose(){}                    // 钩子方法（可选覆盖）  virtualvoiddoClose(ConnectionCloseType type) PURE;  // 抽象方法（必须实现）  virtualvoidonPostClose(){}                   // 钩子方法（可选覆盖）private:  voidreleaseResources(){    // 通用清理逻辑    read_buffer_.drain(read_buffer_.length());    write_buffer_.drain(write_buffer_.length());  }  Buffer::Instance read_buffer_;  Buffer::Instance write_buffer_;};

// 文件位置: envoy/protobuf/message_validator.hclass ValidationVisitor {public:  virtual ~ValidationVisitor() = default;  virtualvoidonUnknownField(absl::string_view description) PURE;  virtualvoidonDeprecatedField(absl::string_view description, bool soft_deprecation) PURE;  virtualvoidonWorkInProgress(absl::string_view description) PURE;};// 严格模式访问者class StrictValidationVisitor : public ValidationVisitor {public:  voidonUnknownField(absl::string_view description)override{    throw EnvoyException(fmt::format("Unknown field: {}", description));  }  voidonDeprecatedField(absl::string_view description, bool soft_deprecation)override{    if (!soft_deprecation) {      throw EnvoyException(fmt::format("Deprecated field: {}", description));    }  }  voidonWorkInProgress(absl::string_view description)override{    throw EnvoyException(fmt::format("WIP feature: {}", description));  }};// 宽松模式访问者class LenientValidationVisitor : public ValidationVisitor {public:  voidonUnknownField(absl::string_view description)override{    ENVOY_LOG(warn, "Unknown field: {}", description);  }  voidonDeprecatedField(absl::string_view description, bool soft_deprecation)override{    ENVOY_LOG(warn, "Deprecated field: {}", description);  }  voidonWorkInProgress(absl::string_view description)override{    ENVOY_LOG(info, "WIP feature: {}", description);  }};

class ClusterManager {public:  virtual ~ClusterManager() = default;  // 集群管理  virtual Upstream::ClusterInfoConstSharedPtr getCluster(absl::string_view cluster) PURE;  virtual ThreadLocalCluster* getThreadLocalCluster(absl::string_view cluster) PURE;  virtualbooladdOrUpdateCluster(const envoy::config::cluster::v3::Cluster& cluster,                                  const std::string& version_info) PURE;  // HTTP 连接池管理  virtual Http::ConnectionPool::Instance* httpConnPoolForCluster(absl::string_view cluster) PURE;  // TCP 连接池管理  virtual Tcp::ConnectionPool::Instance* tcpConnPoolForCluster(absl::string_view cluster) PURE;  // CDS 回调  virtualvoidsetPrimaryClustersInitializedCb(std::function<void()> callback) PURE;};

// 文件位置: envoy/thread_local/thread_local.hclass ThreadLocalObject {public:  virtual ~ThreadLocalObject() = default;};class Slot {public:  virtual ~Slot() = default;  virtualvoidset(std::function<std::shared_ptr<ThreadLocalObject>(Event::Dispatcher&)> create_cb) PURE;  virtual std::shared_ptr<ThreadLocalObject> get() PURE;};class Instance {public:  virtual ~Instance() = default;  virtual SlotPtr allocateSlot() PURE;  template <class T>  std::shared_ptr<T> getTyped(Slot& slot){    return std::dynamic_pointer_cast<T>(slot.get());  }};// 使用示例class MyThreadLocalData : public ThreadLocalObject {public:  int counter_{0};  std::vector<ConnectionPtr> connections_;};// 分配槽位并设置 TLS 对象auto slot = tls.allocateSlot();slot->set([](Event::Dispatcher& dispatcher) {  return std::make_shared<MyThreadLocalData>();});// 在任一工作线程中访问auto data = tls.getTyped<MyThreadLocalData>(*slot);data->counter_++;  // 无需加锁，每个线程独立

// 事件调度器（反应器）class Dispatcher {public:  virtual ~Dispatcher() = default;  // 运行事件循环  virtualvoidrun(RunType type) PURE;  // 创建定时器  virtual TimerPtr createTimer(const TimerCb& callback) PURE;  // 延迟执行回调  virtualvoidpost(PostCb callback) PURE;  // 创建文件事件监听  virtual FileEventPtr createFileEvent(int fd, FileReadyCb cb, int events,                                       FileTriggerType trigger_type) PURE;};// 定时器实现class Timer {public:  virtualvoidenableTimer(const std::chrono::milliseconds& timeout) PURE;  virtualvoiddisableTimer() PURE;};// 使用示例auto timer = dispatcher->createTimer([this]() {  this->onTimeout();});timer->enableTimer(std::chrono::milliseconds(1000));

// 异步连接操作class Connection {public:  virtualvoidconnect() PURE;  virtualvoidclose() PURE;  virtualvoidaddConnectionCallbacks(ConnectionCallbacks& callbacks) PURE;};// 异步操作完成时的回调（主动器）class ConnectionCallbacks {public:  virtualvoidonEvent(ConnectionEvent event) PURE;  virtualvoidonAboveWriteBufferHighWatermark() PURE;  virtualvoidonBelowWriteBufferLowWatermark() PURE;};// 异步连接示例class AsyncConnector {public:  voidconnect(Dispatcher& dispatcher, const Address::InstanceConstSharedPtr& address){    auto connection = dispatcher.createClientConnection(address, ...);    connection->addConnectionCallbacks(*this);    connection->connect();  // 异步连接，立即返回    // 连接完成后 onEvent(Connected) 会被调用  }  voidonEvent(ConnectionEvent event)override{    if (event == ConnectionEvent::Connected) {      // 连接成功，开始发送数据      sendRequest();    }  }};

// 插件基类class NamedNetworkFilterConfigFactory : public Config::TypedFactory {public:  std::string category()constoverride{ return "envoy.network.filters"; }};// 插件自动发现与注册// 在编译时，所有 REGISTER_FACTORY 宏都会创建静态注册对象// 运行时，配置加载器会查询 FactoryRegistry 获取对应的工厂// 配置加载流程voidConfiguration::loadConfiguration(const std::string& config){  // 1. 解析配置  auto proto_config = parseConfig(config);  // 2. 根据类型查找工厂  auto* factory = FactoryRegistry<NamedNetworkFilterConfigFactory>::getFactory(      proto_config.name());  // 3. 使用工厂创建插件实例  auto filter = factory->createFilterFactoryFromProto(proto_config, context);}

// 事件循环的主循环void DispatcherImpl::run(RunTypetype) {  running_ = true;  while (running_) {    // 1. 等待事件    event_base_loop(event_base_, EVLOOP_ONCE | ...);    // 2. 处理就绪的事件    processReadyEvents();    // 3. 执行延迟回调    drainPostCallbacks();    // 4. 检查退出条件    if (type == RunType::NonBlock && !hasPendingEvents()) {      break;    }  }}// 事件驱动流程// I/O事件 → 事件循环 → 回调处理 → 产生新事件 → 继续循环

1. 配置加载   └── 工厂模式：创建过滤器实例2. 过滤器链构建   └── 建造者模式：构建过滤器链   └── 组合模式：组织过滤器结构3. 请求处理   └── 责任链模式：依次调用过滤器   └── 装饰器模式：每个过滤器添加功能4. 异步处理   └── 命令模式：封装回调   └── 观察者模式：事件通知

// 确保使用正确的 categoryclass MyFilterFactory : public NamedNetworkFilterConfigFactory {public:  std::string category() const override {     return "envoy.network.filters";  // 必须与扩展点匹配  }};// 使用正确的注册宏REGISTER_FACTORY(MyFilterFactory, NamedNetworkFilterConfigFactory);

// 无状态过滤器：可以在多个线程间共享classStatelessFilter : publicHttp::StreamDecoderFilter{  // 不使用成员变量存储请求特定数据};// 有状态过滤器：需要使用 TLS 或确保线程安全classStatefulFilter : publicHttp::StreamDecoderFilter{  // 使用 TLS 存储每线程数据  ThreadLocal::SlotPtr tls_slot_;};

class MyFilter : public Http::StreamDecoderFilter {public:  // 在 onDestroy 中清理资源  voidonDestroy()override{    cancelPendingOperations();    releaseResources();  }private:  Cancellable* pending_operation_{nullptr};};