#pragma once

#include <functional>

#include <memory>

#include "envoy/buffer/buffer.h"

#include "envoy/common/optref.h"

#include "envoy/extensions/filters/network/http_connection_manager/v3/http_connection_manager.pb.h"

#include "envoy/extensions/filters/network/http_connection_manager/v3/http_connection_manager.pb.validate.h"

#include "envoy/http/filter.h"

#include "envoy/http/header_map.h"

#include "envoy/matcher/matcher.h"

#include "envoy/network/socket.h"

#include "envoy/protobuf/message_validator.h"

#include "source/common/buffer/watermark_buffer.h"

#include "source/common/common/dump_state_utils.h"

#include "source/common/common/linked_object.h"

#include "source/common/common/logger.h"

#include "source/common/grpc/common.h"

#include "source/common/http/header_utility.h"

#include "source/common/http/headers.h"

#include "source/common/http/matching/data_impl.h"

#include "source/common/http/utility.h"

#include "source/common/local_reply/local_reply.h"

#include "source/common/matcher/matcher.h"

#include "source/common/protobuf/utility.h"

#include "source/common/runtime/runtime_features.h"

#include "source/common/stream_info/stream_info_impl.h"

namespace Envoy {

namespace Http {

class FilterManager;

class DownstreamFilterManager;

struct ActiveStreamFilterBase;

struct ActiveStreamDecoderFilter;

struct ActiveStreamEncoderFilter;

using ActiveStreamDecoderFilterPtr = std::unique_ptr<ActiveStreamDecoderFilter>;

using ActiveStreamEncoderFilterPtr = std::unique_ptr<ActiveStreamEncoderFilter>;

constexpr absl::string_view LocalReplyFilterStateKey =

    "envoy.filters.network.http_connection_manager.local_reply_owner";

class LocalReplyOwnerObject : public StreamInfo::FilterState::Object {

public:

  LocalReplyOwnerObject(const std::string& filter_config_name)

private:

  const std::string filter_config_name_;

};

// TODO(wbpcode): Rather than allocating every filter with an unique pointer, we could

// construct the filter in place in the vector. This should reduce the heap allocation and

// memory fragmentation.

// HTTP decoder filters. If filters are configured in the following order (assume all three

// filters are both decoder/encoder filters):

//   http_filters:

//     - A

//     - B

//     - C

// The decoder filter chain will iterate through filters A, B, C.

struct StreamDecoderFilters {

  using Element = ActiveStreamDecoderFilter;

  using Iterator = std::vector<ActiveStreamDecoderFilterPtr>::iterator;

  std::vector<ActiveStreamDecoderFilterPtr> entries_;

};

// HTTP encoder filters. If filters are configured in the following order (assume all three

// filters are both decoder/encoder filters):

//   http_filters:

//     - A

//     - B

//     - C

// Unlike the decoder filter, the encoder filter chain will iterate with the

// reverse order of the configured filters, i.e., C, B, A. This is why we use reverse_iterator

// here.

struct StreamEncoderFilters {

  using Element = ActiveStreamEncoderFilter;

  using Iterator = std::vector<ActiveStreamEncoderFilterPtr>::reverse_iterator;

  std::vector<ActiveStreamEncoderFilterPtr> entries_;

};

/**

 * Base class wrapper for both stream encoder and decoder filters.

 *

 * This class is responsible for performing matching and updating match data when a match tree is

 * configured for the associated filter. When not using a match tree, only minimal overhead (i.e.

 * memory overhead of unused fields) should apply.

 */

struct ActiveStreamFilterBase : public virtual StreamFilterCallbacks,

