Envoy:httpfilter相关代码阅读

Network::FilterFactoryCb
HttpConnectionManagerFilterConfigFactory::createFilterFactoryFromProtoTyped()
---→
std::shared_ptr<HttpConnectionManagerConfig>  Utility::createConfig()
---→ 
std::make_shared<HttpConnectionManagerConfig>() { 
  ......
  // 操作的是http_filters
  const auto& filters = config.http_filters(); 
  DependencyManager dependency_manager;
  for (int32_t i = 0; i < filters.size(); i++) {
     processFilter(filters[i], i, "http", "http", i == filters.size() - 1, filter_factories_,
           dependency_manager);
  }
  ......
}
----→ 
void HttpConnectionManagerConfig::processFilter() {
  ......
  auto* factory =
  Config::Utility::getAndCheckFactory<Server::Configuration::NamedHttpFilterConfigFactory>(
    proto_config, proto_config.is_optional());
  ProtobufTypes::MessagePtr message = Config::Utility::translateToFactoryConfig(
     proto_config, context_.messageValidationVisitor(), *factory);
  Http::FilterFactoryCb callback =
  factory->createFilterFactoryFromProto(*message, stats_prefix_, context_);
  ......
  // 这里将filterfactorycb存放到filter_factories中
  filter_factories.push_back(std::move(filter_config_provider));
}

