src/core/lib/transport/metadata_batch.h

/*
 *
 * Copyright 2015 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

#ifndef GRPC_CORE_LIB_TRANSPORT_METADATA_BATCH_H
#define GRPC_CORE_LIB_TRANSPORT_METADATA_BATCH_H

#include <grpc/support/port_platform.h>

#include <stdlib.h>

#include <cstdint>
#include <string>
#include <type_traits>
#include <utility>

#include "absl/container/inlined_vector.h"
#include "absl/functional/function_ref.h"
#include "absl/meta/type_traits.h"
#include "absl/strings/numbers.h"
#include "absl/strings/string_view.h"
#include "absl/types/optional.h"

#include <grpc/impl/codegen/compression_types.h>
#include <grpc/status.h>
#include <grpc/support/log.h>

#include "src/core/lib/compression/compression_internal.h"
#include "src/core/lib/gprpp/chunked_vector.h"
#include "src/core/lib/gprpp/table.h"
#include "src/core/lib/gprpp/time.h"
#include "src/core/lib/resource_quota/arena.h"
#include "src/core/lib/slice/slice.h"
#include "src/core/lib/transport/parsed_metadata.h"

namespace grpc_core {

// grpc-timeout metadata trait.
// ValueType is defined as Timestamp - an absolute timestamp (i.e. a
// deadline!), that is converted to a duration by transports before being
// sent.
// TODO(ctiller): Move this elsewhere. During the transition we need to be able
// to name this in MetadataMap, but ultimately once the transition is done we
// should not need to.
struct GrpcTimeoutMetadata {
  static constexpr bool kRepeatable = false;
  using ValueType = Timestamp;
  using MementoType = Duration;
  static absl::string_view key() { return "grpc-timeout"; }
  static MementoType ParseMemento(Slice value, MetadataParseErrorFn on_error);
  static ValueType MementoToValue(MementoType timeout);
  static Slice Encode(ValueType x);
  static std::string DisplayValue(MementoType x) { return x.ToString(); }
};

// TE metadata trait.
struct TeMetadata {
  static constexpr bool kRepeatable = false;
  // HTTP2 says that TE can either be empty or "trailers".
  // Empty means this trait is not included, "trailers" means kTrailers, and
  // kInvalid is used to remember an invalid value.
  enum ValueType : uint8_t {
    kTrailers,
    kInvalid,
  };
  using MementoType = ValueType;
  static absl::string_view key() { return "te"; }
  static MementoType ParseMemento(Slice value, MetadataParseErrorFn on_error);
  static ValueType MementoToValue(MementoType te) { return te; }
  static StaticSlice Encode(ValueType x) {
    GPR_ASSERT(x == kTrailers);
    return StaticSlice::FromStaticString("trailers");
  }
  static const char* DisplayValue(MementoType te);
};

// content-type metadata trait.
struct ContentTypeMetadata {
  static constexpr bool kRepeatable = false;
  // gRPC says that content-type can be application/grpc[;something]
  // Core has only ever verified the prefix.
  // IF we want to start verifying more, we can expand this type.
  enum ValueType {
    kApplicationGrpc,
    kEmpty,
    kInvalid,
  };
  using MementoType = ValueType;
  static absl::string_view key() { return "content-type"; }
  static MementoType ParseMemento(Slice value, MetadataParseErrorFn on_error);
  static ValueType MementoToValue(MementoType content_type) {
    return content_type;
  }

  static StaticSlice Encode(ValueType x);
  static const char* DisplayValue(MementoType content_type);
};

// scheme metadata trait.
struct HttpSchemeMetadata {
  static constexpr bool kRepeatable = false;
  enum ValueType {
    kHttp,
    kHttps,
    kInvalid,
  };
  using MementoType = ValueType;
  static absl::string_view key() { return ":scheme"; }
  static MementoType ParseMemento(Slice value, MetadataParseErrorFn on_error) {
    return Parse(value.as_string_view(), on_error);
  }
  static ValueType Parse(absl::string_view value,
                         MetadataParseErrorFn on_error);
  static ValueType MementoToValue(MementoType content_type) {
    return content_type;
  }
  static StaticSlice Encode(ValueType x);
  static const char* DisplayValue(MementoType content_type);
};

// method metadata trait.
struct HttpMethodMetadata {
  static constexpr bool kRepeatable = false;
  enum ValueType {
    kPost,
    kGet,
    kPut,
    kInvalid,
  };
  using MementoType = ValueType;
  static absl::string_view key() { return ":method"; }
  static MementoType ParseMemento(Slice value, MetadataParseErrorFn on_error);
  static ValueType MementoToValue(MementoType content_type) {
    return content_type;
  }
  static StaticSlice Encode(ValueType x);
  static const char* DisplayValue(MementoType content_type);
};

// Base type for metadata pertaining to a single compression algorithm
// (e.g., "grpc-encoding").
struct CompressionAlgorithmBasedMetadata {
  using ValueType = grpc_compression_algorithm;
  using MementoType = ValueType;
  static MementoType ParseMemento(Slice value, MetadataParseErrorFn on_error);
  static ValueType MementoToValue(MementoType x) { return x; }
  static Slice Encode(ValueType x) {
    GPR_ASSERT(x != GRPC_COMPRESS_ALGORITHMS_COUNT);
    return Slice::FromStaticString(CompressionAlgorithmAsString(x));
  }
  static const char* DisplayValue(MementoType x) {
    if (const char* p = CompressionAlgorithmAsString(x)) {
      return p;
    } else {
      return "<discarded-invalid-value>";
    }
  }
};

// grpc-encoding metadata trait.
struct GrpcEncodingMetadata : public CompressionAlgorithmBasedMetadata {
  static constexpr bool kRepeatable = false;
  static absl::string_view key() { return "grpc-encoding"; }
};

// grpc-internal-encoding-request metadata trait.
struct GrpcInternalEncodingRequest : public CompressionAlgorithmBasedMetadata {
  static constexpr bool kRepeatable = false;
  static absl::string_view key() { return "grpc-internal-encoding-request"; }
};

// grpc-accept-encoding metadata trait.
struct GrpcAcceptEncodingMetadata {
  static constexpr bool kRepeatable = false;
  static absl::string_view key() { return "grpc-accept-encoding"; }
  using ValueType = CompressionAlgorithmSet;
  using MementoType = ValueType;
  static MementoType ParseMemento(Slice value, MetadataParseErrorFn) {
    return CompressionAlgorithmSet::FromString(value.as_string_view());
  }
  static ValueType MementoToValue(MementoType x) { return x; }
  static Slice Encode(ValueType x) { return x.ToSlice(); }
  static absl::string_view DisplayValue(MementoType x) { return x.ToString(); }
};

struct SimpleSliceBasedMetadata {
  using ValueType = Slice;
  using MementoType = Slice;
  static MementoType ParseMemento(Slice value, MetadataParseErrorFn) {
    return value.TakeOwned();
  }
  static ValueType MementoToValue(MementoType value) { return value; }
  static Slice Encode(const ValueType& x) { return x.Ref(); }
  static absl::string_view DisplayValue(const MementoType& value) {
    return value.as_string_view();
  }
};

// user-agent metadata trait.
struct UserAgentMetadata : public SimpleSliceBasedMetadata {
  static constexpr bool kRepeatable = false;
  static absl::string_view key() { return "user-agent"; }
};

// grpc-message metadata trait.
struct GrpcMessageMetadata : public SimpleSliceBasedMetadata {
  static constexpr bool kRepeatable = false;
  static absl::string_view key() { return "grpc-message"; }
};

// host metadata trait.
struct HostMetadata : public SimpleSliceBasedMetadata {
  static constexpr bool kRepeatable = false;
  static absl::string_view key() { return "host"; }
};

// endpoint-load-metrics-bin metadata trait.
struct EndpointLoadMetricsBinMetadata : public SimpleSliceBasedMetadata {
  static constexpr bool kRepeatable = false;
  static absl::string_view key() { return "endpoint-load-metrics-bin"; }
};

// grpc-server-stats-bin metadata trait.
struct GrpcServerStatsBinMetadata : public SimpleSliceBasedMetadata {
  static constexpr bool kRepeatable = false;
  static absl::string_view key() { return "grpc-server-stats-bin"; }
};

// grpc-trace-bin metadata trait.
struct GrpcTraceBinMetadata : public SimpleSliceBasedMetadata {
  static constexpr bool kRepeatable = false;
  static absl::string_view key() { return "grpc-trace-bin"; }
};

// grpc-tags-bin metadata trait.
struct GrpcTagsBinMetadata : public SimpleSliceBasedMetadata {
  static constexpr bool kRepeatable = false;
  static absl::string_view key() { return "grpc-tags-bin"; }
};

// :authority metadata trait.
struct HttpAuthorityMetadata : public SimpleSliceBasedMetadata {
  static constexpr bool kRepeatable = false;
  static absl::string_view key() { return ":authority"; }
};

// :path metadata trait.
struct HttpPathMetadata : public SimpleSliceBasedMetadata {
  static constexpr bool kRepeatable = false;
  static absl::string_view key() { return ":path"; }
};

// We separate SimpleIntBasedMetadata into two pieces: one that does not
// depend on the invalid value, and one that does. This allows the compiler to
// easily see the functions that are shared, and helps reduce code bloat here.
template <typename Int>
struct SimpleIntBasedMetadataBase {
  using ValueType = Int;
  using MementoType = Int;
  static ValueType MementoToValue(MementoType value) { return value; }
  static Slice Encode(ValueType x) { return Slice::FromInt64(x); }
  static Int DisplayValue(MementoType x) { return x; }
};

template <typename Int, Int kInvalidValue>
struct SimpleIntBasedMetadata : public SimpleIntBasedMetadataBase<Int> {
  static constexpr Int invalid_value() { return kInvalidValue; }
  static Int ParseMemento(Slice value, MetadataParseErrorFn on_error) {
    Int out;
    if (!absl::SimpleAtoi(value.as_string_view(), &out)) {
      on_error("not an integer", value);
      out = kInvalidValue;
    }
    return out;
  }
};

// grpc-status metadata trait.
struct GrpcStatusMetadata
    : public SimpleIntBasedMetadata<grpc_status_code, GRPC_STATUS_UNKNOWN> {
  static constexpr bool kRepeatable = false;
  static absl::string_view key() { return "grpc-status"; }
};

// grpc-previous-rpc-attempts metadata trait.
struct GrpcPreviousRpcAttemptsMetadata
    : public SimpleIntBasedMetadata<uint32_t, 0> {
  static constexpr bool kRepeatable = false;
  static absl::string_view key() { return "grpc-previous-rpc-attempts"; }
};

// grpc-retry-pushback-ms metadata trait.
struct GrpcRetryPushbackMsMetadata {
  static constexpr bool kRepeatable = false;
  static absl::string_view key() { return "grpc-retry-pushback-ms"; }
  using ValueType = Duration;
  using MementoType = Duration;
  static ValueType MementoToValue(MementoType x) { return x; }
  static Slice Encode(Duration x) { return Slice::FromInt64(x.millis()); }
  static int64_t DisplayValue(Duration x) { return x.millis(); }
  static Duration ParseMemento(Slice value, MetadataParseErrorFn on_error);
};

// :status metadata trait.
// TODO(ctiller): consider moving to uint16_t
struct HttpStatusMetadata : public SimpleIntBasedMetadata<uint32_t, 0> {
  static constexpr bool kRepeatable = false;
  static absl::string_view key() { return ":status"; }
};

// "secret" metadata trait used to pass load balancing token between filters.
// This should not be exposed outside of gRPC core.
class GrpcLbClientStats;

struct GrpcLbClientStatsMetadata {
  static constexpr bool kRepeatable = false;
  static absl::string_view key() { return "grpclb_client_stats"; }
  using ValueType = GrpcLbClientStats*;
  using MementoType = ValueType;
  static ValueType MementoToValue(MementoType value) { return value; }
  static Slice Encode(ValueType) { abort(); }
  static const char* DisplayValue(MementoType) { return "<internal-lb-stats>"; }
  static MementoType ParseMemento(Slice, MetadataParseErrorFn) {
    return nullptr;
  }
};

// lb-token metadata
struct LbTokenMetadata : public SimpleSliceBasedMetadata {
  static constexpr bool kRepeatable = false;
  static absl::string_view key() { return "lb-token"; }
};

// lb-cost-bin metadata
struct LbCostBinMetadata {
  static constexpr bool kRepeatable = true;
  static absl::string_view key() { return "lb-cost-bin"; }
  struct ValueType {
    double cost;
    std::string name;
  };
  using MementoType = ValueType;
  static ValueType MementoToValue(MementoType value) { return value; }
  static Slice Encode(const ValueType& x);
  static std::string DisplayValue(MementoType x);
  static MementoType ParseMemento(Slice value, MetadataParseErrorFn on_error);
};

// Annotation added by a transport to note whether a failed request was never
// placed on the wire, or never seen by a server.
struct GrpcStreamNetworkState {
  static absl::string_view DebugKey() { return "GrpcStreamNetworkState"; }
  static constexpr bool kRepeatable = false;
  enum ValueType : uint8_t {
    kNotSentOnWire,
    kNotSeenByServer,
  };
  static std::string DisplayValue(ValueType x);
};

// Annotation added by a server transport to note the peer making a request.
struct PeerString {
  static absl::string_view DebugKey() { return "PeerString"; }
  static constexpr bool kRepeatable = false;
  using ValueType = absl::string_view;
  static std::string DisplayValue(ValueType x);
};

// Annotation added by various systems to describe the reason for a failure.
struct GrpcStatusContext {
  static absl::string_view DebugKey() { return "GrpcStatusContext"; }
  static constexpr bool kRepeatable = true;
  using ValueType = std::string;
  static const std::string& DisplayValue(const std::string& x);
};

namespace metadata_detail {

// Build a key/value formatted debug string.
// Output looks like 'key1: value1, key2: value2'
// The string is expected to be readable, but not necessarily parsable.
class DebugStringBuilder {
 public:
  // Add one key/value pair to the output.
  void Add(absl::string_view key, absl::string_view value);

  // Finalize the output and return the string.
  // Subsequent Add calls are UB.
  std::string TakeOutput() { return std::move(out_); }

 private:
  std::string out_;
};

// IsEncodable: Given a trait, determine if that trait is encodable, or is
// just a value attached to a MetadataMap. We use the presence of the key()
// static method to determine if a trait is encodable or not - encodable
// traits have string names, and non-encodable traits do not.
template <typename Trait, typename Ignored = void>
struct IsEncodableTrait {
  static const bool value = false;
};

template <typename Trait>
struct IsEncodableTrait<Trait, absl::void_t<decltype(Trait::key())>> {
  static const bool value = true;
};

// Helper type - maps a string name to a trait.
template <typename MustBeVoid, typename... Traits>
struct NameLookup;

template <typename Trait, typename... Traits>
struct NameLookup<absl::enable_if_t<IsEncodableTrait<Trait>::value, void>,
                  Trait, Traits...> {
  // Call op->Found(Trait()) if op->name == Trait::key() for some Trait in
  // Traits. If not found, call op->NotFound().
  template <typename Op>
  static auto Lookup(absl::string_view key, Op* op)
      -> decltype(op->Found(Trait())) {
    if (key == Trait::key()) {
      return op->Found(Trait());
    }
    return NameLookup<void, Traits...>::Lookup(key, op);
  }
};

template <typename Trait, typename... Traits>
struct NameLookup<absl::enable_if_t<!IsEncodableTrait<Trait>::value, void>,
                  Trait, Traits...> {
  template <typename Op>
  static auto Lookup(absl::string_view key, Op* op)
      -> decltype(NameLookup<void, Traits...>::Lookup(key, op)) {
    return NameLookup<void, Traits...>::Lookup(key, op);
  }
};

template <>
struct NameLookup<void> {
  template <typename Op>
  static auto Lookup(absl::string_view key, Op* op)
      -> decltype(op->NotFound(key)) {
    return op->NotFound(key);
  }
};

// Helper to take a slice to a memento to a value.
// By splitting this part out we can scale code size as the number of
// (memento, value) types, rather than as the number of traits.
template <typename ParseMementoFn, typename MementoToValueFn>
struct ParseValue {
  template <ParseMementoFn parse_memento, MementoToValueFn memento_to_value>
  static GPR_ATTRIBUTE_NOINLINE auto Parse(Slice* value,
                                           MetadataParseErrorFn on_error)
      -> decltype(memento_to_value(parse_memento(std::move(*value),
                                                 on_error))) {
    return memento_to_value(parse_memento(std::move(*value), on_error));
  }
};

// This is an "Op" type for NameLookup.
// Used for MetadataMap::Parse, its Found/NotFound methods turn a slice into a
// ParsedMetadata object.
template <typename Container>
class ParseHelper {
 public:
  ParseHelper(Slice value, MetadataParseErrorFn on_error, size_t transport_size)
      : value_(std::move(value)),
        on_error_(on_error),
        transport_size_(transport_size) {}

  template <typename Trait>
  GPR_ATTRIBUTE_NOINLINE ParsedMetadata<Container> Found(Trait trait) {
    return ParsedMetadata<Container>(
        trait,
        ParseValueToMemento<typename Trait::MementoType, Trait::ParseMemento>(),
        transport_size_);
  }

  GPR_ATTRIBUTE_NOINLINE ParsedMetadata<Container> NotFound(
      absl::string_view key) {
    return ParsedMetadata<Container>(Slice::FromCopiedString(key),
                                     std::move(value_));
  }

 private:
  template <typename T, T (*parse_memento)(Slice, MetadataParseErrorFn)>
  GPR_ATTRIBUTE_NOINLINE T ParseValueToMemento() {
    return parse_memento(std::move(value_), on_error_);
  }

  Slice value_;
  MetadataParseErrorFn on_error_;
  const size_t transport_size_;
};

// This is an "Op" type for NameLookup.
// Used for MetadataMap::Append, its Found/NotFound methods turn a slice into
// a value and add it to a container.
template <typename Container>
class AppendHelper {
 public:
  AppendHelper(Container* container, Slice value, MetadataParseErrorFn on_error)
      : container_(container), value_(std::move(value)), on_error_(on_error) {}

  template <typename Trait>
  GPR_ATTRIBUTE_NOINLINE void Found(Trait trait) {
    container_->Set(
        trait, ParseValue<decltype(Trait::ParseMemento),
                          decltype(Trait::MementoToValue)>::
                   template Parse<Trait::ParseMemento, Trait::MementoToValue>(
                       &value_, on_error_));
  }

  GPR_ATTRIBUTE_NOINLINE void NotFound(absl::string_view key) {
    container_->unknown_.Append(key, std::move(value_));
  }

 private:
  Container* const container_;
  Slice value_;
  MetadataParseErrorFn on_error_;
};

// This is an "Op" type for NameLookup.
// Used for MetadataMap::Remove, its Found/NotFound methods remove a key from
// the container.
template <typename Container>
class RemoveHelper {
 public:
  explicit RemoveHelper(Container* container) : container_(container) {}

  template <typename Trait>
  GPR_ATTRIBUTE_NOINLINE void Found(Trait trait) {
    container_->Remove(trait);
  }

  GPR_ATTRIBUTE_NOINLINE void NotFound(absl::string_view key) {
    container_->unknown_.Remove(key);
  }

 private:
  Container* const container_;
};

// This is an "Op" type for NameLookup.
// Used for MetadataMap::GetStringValue, its Found/NotFound methods generated
// a string value from the container.
template <typename Container>
class GetStringValueHelper {
 public:
  explicit GetStringValueHelper(const Container* container,
                                std::string* backing)
      : container_(container), backing_(backing) {}

  template <typename Trait>
  GPR_ATTRIBUTE_NOINLINE absl::enable_if_t<
      Trait::kRepeatable == false &&
          std::is_same<Slice, typename Trait::ValueType>::value,
      absl::optional<absl::string_view>>
  Found(Trait) {
    const auto* value = container_->get_pointer(Trait());
    if (value == nullptr) return absl::nullopt;
    return value->as_string_view();
  }

  template <typename Trait>
  GPR_ATTRIBUTE_NOINLINE absl::enable_if_t<
      Trait::kRepeatable == true &&
          !std::is_same<Slice, typename Trait::ValueType>::value,
      absl::optional<absl::string_view>>
  Found(Trait) {
    const auto* value = container_->get_pointer(Trait());
    if (value == nullptr) return absl::nullopt;
    backing_->clear();
    for (const auto& v : *value) {
      if (!backing_->empty()) backing_->push_back(',');
      auto new_segment = Trait::Encode(v);
      backing_->append(new_segment.begin(), new_segment.end());
    }
    return *backing_;
  }

  template <typename Trait>
  GPR_ATTRIBUTE_NOINLINE absl::enable_if_t<
      Trait::kRepeatable == false &&
          !std::is_same<Slice, typename Trait::ValueType>::value,
      absl::optional<absl::string_view>>
  Found(Trait) {
    const auto* value = container_->get_pointer(Trait());
    if (value == nullptr) return absl::nullopt;
    *backing_ = std::string(Trait::Encode(*value).as_string_view());
    return *backing_;
  }

  GPR_ATTRIBUTE_NOINLINE absl::optional<absl::string_view> NotFound(
      absl::string_view key) {
    return container_->unknown_.GetStringValue(key, backing_);
  }

 private:
  const Container* const container_;
  std::string* backing_;
};

// Sink for key value logs
using LogFn = absl::FunctionRef<void(absl::string_view, absl::string_view)>;

template <typename T>
struct AdaptDisplayValueToLog {
  static std::string ToString(const T& value) { return std::to_string(value); }
};

template <>
struct AdaptDisplayValueToLog<std::string> {
  static std::string ToString(const std::string& value) { return value; }
};

template <>
struct AdaptDisplayValueToLog<const std::string&> {
  static std::string ToString(const std::string& value) { return value; }
};

template <>
struct AdaptDisplayValueToLog<Slice> {
  static std::string ToString(Slice value) {
    return std::string(value.as_string_view());
  }
};

template <>
struct AdaptDisplayValueToLog<StaticSlice> {
  static absl::string_view ToString(StaticSlice value) {
    return value.as_string_view();
  }
};

template <typename T, typename U, typename V>
GPR_ATTRIBUTE_NOINLINE void LogKeyValueTo(absl::string_view key, const T& value,
                                          V (*display_value)(U), LogFn log_fn) {
  log_fn(key, AdaptDisplayValueToLog<V>::ToString(display_value(value)));
}

// Generate a strong type for metadata values per trait.
template <typename Which, typename Ignored = void>
struct Value;

template <typename Which>
struct Value<Which, absl::enable_if_t<Which::kRepeatable == false &&
                                          IsEncodableTrait<Which>::value,
                                      void>> {
  Value() = default;
  explicit Value(const typename Which::ValueType& value) : value(value) {}
  explicit Value(typename Which::ValueType&& value)
      : value(std::forward<typename Which::ValueType>(value)) {}
  Value(const Value&) = delete;
  Value& operator=(const Value&) = delete;
  Value(Value&&) noexcept = default;
  Value& operator=(Value&& other) noexcept {
    value = std::move(other.value);
    return *this;
  }
  template <typename Encoder>
  void EncodeTo(Encoder* encoder) const {
    encoder->Encode(Which(), value);
  }
  void LogTo(LogFn log_fn) const {
    LogKeyValueTo(Which::key(), value, Which::Encode, log_fn);
  }
  using StorageType = typename Which::ValueType;
  GPR_NO_UNIQUE_ADDRESS StorageType value;
};

template <typename Which>
struct Value<Which, absl::enable_if_t<Which::kRepeatable == false &&
                                          !IsEncodableTrait<Which>::value,
                                      void>> {
  Value() = default;
  explicit Value(const typename Which::ValueType& value) : value(value) {}
  explicit Value(typename Which::ValueType&& value)
      : value(std::forward<typename Which::ValueType>(value)) {}
  Value(const Value&) = delete;
  Value& operator=(const Value&) = delete;
  Value(Value&&) noexcept = default;
  Value& operator=(Value&& other) noexcept {
    value = std::move(other.value);
    return *this;
  }
  template <typename Encoder>
  void EncodeTo(Encoder*) const {}
  void LogTo(LogFn log_fn) const {
    LogKeyValueTo(Which::DebugKey(), value, Which::DisplayValue, log_fn);
  }
  using StorageType = typename Which::ValueType;
  GPR_NO_UNIQUE_ADDRESS StorageType value;
};

template <typename Which>
struct Value<Which, absl::enable_if_t<Which::kRepeatable == true &&
                                          IsEncodableTrait<Which>::value,
                                      void>> {
  Value() = default;
  explicit Value(const typename Which::ValueType& value) {
    this->value.push_back(value);
  }
  explicit Value(typename Which::ValueType&& value) {
    this->value.emplace_back(std::forward<typename Which::ValueType>(value));
  }
  Value(const Value&) = delete;
  Value& operator=(const Value&) = delete;
  Value(Value&& other) noexcept : value(std::move(other.value)) {}
  Value& operator=(Value&& other) noexcept {
    value = std::move(other.value);
    return *this;
  }
  template <typename Encoder>
  void EncodeTo(Encoder* encoder) const {
    for (const auto& v : value) {
      encoder->Encode(Which(), v);
    }
  }
  void LogTo(LogFn log_fn) const {
    for (const auto& v : value) {
      LogKeyValueTo(Which::key(), v, Which::Encode, log_fn);
    }
  }
  using StorageType = absl::InlinedVector<typename Which::ValueType, 1>;
  StorageType value;
};

template <typename Which>
struct Value<Which, absl::enable_if_t<Which::kRepeatable == true &&
                                          !IsEncodableTrait<Which>::value,
                                      void>> {
  Value() = default;
  explicit Value(const typename Which::ValueType& value) {
    this->value.push_back(value);
  }
  explicit Value(typename Which::ValueType&& value) {
    this->value.emplace_back(std::forward<typename Which::ValueType>(value));
  }
  Value(const Value&) = delete;
  Value& operator=(const Value&) = delete;
  Value(Value&& other) noexcept : value(std::move(other.value)) {}
  Value& operator=(Value&& other) noexcept {
    value = std::move(other.value);
    return *this;
  }
  template <typename Encoder>
  void EncodeTo(Encoder*) const {}
  void LogTo(LogFn log_fn) const {
    for (const auto& v : value) {
      LogKeyValueTo(Which::DebugKey(), v, Which::DisplayValue, log_fn);
    }
  }
  using StorageType = absl::InlinedVector<typename Which::ValueType, 1>;
  StorageType value;
};

// Encoder to copy some metadata
template <typename Output>
class CopySink {
 public:
  explicit CopySink(Output* dst) : dst_(dst) {}

  template <class T, class V>
  void Encode(T trait, V value) {
    dst_->Set(trait, value);
  }

  template <class T>
  void Encode(T trait, const Slice& value) {
    dst_->Set(trait, std::move(value.AsOwned()));
  }

  void Encode(const Slice& key, const Slice& value) {
    dst_->unknown_.Append(key.as_string_view(), value.Ref());
  }

 private:
  Output* dst_;
};

// Callable for the ForEach in Encode() -- for each value, call the
// appropriate encoder method.
template <typename Encoder>
struct EncodeWrapper {
  Encoder* encoder;
  template <typename Which>
  void operator()(const Value<Which>& which) {
    which.EncodeTo(encoder);
  }
};

// Callable for the ForEach in Log()
struct LogWrapper {
  LogFn log_fn;
  template <typename Which>
  void operator()(const Value<Which>& which) {
    which.LogTo(log_fn);
  }
};

// Encoder to compute TransportSize
class TransportSizeEncoder {
 public:
  void Encode(const Slice& key, const Slice& value) {
    size_ += key.length() + value.length() + 32;
  }

  template <typename Which>
  void Encode(Which, const typename Which::ValueType& value) {
    Add(Which(), value);
  }

  void Encode(ContentTypeMetadata,
              const typename ContentTypeMetadata::ValueType& value) {
    if (value == ContentTypeMetadata::kInvalid) return;
    Add(ContentTypeMetadata(), value);
  }

  size_t size() const { return size_; }

 private:
  template <typename Which>
  void Add(Which, const typename Which::ValueType& value) {
    size_ += Which::key().length() + Which::Encode(value).length() + 32;
  }

  uint32_t size_ = 0;
};

// Handle unknown (non-trait-based) fields in the metadata map.
class UnknownMap {
 public:
  explicit UnknownMap(Arena* arena) : unknown_(arena) {}

  using BackingType = ChunkedVector<std::pair<Slice, Slice>, 10>;

  void Append(absl::string_view key, Slice value);
  void Remove(absl::string_view key);
  absl::optional<absl::string_view> GetStringValue(absl::string_view key,
                                                   std::string* backing) const;

  BackingType::ConstForwardIterator begin() const { return unknown_.cbegin(); }
  BackingType::ConstForwardIterator end() const { return unknown_.cend(); }

  bool empty() const { return unknown_.empty(); }
  size_t size() const { return unknown_.size(); }
  void Clear() { unknown_.Clear(); }
  Arena* arena() const { return unknown_.arena(); }

 private:
  // Backing store for added metadata.
  ChunkedVector<std::pair<Slice, Slice>, 10> unknown_;
};

}  // namespace metadata_detail

// Helper function for encoders
// Given a metadata trait, convert the value to a slice.
template <typename Which>
absl::enable_if_t<std::is_same<typename Which::ValueType, Slice>::value,
                  const Slice&>
MetadataValueAsSlice(const Slice& slice) {
  return slice;
}

template <typename Which>
absl::enable_if_t<!std::is_same<typename Which::ValueType, Slice>::value, Slice>
MetadataValueAsSlice(typename Which::ValueType value) {
  return Slice(Which::Encode(value));
}

// MetadataMap encodes the mapping of metadata keys to metadata values.
//
// MetadataMap takes a derived class and list of traits. Each of these trait
// objects defines one metadata field that is used by core, and so should have
// more specialized handling than just using the generic APIs.
//
// MetadataMap is the backing type for some derived type via the curiously
// recursive template pattern. This is because many types consumed by
// MetadataMap require the container type to operate on, and many of those
// types are instantiated one per trait. A naive implementation without the
// Derived type would, for traits A,B,C, then instantiate for some
// T<Container, Trait>:
//  - T<MetadataMap<A,B,C>, A>,
//  - T<MetadataMap<A,B,C>, B>,
//  - T<MetadataMap<A,B,C>, C>.
// Since these types ultimately need to be recorded in the .dynstr segment
// for dynamic linkers (if gRPC is linked as a static library) this would
// create O(N^2) bytes of symbols even in stripped libraries. To avoid this
// we use the derived type (e.g. grpc_metadata_batch right now) to capture
// the container type, and we would write T<grpc_metadata_batch, A>, etc...
// Note that now the container type uses a number of bytes that is independent
// of the number of traits, and so we return to a linear symbol table growth
// function.
//
// Each trait object has one of two possible signatures, depending on whether
// that traits field is encodable or not.
// Non-encodable traits are carried in a MetadataMap, but are never passed to
// the application nor serialized to wire.
//
// Encodable traits have the following signature:
// // Traits for the "grpc-xyz" metadata field:
// struct GrpcXyzMetadata {
//   // Can this metadata field be repeated?
//   static constexpr bool kRepeatable = ...;
//   // The type that's stored on MetadataBatch
//   using ValueType = ...;
//   // The type that's stored in compression/decompression tables
//   using MementoType = ...;
//   // The string key for this metadata type (for transports that require it)
//   static absl::string_view key() { return "grpc-xyz"; }
//   // Parse a memento from a slice
//   // Takes ownership of value
//   // Calls fn in the case of an error that should be reported to the user
//   static MementoType ParseMemento(Slice value, MementoParseErrorFn fn) {
//   ...
//   }
//   // Convert a memento to a value
//   static ValueType MementoToValue(MementoType memento) { ... }
//   // Convert a value to its canonical text wire format (the format that
//   // ParseMemento will accept!)
//   static Slice Encode(const ValueType& value);
//   // Convert a value to something that can be passed to StrCat and
//   displayed
//   // for debugging
//   static SomeStrCatableType DisplayValue(MementoType value) { ... }
// };
//
// Non-encodable traits are determined by missing the key() method, and have
// the following signature (and by convention omit the Metadata part of the
// type name):
// // Traits for the GrpcXyz field:
// struct GrpcXyz {
//   // The string key that should be used for debug dumps - should not be a
//   // valid http2 key (ie all lower case)
//   static absl::string_view DebugKey() { return "GRPC_XYZ"; }
//   // Can this metadata field be repeated?
//   static constexpr bool kRepeatable = ...;
//   // The type that's stored on MetadataBatch
//   using ValueType = ...;
//   // Convert a value to something that can be passed to StrCat and
//   displayed
//   // for debugging
//   static SomeStrCatableType DisplayValue(ValueType value) { ... }
// };
//
// About parsing and mementos:
//
// Many gRPC transports exchange metadata as key/value strings, but also allow
// for a more efficient representation as a single integer. We can use this
// integer representation to avoid reparsing too, by storing the parsed value
// in the compression table. This is what mementos are used for.
//
// A trait offers the capability to turn a slice into a memento via
// ParseMemento. This is exposed to users of MetadataMap via the Parse()
// method, that returns a ParsedMetadata object. That ParsedMetadata object
// can in turn be used to set the same value on many different MetadataMaps
// without having to reparse.
//
// Implementation wise, ParsedMetadata is a type erased wrapper around
// MementoType. When we set a value on MetadataMap, we first turn that memento
// into a value. For most types, this is going to be a no-op, but for example
// for grpc-timeout we make the memento the timeout expressed on the wire, but
// we make the value the timestamp of when the timeout will expire (i.e. the
// deadline).
template <class Derived, typename... Traits>
class MetadataMap {
 public:
  explicit MetadataMap(Arena* arena);
  ~MetadataMap();

  MetadataMap(const MetadataMap&) = delete;
  MetadataMap& operator=(const MetadataMap&) = delete;
  MetadataMap(MetadataMap&&) noexcept;
  // We never create MetadataMap directly, instead we create Derived, but we
  // want to be able to move it without redeclaring this.
  // NOLINTNEXTLINE(misc-unconventional-assign-operator)
  Derived& operator=(MetadataMap&&) noexcept;

  // Encode this metadata map into some encoder.
  // For each field that is set in the MetadataMap, call
  // encoder->Encode.
  //
  // For fields for which we have traits, this will be a method with
  // the signature:
  //    void Encode(TraitsType, typename TraitsType::ValueType value);
  // For fields for which we do not have traits, this will be a method
  // with the signature:
  //    void Encode(grpc_mdelem md);
  // TODO(ctiller): It's expected that the latter Encode method will
  // become Encode(Slice, Slice) by the end of the current metadata API
  // transitions.
  template <typename Encoder>
  void Encode(Encoder* encoder) const {
    table_.ForEach(metadata_detail::EncodeWrapper<Encoder>{encoder});
    for (const auto& unk : unknown_) {
      encoder->Encode(unk.first, unk.second);
    }
  }

  // Similar to Encode, but targeted at logging: for each metadatum,
  // call f(key, value) as absl::string_views.
  void Log(metadata_detail::LogFn log_fn) const {
    table_.ForEach(metadata_detail::LogWrapper{log_fn});
    for (const auto& unk : unknown_) {
      log_fn(unk.first.as_string_view(), unk.second.as_string_view());
    }
  }

  std::string DebugString() const {
    metadata_detail::DebugStringBuilder builder;
    Log([&builder](absl::string_view key, absl::string_view value) {
      builder.Add(key, value);
    });
    return builder.TakeOutput();
  }

  // Get the pointer to the value of some known metadata.
  // Returns nullptr if the metadata is not present.
  // Causes a compilation error if Which is not an element of Traits.
  template <typename Which>
  const typename metadata_detail::Value<Which>::StorageType* get_pointer(
      Which) const {
    if (auto* p = table_.template get<Value<Which>>()) return &p->value;
    return nullptr;
  }

  // Get the pointer to the value of some known metadata.
  // Returns nullptr if the metadata is not present.
  // Causes a compilation error if Which is not an element of Traits.
  template <typename Which>
  typename metadata_detail::Value<Which>::StorageType* get_pointer(Which) {
    if (auto* p = table_.template get<Value<Which>>()) return &p->value;
    return nullptr;
  }

  // Get the pointer to the value of some known metadata.
  // Adds the default value for the metadata is not present.
  // Causes a compilation error if Which is not an element of Traits.
  template <typename Which>
  typename metadata_detail::Value<Which>::StorageType* GetOrCreatePointer(
      Which) {
    return &table_.template get_or_create<Value<Which>>()->value;
  }

  // Get the value of some known metadata.
  // Returns nullopt if the metadata is not present.
  // Causes a compilation error if Which is not an element of Traits.
  template <typename Which>
  absl::optional<typename Which::ValueType> get(Which) const {
    if (auto* p = table_.template get<Value<Which>>()) return p->value;
    return absl::nullopt;
  }

  // Set the value of some known metadata.
  // Returns a pointer to the new value.
  template <typename Which, typename... Args>
  absl::enable_if_t<Which::kRepeatable == false, void> Set(Which,
                                                           Args&&... args) {
    table_.template set<Value<Which>>(std::forward<Args>(args)...);
  }
  template <typename Which, typename... Args>
  absl::enable_if_t<Which::kRepeatable == true, void> Set(Which,
                                                          Args&&... args) {
    GetOrCreatePointer(Which())->emplace_back(std::forward<Args>(args)...);
  }

  // Remove a specific piece of known metadata.
  template <typename Which>
  void Remove(Which) {
    table_.template clear<Value<Which>>();
  }

  // Remove some metadata by name
  void Remove(absl::string_view key) {
    metadata_detail::RemoveHelper<Derived> helper(static_cast<Derived*>(this));
    metadata_detail::NameLookup<void, Traits...>::Lookup(key, &helper);
  }

  void Remove(const char* key) { Remove(absl::string_view(key)); }

  // Retrieve some metadata by name
  absl::optional<absl::string_view> GetStringValue(absl::string_view name,
                                                   std::string* buffer) const {
    metadata_detail::GetStringValueHelper<Derived> helper(
        static_cast<const Derived*>(this), buffer);
    return metadata_detail::NameLookup<void, Traits...>::Lookup(name, &helper);
  }

  // Extract a piece of known metadata.
  // Returns nullopt if the metadata was not present, or the value if it was.
  // The same as:
  //  auto value = m.get(T());
  //  m.Remove(T());
  template <typename Which>
  absl::enable_if_t<Which::kRepeatable == false,
                    absl::optional<typename Which::ValueType>>
  Take(Which which) {
    if (auto* p = get_pointer(which)) {
      absl::optional<typename Which::ValueType> value(std::move(*p));
      Remove(which);
      return value;
    }
    return {};
  }

  // Extract repeated known metadata.
  // Returns an empty vector if the metadata was not present.
  template <typename Which>
  absl::enable_if_t<Which::kRepeatable == true,
                    typename metadata_detail::Value<Which>::StorageType>
  Take(Which which) {
    if (auto* p = get_pointer(which)) {
      typename Value<Which>::StorageType value = std::move(*p);
      Remove(which);
      return value;
    }
    return {};
  }

  // Parse metadata from a key/value pair, and return an object representing
  // that result.
  // TODO(ctiller): key should probably be an absl::string_view.
  // Once we don't care about interning anymore, make that change!
  static ParsedMetadata<Derived> Parse(absl::string_view key, Slice value,
                                       uint32_t transport_size,
                                       MetadataParseErrorFn on_error) {
    metadata_detail::ParseHelper<Derived> helper(value.TakeOwned(), on_error,
                                                 transport_size);
    return metadata_detail::NameLookup<void, Traits...>::Lookup(key, &helper);
  }

  // Set a value from a parsed metadata object.
  void Set(const ParsedMetadata<Derived>& m) {
    m.SetOnContainer(static_cast<Derived*>(this));
  }

  // Append a key/value pair - takes ownership of value
  void Append(absl::string_view key, Slice value,
              MetadataParseErrorFn on_error) {
    metadata_detail::AppendHelper<Derived> helper(static_cast<Derived*>(this),
                                                  value.TakeOwned(), on_error);
    metadata_detail::NameLookup<void, Traits...>::Lookup(key, &helper);
  }

  void Clear();
  size_t TransportSize() const;
  Derived Copy() const;
  bool empty() const { return table_.empty() && unknown_.empty(); }
  size_t count() const { return table_.count() + unknown_.size(); }

 private:
  friend class metadata_detail::AppendHelper<Derived>;
  friend class metadata_detail::GetStringValueHelper<Derived>;
  friend class metadata_detail::RemoveHelper<Derived>;
  friend class metadata_detail::CopySink<Derived>;
  friend class ParsedMetadata<Derived>;

  template <typename Which>
  using Value = metadata_detail::Value<Which>;

  // Table of known metadata types.
  Table<Value<Traits>...> table_;
  metadata_detail::UnknownMap unknown_;
};

// Ok/not-ok check for metadata maps that contain GrpcStatusMetadata, so that
// they can be used as result types for TrySeq.
template <typename Derived, typename... Args>
inline bool IsStatusOk(const MetadataMap<Derived, Args...>& m) {
  return m.get(GrpcStatusMetadata()).value_or(GRPC_STATUS_UNKNOWN) ==
         GRPC_STATUS_OK;
}

template <typename Derived, typename... Traits>
MetadataMap<Derived, Traits...>::MetadataMap(Arena* arena) : unknown_(arena) {}

template <typename Derived, typename... Traits>
MetadataMap<Derived, Traits...>::MetadataMap(MetadataMap&& other) noexcept
    : table_(std::move(other.table_)), unknown_(std::move(other.unknown_)) {}

// We never create MetadataMap directly, instead we create Derived, but we
// want to be able to move it without redeclaring this.
// NOLINTNEXTLINE(misc-unconventional-assign-operator)
template <typename Derived, typename... Traits>
Derived& MetadataMap<Derived, Traits...>::operator=(
    MetadataMap&& other) noexcept {
  table_ = std::move(other.table_);
  unknown_ = std::move(other.unknown_);
  return static_cast<Derived&>(*this);
}

template <typename Derived, typename... Traits>
MetadataMap<Derived, Traits...>::~MetadataMap() = default;

template <typename Derived, typename... Traits>
void MetadataMap<Derived, Traits...>::Clear() {
  table_.ClearAll();
  unknown_.Clear();
}

template <typename Derived, typename... Traits>
size_t MetadataMap<Derived, Traits...>::TransportSize() const {
  metadata_detail::TransportSizeEncoder enc;
  Encode(&enc);
  return enc.size();
}

template <typename Derived, typename... Traits>
Derived MetadataMap<Derived, Traits...>::Copy() const {
  Derived out(unknown_.arena());
  metadata_detail::CopySink<Derived> sink(&out);
  Encode(&sink);
  return out;
}

}  // namespace grpc_core

struct grpc_metadata_batch;

using grpc_metadata_batch_base = grpc_core::MetadataMap<
    grpc_metadata_batch,
    // Colon prefixed headers first
    grpc_core::HttpPathMetadata, grpc_core::HttpAuthorityMetadata,
    grpc_core::HttpMethodMetadata, grpc_core::HttpStatusMetadata,
    grpc_core::HttpSchemeMetadata,
    // Non-colon prefixed headers begin here
    grpc_core::ContentTypeMetadata, grpc_core::TeMetadata,
    grpc_core::GrpcEncodingMetadata, grpc_core::GrpcInternalEncodingRequest,
    grpc_core::GrpcAcceptEncodingMetadata, grpc_core::GrpcStatusMetadata,
    grpc_core::GrpcTimeoutMetadata, grpc_core::GrpcPreviousRpcAttemptsMetadata,
    grpc_core::GrpcRetryPushbackMsMetadata, grpc_core::UserAgentMetadata,
    grpc_core::GrpcMessageMetadata, grpc_core::HostMetadata,
    grpc_core::EndpointLoadMetricsBinMetadata,
    grpc_core::GrpcServerStatsBinMetadata, grpc_core::GrpcTraceBinMetadata,
    grpc_core::GrpcTagsBinMetadata, grpc_core::GrpcLbClientStatsMetadata,
    grpc_core::LbCostBinMetadata, grpc_core::LbTokenMetadata,
    // Non-encodable things
    grpc_core::GrpcStreamNetworkState, grpc_core::PeerString,
    grpc_core::GrpcStatusContext>;

struct grpc_metadata_batch : public grpc_metadata_batch_base {
  using grpc_metadata_batch_base::grpc_metadata_batch_base;
};

#endif /* GRPC_CORE_LIB_TRANSPORT_METADATA_BATCH_H */