                                Logger::Loggable<Logger::Id::http> {

  ActiveStreamFilterBase(FilterManager& parent, absl::string_view filter_config_name)

  // Functions in the following block are called after the filter finishes processing

  // corresponding data. Those functions handle state updates and data storage (if needed)

  // according to the status returned by filter's callback functions.

  bool commonHandleAfter1xxHeadersCallback(Filter1xxHeadersStatus status);

  bool commonHandleAfterHeadersCallback(FilterHeadersStatus status, bool& end_stream);

  bool commonHandleAfterDataCallback(FilterDataStatus status, Buffer::Instance& provided_data,

                                     bool& buffer_was_streaming);

  bool commonHandleAfterTrailersCallback(FilterTrailersStatus status);

  // Buffers provided_data.

  void commonHandleBufferData(Buffer::Instance& provided_data);

  // If iteration has stopped for all frame types, calls this function to buffer the data before

  // the filter processes data. The function also updates streaming state.

  void commonBufferDataIfStopAll(Buffer::Instance& provided_data, bool& buffer_was_streaming);

  void commonContinue();

  virtual bool canContinue() PURE;

  virtual Buffer::InstancePtr createBuffer() PURE;

  virtual Buffer::InstancePtr& bufferedData() PURE;

  virtual bool observedEndStream() PURE;

  virtual bool has1xxHeaders() PURE;

  virtual void do1xxHeaders() PURE;

  virtual void doHeaders(bool end_stream) PURE;

  virtual void doData(bool end_stream) PURE;

  virtual void doTrailers() PURE;

  virtual bool hasTrailers() PURE;

  virtual void doMetadata() PURE;

  // TODO(soya3129): make this pure when adding impl to encoder filter.

  virtual void handleMetadataAfterHeadersCallback() PURE;

  // Http::StreamFilterCallbacks

  OptRef<const Network::Connection> connection() override;

  Event::Dispatcher& dispatcher() override;

  Router::RouteConstSharedPtr route() override;

  void resetStream(Http::StreamResetReason reset_reason,

                   absl::string_view transport_failure_reason) override;

  Upstream::ClusterInfoConstSharedPtr clusterInfo() override;

  uint64_t streamId() const override;

  StreamInfo::StreamInfo& streamInfo() override;

  Tracing::Span& activeSpan() override;

  OptRef<const Tracing::Config> tracingConfig() const override;

  const ScopeTrackedObject& scope() override;

  void restoreContextOnContinue(ScopeTrackedObjectStack& tracked_object_stack) override;

  void resetIdleTimer() override;

  const Router::RouteSpecificFilterConfig* mostSpecificPerFilterConfig() const override;

  Router::RouteSpecificFilterConfigs perFilterConfigs() const override;

  Http1StreamEncoderOptionsOptRef http1StreamEncoderOptions() override;

  OptRef<DownstreamStreamFilterCallbacks> downstreamCallbacks() override;

  OptRef<UpstreamStreamFilterCallbacks> upstreamCallbacks() override;

  RequestHeaderMapOptRef requestHeaders() override;

  RequestTrailerMapOptRef requestTrailers() override;

  ResponseHeaderMapOptRef informationalHeaders() override;

  ResponseHeaderMapOptRef responseHeaders() override;

  ResponseTrailerMapOptRef responseTrailers() override;

  void setBufferLimit(uint64_t limit) override;

  uint64_t bufferLimit() override;

  // Functions to set or get iteration state.

  Router::RouteConstSharedPtr getRoute() const;

  void sendLocalReply(Code code, absl::string_view body,

                      std::function<void(ResponseHeaderMap& headers)> modify_headers,

                      const absl::optional<Grpc::Status::GrpcStatus> grpc_status,

                      absl::string_view details);

  // A vector to save metadata when the current filter's [de|en]codeMetadata() can not be called,

  // either because [de|en]codeHeaders() of the current filter returns StopAllIteration or because

  // [de|en]codeHeaders() adds new metadata to [de|en]code, but we don't know

  // [de|en]codeHeaders()'s return value yet. The storage is created on demand.

  std::unique_ptr<MetadataMapVector> saved_request_metadata_{nullptr};

  std::unique_ptr<MetadataMapVector> saved_response_metadata_{nullptr};

  // The state of iteration.

  enum class IterationState : uint8_t {

    Continue,            // Iteration has not stopped for any frame type.

    StopSingleIteration, // Iteration has stopped for headers, 100-continue, or data.

    StopAllBuffer,       // Iteration has stopped for all frame types, and following data should

                         // be buffered.

    StopAllWatermark,    // Iteration has stopped for all frame types, and following data should

                         // be buffered until high watermark is reached.

  };

  FilterManager& parent_;

  IterationState iteration_state_{};

  const FilterContext filter_context_;

  // If the filter resumes iteration from a StopAllBuffer/Watermark state, the current filter

  // hasn't parsed data and trailers. As a result, the filter iteration should start with the

  // current filter instead of the next one. If true, filter iteration starts with the current

  // filter. Otherwise, starts with the next filter in the chain.

  bool iterate_from_current_filter_{};

  bool headers_continued_{};

  bool continued_1xx_headers_{};

  // If true, end_stream is called for this filter.

  bool end_stream_{};

  // If true, the filter has processed headers.

  bool processed_headers_{};

};

/**

 * Wrapper for a stream decoder filter.

 */

struct ActiveStreamDecoderFilter : public ActiveStreamFilterBase,