ConnectionImpl::onFileEvent()-→
.......
------>
Envoy::StatusOr<ParserStatus> ServerConnectionImpl::onHeadersCompleteBase() {
......
  active_request.request_decoder_->decodeHeaders(std::move(headers), false);
......
}
----→ 
void ConnectionManagerImpl::ActiveStream::decodeHeaders(RequestHeaderMapPtr&& headers, bool end_stream) {
......
  const bool upgrade_rejected = filter_manager_.createFilterChain() == false;
......
}
---→ 
bool FilterManager::createFilterChain() {
......
  if (upgrade != nullptr) {
    const Router::RouteEntry::UpgradeMap* upgrade_map = filter_manager_callbacks_.upgradeMap();
    if (filter_chain_factory_.createUpgradeFilterChain(upgrade->value().getStringView(),
        upgrade_map, *this)) {
        filter_manager_callbacks_.upgradeFilterChainCreated();
        return true;
    } else {
        upgrade_rejected = true;
        // Fall through to the default filter chain. The function calling this
        // will send a local reply indicating that the upgrade failed.
   }
 }
filter_chain_factory_.createFilterChain(*this);
......
}
----→ 
HttpConnectionManagerConfig::createFilterChain()
HttpConnectionManagerConfig::createUpgradeFilterChain() 
----→ 
HttpConnectionManagerConfig::createFilterChainForFactories() {
 ......
   for (const auto& filter_config_provider : filter_factories) {
      auto config = filter_config_provider->config();
      if (config.has_value()) {
       // 这里对应的是http_filter 创建的工厂调用函数里面的FilterFactoryCb函数
          config.value()(callbacks);
          continue;
       }
......
}
-----→
以BandwidthLimitFilterConfig 为例子，config.value()(callbacks) 对应的是:
[filter_config](Http::FilterChainFactoryCallbacks& callbacks) -> void { // 这里的callbacks对应的是fiter_manager
      callbacks.addStreamFilter(std::make_shared<BandwidthLimiter>(filter_config))
} //callback对应的是fiter_manager，所以这里调用的是下面的函数：
----→ 
void addStreamFilter(StreamFilterSharedPtr filter) override {
    addStreamDecoderFilterWorker(filter, nullptr, true);
    addStreamEncoderFilterWorker(filter, nullptr, true);
    StreamDecoderFilter* decoder_filter = filter.get();
    filters_.push_back(decoder_filter);
}
---→
LinkedList::moveIntoListBack(std::move(wrapper), decoder_filters_); 或者
LinkedList::moveIntoList(std::move(wrapper), encoder_filters_);或者同时添加进去。

ParserStatus ServerConnectionImpl::onMessageCompleteBase() 
ParserStatus ClientConnectionImpl::onMessageCompleteBase()
----→ 
void ConnectionManagerImpl::ActiveStream::encodeHeaders(ResponseHeaderMap& response_headers, bool end_stream) 
void encodeData(Buffer::Instance& data, bool end_stream) 
void encodeTrailers(ResponseTrailerMap& trailers)
----→ 
void FilterManager::encodeHeaders(ActiveStreamEncoderFilter* filter, ResponseHeaderMap& headers, bool end_stream)  
void encodeData(Buffer::Instance& data, bool end_stream)
void encodeTrailers(ResponseTrailerMapPtr&& trailers)
----→ 会去遍历encoder_filters去依次执行对应的
encodeHeaders()\encodeData()\encodeTrailers()函数

ParserStatus ServerConnectionImpl::onMessageCompleteBase() 
ParserStatus ClientConnectionImpl::onMessageCompleteBase()
----→  
void ConnectionManagerImpl::ActiveStream::decodeHeaders(RequestHeaderMapPtr&& headers,
      bool end_stream)
void decodeData(Buffer::Instance& data, bool end_stream)
void decodeTrailers(RequestTrailerMapPtr&& trailers)
---→ 
void FilterManager::decodeHeaders(ActiveStreamDecoderFilter* filter, RequestHeaderMap& headers,
bool end_stream) 
void FilterManager::decodeData(ActiveStreamDecoderFilter* filter, Buffer::Instance& data,
bool end_stream,
FilterIterationStartState filter_iteration_start_state)
void FilterManager::decodeTrailers(ActiveStreamDecoderFilter* filter, RequestTrailerMap& trailers)
--→ 会去遍历decoder_filters去依次执行对应的
decodeHeaders()\decodeData()\decodeTrailers()函数

ParserStatus ServerConnectionImpl::onMessageCompleteBase() {
  ASSERT(!handling_upgrade_);
  if (active_request_.has_value()) {
    auto& active_request = active_request_.value();
    if (active_request.request_decoder_) {
      active_request.response_encoder_.readDisable(true);
    }
    active_request.remote_complete_ = true;
    if (deferred_end_stream_headers_) {
      active_request.request_decoder_->decodeHeaders(
          std::move(absl::get<RequestHeaderMapPtr>(headers_or_trailers_)), true);
      deferred_end_stream_headers_ = false;
    } else if (processing_trailers_) {
      active_request.request_decoder_->decodeTrailers(
          std::move(absl::get<RequestTrailerMapPtr>(headers_or_trailers_)));
    } else {
      Buffer::OwnedImpl buffer;
      active_request.request_decoder_->decodeData(buffer, true);
    }
    // Reset to ensure no information from one requests persists to the next.
    headers_or_trailers_.emplace<RequestHeaderMapPtr>(nullptr);
  }
  // Always pause the parser so that the calling code can process 1 request at a time and apply
  // back pressure. However this means that the calling code needs to detect if there is more data
  // in the buffer and dispatch it again.
  return parser_->pause();
}
  
ParserStatus ClientConnectionImpl::onMessageCompleteBase() {
  ENVOY_CONN_LOG(trace, "message complete", connection_);
  if (ignore_message_complete_for_1xx_) {
    ignore_message_complete_for_1xx_ = false;
    return ParserStatus::Success;
  }
  if (pending_response_.has_value()) {
    ASSERT(!pending_response_done_);
    // After calling decodeData() with end stream set to true, we should no longer be able to reset.
    PendingResponse& response = pending_response_.value();
    // Encoder is used as part of decode* calls later in this function so pending_response_ can not
    // be reset just yet. Preserve the state in pending_response_done_ instead.
    pending_response_done_ = true;
  
    if (deferred_end_stream_headers_) {
      response.decoder_->decodeHeaders(
          std::move(absl::get<ResponseHeaderMapPtr>(headers_or_trailers_)), true);
      deferred_end_stream_headers_ = false;
    } else if (processing_trailers_) {
      response.decoder_->decodeTrailers(
          std::move(absl::get<ResponseTrailerMapPtr>(headers_or_trailers_)));
    } else {
      Buffer::OwnedImpl buffer;
      response.decoder_->decodeData(buffer, true);
    }
  
    // Reset to ensure no information from one requests persists to the next.
    pending_response_.reset();
    headers_or_trailers_.emplace<ResponseHeaderMapPtr>(nullptr);
  }
  // Pause the parser after a response is complete. Any remaining data indicates an error.
  return parser_->pause();
}