                                   public StreamDecoderFilterCallbacks {

  ActiveStreamDecoderFilter(FilterManager& parent, StreamDecoderFilterSharedPtr filter,

                            absl::string_view filter_config_name)

  // ActiveStreamFilterBase

  bool canContinue() override;

  Buffer::InstancePtr createBuffer() override;

  Buffer::InstancePtr& bufferedData() override;

  bool observedEndStream() override;

  void doHeaders(bool end_stream) override;

  void doData(bool end_stream) override;

  void doTrailers() override;

  bool hasTrailers() override;

  void drainSavedRequestMetadata();

  // This function is called after the filter calls decodeHeaders() to drain accumulated metadata.

  void handleMetadataAfterHeadersCallback() override;

  // Http::StreamDecoderFilterCallbacks

  void addDecodedData(Buffer::Instance& data, bool streaming) override;

  void injectDecodedDataToFilterChain(Buffer::Instance& data, bool end_stream) override;

  RequestTrailerMap& addDecodedTrailers() override;

  MetadataMapVector& addDecodedMetadata() override;

  void continueDecoding() override;

  const Buffer::Instance* decodingBuffer() override;

  void modifyDecodingBuffer(std::function<void(Buffer::Instance&)> callback) override;

  void sendLocalReply(Code code, absl::string_view body,

                      std::function<void(ResponseHeaderMap& headers)> modify_headers,

                      const absl::optional<Grpc::Status::GrpcStatus> grpc_status,

                      absl::string_view details) override;

  void encode1xxHeaders(ResponseHeaderMapPtr&& headers) override;

  void encodeHeaders(ResponseHeaderMapPtr&& headers, bool end_stream,

                     absl::string_view details) override;

  void encodeData(Buffer::Instance& data, bool end_stream) override;

  void encodeTrailers(ResponseTrailerMapPtr&& trailers) override;

  void encodeMetadata(MetadataMapPtr&& metadata_map_ptr) override;

  void onDecoderFilterAboveWriteBufferHighWatermark() override;

  void onDecoderFilterBelowWriteBufferLowWatermark() override;

  void addDownstreamWatermarkCallbacks(DownstreamWatermarkCallbacks& watermark_callbacks) override;

  void

  removeDownstreamWatermarkCallbacks(DownstreamWatermarkCallbacks& watermark_callbacks) override;

  bool recreateStream(const Http::ResponseHeaderMap* original_response_headers) override;

  void addUpstreamSocketOptions(const Network::Socket::OptionsSharedPtr& options) override;

  Network::Socket::OptionsSharedPtr getUpstreamSocketOptions() const override;

  Buffer::BufferMemoryAccountSharedPtr account() const override;

  void setUpstreamOverrideHost(Upstream::LoadBalancerContext::OverrideHost) override;

  absl::optional<Upstream::LoadBalancerContext::OverrideHost> upstreamOverrideHost() const override;

  bool shouldLoadShed() const override;

  void sendGoAwayAndClose(bool graceful = false) override;

  // Each decoder filter instance checks if the request passed to the filter is gRPC

  // so that we can issue gRPC local responses to gRPC requests. Filter's decodeHeaders()

  // called here may change the content type, so we must check it before the call.

  void requestDataTooLarge();

  void requestDataDrained();

  // Encoding end_stream by a non-terminal filters (i.e. cache filter) always causes the decoding to

  // be stopped even if independent half-close is enabled. For simplicity, independent half-close is

  // enabled only when the terminal filter is encoding the response.

  void stopDecodingIfNonTerminalFilterEncodedEndStream(bool encoded_end_stream);

  StreamDecoderFilterSharedPtr handle_;

  StreamDecoderFilters::Iterator entry_;

  bool is_grpc_request_{};

};

/**

 * Wrapper for a stream encoder filter.

 */

struct ActiveStreamEncoderFilter : public ActiveStreamFilterBase,

                                   public StreamEncoderFilterCallbacks {

  ActiveStreamEncoderFilter(FilterManager& parent, StreamEncoderFilterSharedPtr filter,

                            absl::string_view filter_config_name)

  // ActiveStreamFilterBase

  bool canContinue() override;

  Buffer::InstancePtr createBuffer() override;

  Buffer::InstancePtr& bufferedData() override;

  bool observedEndStream() override;

  bool has1xxHeaders() override;

  void do1xxHeaders() override;

  void doHeaders(bool end_stream) override;

  void doData(bool end_stream) override;

  void drainSavedResponseMetadata();

  void handleMetadataAfterHeadersCallback() override;

  void doTrailers() override;

  bool hasTrailers() override;

  // Http::StreamEncoderFilterCallbacks

  void addEncodedData(Buffer::Instance& data, bool streaming) override;

  void injectEncodedDataToFilterChain(Buffer::Instance& data, bool end_stream) override;

  ResponseTrailerMap& addEncodedTrailers() override;

  void addEncodedMetadata(MetadataMapPtr&& metadata_map) override;

  void onEncoderFilterAboveWriteBufferHighWatermark() override;

  void onEncoderFilterBelowWriteBufferLowWatermark() override;

  void continueEncoding() override;

  const Buffer::Instance* encodingBuffer() override;

  void modifyEncodingBuffer(std::function<void(Buffer::Instance&)> callback) override;

  void sendLocalReply(Code code, absl::string_view body,

                      std::function<void(ResponseHeaderMap& headers)> modify_headers,

                      const absl::optional<Grpc::Status::GrpcStatus> grpc_status,

                      absl::string_view details) override;

  void responseDataTooLarge();

  void responseDataDrained();

  StreamEncoderFilterSharedPtr handle_;

  StreamEncoderFilters::Iterator entry_;

};

/**

 * Callbacks invoked by the FilterManager to pass filter data/events back to the caller.

 */

class FilterManagerCallbacks {

public:

  /**

   * Called when the provided headers have been encoded by all the filters in the chain.

   * @param response_headers the encoded headers.

   * @param end_stream whether this is a header only response.

   */

  virtual void encodeHeaders(ResponseHeaderMap& response_headers, bool end_stream) PURE;

  /**

   * Called when the provided 100 Continue headers have been encoded by all the filters in the

   * chain.

   * @param response_headers the encoded headers.

   */

  virtual void encode1xxHeaders(ResponseHeaderMap& response_headers) PURE;

  /**

   * Called when the provided data has been encoded by all filters in the chain.

   * @param data the encoded data.

   * @param end_stream whether this is the end of the response.

   */

  virtual void encodeData(Buffer::Instance& data, bool end_stream) PURE;

  /**

   * Called when the provided trailers have been encoded by all filters in the chain.

   * @param trailers the encoded trailers.

   */

  virtual void encodeTrailers(ResponseTrailerMap& trailers) PURE;

  /**

   * Called when the provided metadata has been encoded by all filters in the chain.

   * @param trailers the encoded trailers.

   */

  virtual void encodeMetadata(MetadataMapPtr&& metadata) PURE;

  /**

   * Injects request trailers into a stream that originally did not have request trailers.

   */

  virtual void setRequestTrailers(RequestTrailerMapPtr&& request_trailers) PURE;

  /**

   * Passes ownership of received informational headers to the parent. This may be called multiple

   * times in the case of multiple upstream calls.

   */

  virtual void setInformationalHeaders(ResponseHeaderMapPtr&& response_headers) PURE;

  /**

   * Passes ownership of received response headers to the parent. This may be called multiple times

   * in the case of multiple upstream calls.

   */

  virtual void setResponseHeaders(ResponseHeaderMapPtr&& response_headers) PURE;

  /**

   * Passes ownership of received response trailers to the parent. This may be called multiple times

   * in the case of multiple upstream calls.

   */

  virtual void setResponseTrailers(ResponseTrailerMapPtr&& response_trailers) PURE;

  /**

   * Optionally updates response code stats based on the details in the headers.

   */

  // TODO(snowp): We should consider moving filter access to headers/trailers to happen via the

  // callbacks instead of via the encode/decode callbacks on the filters.

  /**

   * The downstream request headers if set.

   */

  virtual RequestHeaderMapOptRef requestHeaders() PURE;

  /**

   * The downstream request trailers if present.

   */

  virtual RequestTrailerMapOptRef requestTrailers() PURE;

  /**

   * Retrieves a pointer to the continue headers set via the call to setInformationalHeaders.

   */

  virtual ResponseHeaderMapOptRef informationalHeaders() PURE;

  /**

   * Retrieves a pointer to the response headers set via the last call to setResponseHeaders.

   * Note that response headers might be set multiple times (e.g. if a local reply is issued after

   * headers have been received but before headers have been encoded), so it is not safe in general

   * to assume that any set of headers will be valid for the duration of a stream.

   */

  virtual ResponseHeaderMapOptRef responseHeaders() PURE;

  /**

   * Retrieves a pointer to the last response trailers set via setResponseTrailers.

   * Note that response trailers might be set multiple times, so it is not safe in general to assume

   * that any set of trailers will be valid for the duration of the stream.

   */

  virtual ResponseTrailerMapOptRef responseTrailers() PURE;

  /**

   * Called after encoding has completed.

   */

  virtual void endStream() PURE;

  /**

   * Attempt to send GOAWAY and close the connection.

   */

  virtual void sendGoAwayAndClose(bool graceful = false) PURE;

  /**

   * Called when the stream write buffer is no longer above the low watermark.

   */

  virtual void onDecoderFilterBelowWriteBufferLowWatermark() PURE;

  /**

   * Called when the stream write buffer is above above the high watermark.

   */

  virtual void onDecoderFilterAboveWriteBufferHighWatermark() PURE;

  /**

   * Called when request activity indicates that the request timeout should be disarmed.

   */

  virtual void disarmRequestTimeout() PURE;

  /**

   * Called when stream activity indicates that the stream idle timeout should be reset.

   */

  virtual void resetIdleTimer() PURE;

  /**

   * Called when the stream should be re-created, e.g. for an internal redirect.

   */

  virtual void recreateStream(StreamInfo::FilterStateSharedPtr filter_state) PURE;

  /**

   * Called when the stream should be reset.

   */

  virtual void

  resetStream(Http::StreamResetReason reset_reason = Http::StreamResetReason::LocalReset,

              absl::string_view transport_failure_reason = "") PURE;

  /**

   * Returns the upgrade map for the current route entry.

   */

  virtual const Router::RouteEntry::UpgradeMap* upgradeMap() PURE;

  /**

   * Returns the cluster info for the current route entry.

   */

  virtual Upstream::ClusterInfoConstSharedPtr clusterInfo() PURE;

  /**

   * Returns the current active span.

   */

  virtual Tracing::Span& activeSpan() PURE;

  // TODO(snowp): It might make more sense to pass (optional?) counters to the FM instead of

  // calling back out to the AS to record them.

  /**

   * Called when a stream fails due to the response data being too large.

   */

  virtual void onResponseDataTooLarge() PURE;

  /**

   * Called when a stream fails due to the request data being too large.

   */

  virtual void onRequestDataTooLarge() PURE;

  /**

   * Returns the Http1StreamEncoderOptions associated with the response encoder.

   */

  virtual Http1StreamEncoderOptionsOptRef http1StreamEncoderOptions() PURE;

  /**

   * Returns the UpstreamStreamFilterCallbacks for upstream HTTP filters.

   */

  /**

   * Called when a local reply is made by the filter manager.

   * @param code the response code of the local reply.

   */

  virtual void onLocalReply(Code code) PURE;

  /**

   * Returns the tracing configuration to use for this stream.

   */

  virtual OptRef<const Tracing::Config> tracingConfig() const PURE;

  /**

   * Returns the tracked scope to use for this stream.

   */

  virtual const ScopeTrackedObject& scope() PURE;

  /**

   * Returns the DownstreamStreamFilterCallbacks for downstream HTTP filters.

   */

  /**

   * Returns if close from the upstream is to be handled with half-close semantics.

   * This is used for HTTP/1.1 codec.

   */

  virtual bool isHalfCloseEnabled() PURE;

};

/**

 * This class allows the remote address to be overridden for HTTP stream info. This is used for

 * XFF handling. This is required to avoid providing stream info with a non-const connection info

 * provider. Private inheritance from ConnectionInfoProvider is used to make sure users get the

 * address provider via the normal getter.

 */

class OverridableRemoteConnectionInfoSetterStreamInfo : public StreamInfo::StreamInfoImpl,

                                                        private Network::ConnectionInfoProvider {

public:

  using StreamInfoImpl::StreamInfoImpl;

  void setDownstreamRemoteAddress(

  // StreamInfo::StreamInfo

  // Network::ConnectionInfoProvider

private:

  Network::Address::InstanceConstSharedPtr overridden_downstream_remote_address_;

};

/**

 * FilterManager manages decoding a request through a series of decoding filter and the encoding

 * of the resulting response.

 */

class FilterManager : public ScopeTrackedObject, Logger::Loggable<Logger::Id::http> {

public:

  FilterManager(FilterManagerCallbacks& filter_manager_callbacks, Event::Dispatcher& dispatcher,

                OptRef<const Network::Connection> connection, uint64_t stream_id,

                Buffer::BufferMemoryAccountSharedPtr account, bool proxy_100_continue,

                uint64_t buffer_limit)

  // ScopeTrackedObject

  /**

   * Decodes the provided headers starting at the first filter in the chain.

   * @param headers the headers to decode.

   * @param end_stream whether the request is header only.

   */

  /**

   * Decodes the provided data starting at the first filter in the chain.

   * @param data the data to decode.

   * @param end_stream whether this data is the end of the request.

   */

  /**

   * Decodes the provided trailers starting at the first filter in the chain.

   * @param trailers the trailers to decode.

   */

  /**

   * Decodes the provided metadata starting at the first filter in the chain.

   * @param metadata_map the metadata to decode.

   */

  void disarmRequestTimeout();

  /**

   * If end_stream is true, marks encoding as complete. This is a noop if end_stream is false.

   * @param end_stream whether encoding is complete.

   */

  void maybeEndEncode(bool end_stream);

  /**

   * If terminal_filter_decoded_end_stream is true, marks decoding as complete. This is a noop if

   * terminal_filter_decoded_end_stream is false.

   * @param end_stream whether decoding is complete.

   */

  void maybeEndDecode(bool terminal_filter_decoded_end_stream);

  void checkAndCloseStreamIfFullyClosed();

  virtual void sendLocalReply(Code code, absl::string_view body,

                              const std::function<void(ResponseHeaderMap& headers)>& modify_headers,

                              const absl::optional<Grpc::Status::GrpcStatus> grpc_status,

                              absl::string_view details) PURE;

  void resetStream(StreamResetReason reason, absl::string_view transport_failure_reason);

  /**

   * Executes a prepared, but not yet propagated, local reply.

   * Prepared local replies can occur in the decoder filter chain iteration.

   */

  virtual void executeLocalReplyIfPrepared() PURE;

  // Possibly increases buffer_limit_ to the value of limit.

  void setBufferLimit(uint64_t limit);

  /**

   * @return uint64_t the current buffer limit.

   */

  /**

   * @return bool whether any above high watermark triggers are currently active

   */

  // Pass on watermark callbacks to watermark subscribers. This boils down to passing watermark

  // events for this stream and the downstream connection to the router filter.

  void callHighWatermarkCallbacks();

  void callLowWatermarkCallbacks();

  /**

   * Marks local processing as complete.

   * TODO(yanvlasov): deprecate and decommission this function.

   */

  /**

   * Whether the filters have been destroyed.

   */

  /**

   * Whether remote processing has been marked as complete.

   */

  /**

   * Instructs the FilterManager to not create a filter chain. This makes it possible to issue

   * a local reply without the overhead of creating and traversing the filters.

   */

  virtual StreamInfo::StreamInfo& streamInfo() PURE;

  virtual const StreamInfo::StreamInfo& streamInfo() const PURE;

  enum class UpgradeResult : uint8_t { UpgradeUnneeded, UpgradeAccepted, UpgradeRejected };

  /**

   * Filter chain creation result.

   */

  class CreateChainResult {

  public:

    /**

     * @param created whether the filter chain was created.

     * @param upgrade the upgrade result.

     */

    CreateChainResult(bool created, UpgradeResult upgrade = UpgradeResult::UpgradeUnneeded)

    /**

     * @return whether the filter chain was created.

     */

    /**

     * @return whether the upgrade was accepted.

     */

    /**

     * @return whether the upgrade was rejected.

     */

  private:

    bool created_ = false;

    UpgradeResult upgrade_ = UpgradeResult::UpgradeUnneeded;

  };

  /**

   * Set up the Encoder/Decoder filter chain.

   * @param filter_chain_factory the factory to create the filter chain.

   */

  CreateChainResult createFilterChain(const FilterChainFactory& filter_chain_factory);

  void contextOnContinue(ScopeTrackedObjectStack& tracked_object_stack);

    // Stop filter chain iteration by checking encoder or decoder chain.

protected:

  struct State {

    uint32_t filter_call_state_{0};

    // Set after decoder filter chain has completed iteration. Prevents further calls to decoder

    // filters. This flag is used to determine stream completion when the independent half-close is

    // enabled.

    bool decoder_filter_chain_complete_{};

    // Set after encoder filter chain has completed iteration. Prevents further calls to encoder

    // filters. This flag is used to determine stream completion when the independent half-close is

    // enabled.

    bool encoder_filter_chain_complete_{};

    // Set `true` when the filter manager observes end stream on the decoder path (from downstream

    // client) before iteration of the decoder filter chain begins. This flag is used for setting

    // end_stream value when resuming decoder filter chain iteration.

    bool observed_decode_end_stream_{};

    // Set `true` when the filter manager observes end stream on the encoder path (from upstream

    // server or Envoy's local reply) before iteration of the encoder filter chain begins. This flag

    // is used for setting end_stream value when resuming encoder filter chain iteration.

    bool observed_encode_end_stream_{};

    // By default, we will assume there are no 1xx. If encode1xxHeaders

    // is ever called, this is set to true so commonContinue resumes processing the 1xx.

    bool has_1xx_headers_{};

    // These two are latched on initial header read, to determine if the original headers

    // constituted a HEAD or gRPC request, respectively.

    bool is_head_request_{};

    bool is_grpc_request_{};

    // Tracks if headers other than 100-Continue have been encoded to the codec.

    bool non_100_response_headers_encoded_{};

    // True under the stack of onLocalReply, false otherwise.

    bool under_on_local_reply_{};

    // True when the filter chain iteration was aborted with local reply.

    bool decoder_filter_chain_aborted_{};

    bool encoder_filter_chain_aborted_{};

    bool saw_downstream_reset_{};

    // True when the stream was recreated.

    bool recreated_stream_{};

    // The following 3 members are booleans rather than part of the space-saving bitfield as they

    // are passed as arguments to functions expecting bools. Extend State using the bitfield

    // where possible.

    bool encoder_filters_streaming_{true};

    bool decoder_filters_streaming_{true};

    bool destroyed_{false};

    // Result of filter chain creation.

    CreateChainResult create_chain_result_;

    // Used to track which filter is the latest filter that has received data.

    ActiveStreamEncoderFilter* latest_data_encoding_filter_{};

    ActiveStreamDecoderFilter* latest_data_decoding_filter_{};

  };

private:

  friend class DownstreamFilterManager;

  class FilterChainFactoryCallbacksImpl : public Http::FilterChainFactoryCallbacks {

  public:

    FilterChainFactoryCallbacksImpl(FilterManager& manager)

  private:

    FilterManager& manager_;

    absl::string_view filter_config_name_;

    Router::RouteConstSharedPtr route_;

  };

  // Indicates which filter to start the iteration with.

  enum class FilterIterationStartState { AlwaysStartFromNext, CanStartFromCurrent };

  UpgradeResult createUpgradeFilterChain(const FilterChainFactory& filter_chain_factory,

                                         FilterChainFactoryCallbacksImpl& callbacks);

  // Returns the encoder filter to start iteration with.

  StreamEncoderFilters::Iterator

  commonEncodePrefix(ActiveStreamEncoderFilter* filter, bool end_stream,

                     FilterIterationStartState filter_iteration_start_state);

  // Returns the decoder filter to start iteration with.

  StreamDecoderFilters::Iterator

  commonDecodePrefix(ActiveStreamDecoderFilter* filter,

                     FilterIterationStartState filter_iteration_start_state);

  void addDecodedData(ActiveStreamDecoderFilter& filter, Buffer::Instance& data, bool streaming);

  RequestTrailerMap& addDecodedTrailers();

  MetadataMapVector& addDecodedMetadata();

  // Helper function for the case where we have a header only request, but a filter adds a body

  // to it.

  void maybeContinueDecoding(StreamDecoderFilters::Iterator maybe_continue_data_entry);

  void decodeHeaders(ActiveStreamDecoderFilter* filter, RequestHeaderMap& headers, bool end_stream);

  // Sends data through decoding filter chains. filter_iteration_start_state indicates which

  // filter to start the iteration with.

  void decodeData(ActiveStreamDecoderFilter* filter, Buffer::Instance& data, bool end_stream,

                  FilterIterationStartState filter_iteration_start_state);

  void decodeTrailers(ActiveStreamDecoderFilter* filter, RequestTrailerMap& trailers);

  void decodeMetadata(ActiveStreamDecoderFilter* filter, MetadataMap& metadata_map);

  void addEncodedData(ActiveStreamEncoderFilter& filter, Buffer::Instance& data, bool streaming);

  ResponseTrailerMap& addEncodedTrailers();

  void encode1xxHeaders(ActiveStreamEncoderFilter* filter, ResponseHeaderMap& headers);

  // As with most of the encode functions, this runs encodeHeaders on various

  // filters before calling encodeHeadersInternal which does final header munging and passes the

  // headers to the encoder.

  void maybeContinueEncoding(StreamEncoderFilters::Iterator maybe_continue_data_entry);

  void encodeHeaders(ActiveStreamEncoderFilter* filter, ResponseHeaderMap& headers,

                     bool end_stream);

  // Sends data through encoding filter chains. filter_iteration_start_state indicates which

  // filter to start the iteration with, and finally calls encodeDataInternal

  // to update stats, do end stream bookkeeping, and send the data to encoder.

  void encodeData(ActiveStreamEncoderFilter* filter, Buffer::Instance& data, bool end_stream,

                  FilterIterationStartState filter_iteration_start_state);

  void encodeTrailers(ActiveStreamEncoderFilter* filter, ResponseTrailerMap& trailers);

  void encodeMetadata(ActiveStreamEncoderFilter* filter, MetadataMapPtr&& metadata_map_ptr);

  // Returns true if new metadata is decoded. Otherwise, returns false.

  bool processNewlyAddedMetadata();

  // Returns true if filter has stopped iteration for all frame types. Otherwise, returns false.

  // filter_streaming is the variable to indicate if stream is streaming, and its value may be

  // changed by the function.

  bool handleDataIfStopAll(ActiveStreamFilterBase& filter, Buffer::Instance& data,

                           bool& filter_streaming);

  // Returns true if the decoder filter chain should not process any more frames.

  // This includes cases where the chain was explicitly aborted (e.g., local reply)

  // or where the downstream connection has been reset.

  bool isTerminalDecoderFilter(const ActiveStreamDecoderFilter& filter) const;

  FilterManagerCallbacks& filter_manager_callbacks_;

  Event::Dispatcher& dispatcher_;

  // This is unset if there is no downstream connection, e.g. for health check or

  // async requests.

  OptRef<const Network::Connection> connection_;

  const uint64_t stream_id_;

  Buffer::BufferMemoryAccountSharedPtr account_;

  const bool proxy_100_continue_;

  StreamDecoderFilters decoder_filters_;

  StreamEncoderFilters encoder_filters_;

  std::vector<StreamFilterBase*> filters_;

  AccessLog::InstanceSharedPtrVector access_log_handlers_;

  // Stores metadata added in the decoding filter that is being processed. Will be cleared before

  // processing the next filter. The storage is created on demand. We need to store metadata

  // temporarily in the filter in case the filter has stopped all while processing headers.

  std::unique_ptr<MetadataMapVector> request_metadata_map_vector_;

  Buffer::InstancePtr buffered_response_data_;

  Buffer::InstancePtr buffered_request_data_;

  uint64_t buffer_limit_{0};

  uint32_t high_watermark_count_{0};

  std::list<DownstreamWatermarkCallbacks*> watermark_callbacks_;

  Network::Socket::OptionsSharedPtr upstream_options_ =

      std::make_shared<Network::Socket::Options>();

  std::pair<std::string, bool> upstream_override_host_;

  // TODO(snowp): Once FM has been moved to its own file we'll make these private classes of FM,

  // at which point they no longer need to be friends.

  friend ActiveStreamFilterBase;

  friend ActiveStreamDecoderFilter;

  friend ActiveStreamEncoderFilter;

  /**

   * Flags that keep track of which filter calls are currently in progress.

   */

  // clang-format off

    struct FilterCallState {

      static constexpr uint32_t DecodeHeaders   = 0x01;

      static constexpr uint32_t DecodeData      = 0x02;

      static constexpr uint32_t DecodeMetadata  = 0x04;

      static constexpr uint32_t DecodeTrailers  = 0x08;

      static constexpr uint32_t EncodeHeaders   = 0x10;

      static constexpr uint32_t EncodeData      = 0x20;

      static constexpr uint32_t EncodeMetadata  = 0x40;

      static constexpr uint32_t EncodeTrailers  = 0x80;

      // Encode1xxHeaders is a bit of a special state as 1xx

      // headers may be sent during request processing. This state is only used

      // to verify we do not encode1xx headers more than once per

      // filter.

      static constexpr uint32_t Encode1xxHeaders  = 0x100;

      // Used to indicate that one of the following conditions is true:

      // 1. The filter manager is processing the final [En|De]codeData frame.

      // 2. The filter manager is in [En|De]codeHeaders and is processing a

      //    header-only request or response.

      static constexpr uint32_t EndOfStream = 0x200;

      // Masks for filter call state.

      static constexpr uint32_t IsDecodingMask = DecodeHeaders | DecodeData | DecodeMetadata | DecodeTrailers;

      static constexpr uint32_t IsEncodingMask = EncodeHeaders | Encode1xxHeaders | EncodeData | EncodeMetadata | EncodeTrailers;

    };

  // clang-format on

  State state_;

};

// The DownstreamFilterManager has explicit handling to send local replies.

// The UpstreamFilterManager will not, and will instead defer local reply

// management to the DownstreamFilterManager.

class DownstreamFilterManager : public FilterManager {

public:

  DownstreamFilterManager(FilterManagerCallbacks& filter_manager_callbacks,

                          Event::Dispatcher& dispatcher, const Network::Connection& connection,

                          uint64_t stream_id, Buffer::BufferMemoryAccountSharedPtr account,

                          bool proxy_100_continue, uint32_t buffer_limit,

                          FilterChainFactory& filter_chain_factory,

                          const LocalReply::LocalReply& local_reply, Http::Protocol protocol,

                          TimeSource& time_source,

                          StreamInfo::FilterStateSharedPtr parent_filter_state,

                          Server::OverloadManager& overload_manager)

  // TODO(snowp): This should probably return a StreamInfo instead of the impl.

  void setDownstreamRemoteAddress(

  CreateChainResult createDownstreamFilterChain();

  /**

   * Called before local reply is made by the filter manager.

   * @param data the data associated with the local reply.

   */

  void onLocalReply(StreamFilterBase::LocalReplyData& data);

  void sendLocalReply(Code code, absl::string_view body,

                      const std::function<void(ResponseHeaderMap& headers)>& modify_headers,

                      const absl::optional<Grpc::Status::GrpcStatus> grpc_status,

                      absl::string_view details) override;

  /**

   * Whether downstream has observed end_stream.

   */

  /**

   * Return true if the timestamp of the downstream end_stream was recorded.

   * For the HCM to handle the case of internal redirects, timeout error replies

   * and stream resets on premature upstream response.

   */

private:

  /**

   * Sends a local reply by constructing a response and passing it through all the encoder

   * filters. The resulting response will be passed out via the FilterManagerCallbacks.

   * This will be deprecated in favor of prepareLocalReplyViaFilterChain.

   */

  void sendLocalReplyViaFilterChain(

      bool is_grpc_request, Code code, absl::string_view body,

      const std::function<void(ResponseHeaderMap& headers)>& modify_headers, bool is_head_request,

      const absl::optional<Grpc::Status::GrpcStatus> grpc_status, absl::string_view details);

  /**

   * Prepares a local reply that will be sent along the encoder filters in

   * executeLocalReplyViaFilterChain.

   */

  void prepareLocalReplyViaFilterChain(

      bool is_grpc_request, Code code, absl::string_view body,

      const std::function<void(ResponseHeaderMap& headers)>& modify_headers, bool is_head_request,

      const absl::optional<Grpc::Status::GrpcStatus> grpc_status, absl::string_view details);

  /**

   * Executes a prepared local reply along the encoder filters.

   */

  void executeLocalReplyIfPrepared() override;

  /**

   * Sends a local reply by constructing a response and skipping the encoder filters. The

   * resulting response will be passed out via the FilterManagerCallbacks.

   */

  void sendDirectLocalReply(Code code, absl::string_view body,

                            const std::function<void(ResponseHeaderMap& headers)>& modify_headers,

                            bool is_head_request,

                            const absl::optional<Grpc::Status::GrpcStatus> grpc_status);

private:

  OverridableRemoteConnectionInfoSetterStreamInfo stream_info_;

  const LocalReply::LocalReply& local_reply_;

  const FilterChainFactory& filter_chain_factory_;

  Utility::PreparedLocalReplyPtr prepared_local_reply_{nullptr};

  Server::LoadShedPoint* downstream_filter_load_shed_point_{nullptr};

};

} // namespace Http

} // namespace Envoy