Event::Dispatcher& main_thread_dispatcher) {

  return singleton_manager.getTyped<AdminHandler>(

      SINGLETON_MANAGER_REGISTERED_NAME(tap_admin_handler), [&admin, &main_thread_dispatcher] {

        return std::make_shared<AdminHandler>(admin, main_thread_dispatcher);

      });

}

    : admin_(admin.value()), main_thread_dispatcher_(main_thread_dispatcher) {

  const bool rc =

      admin_.addHandler("/tap", "tap filter control", MAKE_ADMIN_HANDLER(handler), true, true);

  RELEASE_ASSERT(rc, "/tap admin endpoint is taken");

  if (admin_.socket().addressType() == Network::Address::Type::Pipe) {

    ENVOY_LOG(warn, "Admin tapping (via /tap) is unreliable when the admin endpoint is a pipe and "

                    "the connection is HTTP/1. Either use an IP address or connect using HTTP/2.");

  }

}

AdminHandler::~AdminHandler() {

  const bool rc = admin_.removeHandler("/tap");

  ASSERT(rc);

}

                                 Server::AdminStream& admin_stream) {

  if (attached_request_ != nullptr) {

    return badRequest(response, "An attached /tap admin stream already exists. Detach it.");

  }

  if (admin_stream.getRequestBody() == nullptr) {

    return badRequest(response, "/tap requires a JSON/YAML body");

  }

  envoy::admin::v3::TapRequest tap_request;

  TRY_NEEDS_AUDIT {

    MessageUtil::loadFromYamlAndValidate(admin_stream.getRequestBody()->toString(), tap_request,

                                         ProtobufMessage::getStrictValidationVisitor());

  }

  END_TRY catch (EnvoyException& e) { return badRequest(response, e.what()); }

  ENVOY_LOG(debug, "tap admin request for config_id={}", tap_request.config_id());

  if (config_id_map_.count(tap_request.config_id()) == 0) {

    return badRequest(

        response, fmt::format("Unknown config id '{}'. No extension has registered with this id.",

                              tap_request.config_id()));

  }

  for (auto config : config_id_map_[tap_request.config_id()]) {

    config->newTapConfig(tap_request.tap_config(), this);

  }

  admin_stream.setEndStreamOnComplete(false);

  admin_stream.addOnDestroyCallback([this] {

    for (auto config : config_id_map_[attached_request_->id()]) {

      ENVOY_LOG(debug, "detach tap admin request for config_id={}", attached_request_->id());

      config->clearTapConfig();

    }

    attached_request_.reset(); // remove ref to attached_request_

  });

  attached_request_ = AttachedRequest::createAttachedRequest(this, tap_request, &admin_stream);

  return Http::Code::OK;

}

Http::Code AdminHandler::badRequest(Buffer::Instance& response, absl::string_view error) {

  ENVOY_LOG(debug, "handler bad request: {}", error);

  response.add(error);

  return Http::Code::BadRequest;

}

void AdminHandler::registerConfig(ExtensionConfig& config, const std::string& config_id) {

  ASSERT(!config_id.empty());

  ASSERT(config_id_map_[config_id].count(&config) == 0);

  config_id_map_[config_id].insert(&config);

  if (attached_request_ != nullptr && attached_request_->id() == config_id) {

    config.newTapConfig(attached_request_->config(), this);

  }

}

void AdminHandler::unregisterConfig(ExtensionConfig& config) {

  ASSERT(!config.adminId().empty());

  std::string admin_id(config.adminId());

  ASSERT(config_id_map_[admin_id].count(&config) == 1);

  config_id_map_[admin_id].erase(&config);

  if (config_id_map_[admin_id].empty()) {

    config_id_map_.erase(admin_id);

  }

}

                                     envoy::config::tap::v3::OutputSink::OutputSinkTypeCase type) {

  UNREFERENCED_PARAMETER(trace_id);

  using ProtoOutputSinkType = envoy::config::tap::v3::OutputSink::OutputSinkTypeCase;

  ASSERT(type == ProtoOutputSinkType::kStreamingAdmin ||

         type == ProtoOutputSinkType::kBufferedAdmin);

  if (type == ProtoOutputSinkType::kStreamingAdmin) {

    return std::make_unique<AdminPerTapSinkHandle>(*this);

  }

  return std::make_unique<BufferedPerTapSinkHandle>(*this);

}

    const std::shared_ptr<envoy::data::tap::v3::TraceWrapper>& trace) {

  if (flushed() || full()) {

  buffer_->emplace_back(std::move(*trace));

}

                                                      envoy::config::tap::v3::OutputSink::Format) {

  ENVOY_LOG(debug, "admin submitting buffered trace to main thread");

  std::shared_ptr<envoy::data::tap::v3::TraceWrapper> shared_trace{std::move(trace)};

  std::weak_ptr<AttachedRequest> weak_attached_request(parent_.attached_request_);

  parent_.main_thread_dispatcher_.post([weak_attached_request, shared_trace] {

    std::shared_ptr<AttachedRequest> attached_request = weak_attached_request.lock();

    if (!attached_request) {

      ENVOY_LOG(debug, "attached request does not exist, not streaming trace");

      return; // No attached request, flushed already

    }

    attached_request->streamMsg(*shared_trace, false);

  });

}

    TraceWrapperPtr&& trace, envoy::config::tap::v3::OutputSink::Format) {

  std::shared_ptr<envoy::data::tap::v3::TraceWrapper> shared_trace(std::move(trace));

  std::weak_ptr<AttachedRequest> weak_attached_request(parent_.attached_request_);

  parent_.main_thread_dispatcher_.post([shared_trace, weak_attached_request] {

    std::shared_ptr<AttachedRequest> attached_request = weak_attached_request.lock();

    if (!attached_request) {

      ENVOY_LOG(debug, "attached request does not exist, not buffering trace");

      return; // No attached request, flushed already.

    }

    TraceBuffer* trace_buffer = attached_request->traceBuffer();

    if (trace_buffer->flushed()) {

      return;

    }

    trace_buffer->bufferTrace(shared_trace);

    if (!trace_buffer->full()) {

      return;

    }

    std::vector<envoy::data::tap::v3::TraceWrapper> buffer = trace_buffer->flush();

    ENVOY_LOG(debug, "admin writing buffered trace list to response");

    for (size_t i = 0; i < buffer.size(); i++) {

      attached_request->streamMsg(buffer[i], (i + 1) == buffer.size());

    }

  });

}

    : config_id_(tap_request.config_id()), config_(tap_request.tap_config()),

      admin_stream_(admin_stream), main_thread_dispatcher_(admin_handler->main_thread_dispatcher_) {

}

    : AttachedRequest(admin_handler, tap_request, admin_stream) {

  const envoy::config::tap::v3::OutputSink& sink =

      tap_request.tap_config().output_config().sinks(0);

  const uint64_t max_buffered_traces = sink.buffered_admin().max_traces();

  const uint64_t timeout_ms = PROTOBUF_GET_MS_OR_DEFAULT(sink.buffered_admin(), timeout, 0);

  trace_buffer_ = std::make_unique<TraceBuffer>(max_buffered_traces);

  if (timeout_ms > 0) {

    timer_ = dispatcher().createTimer(

        [this, admin_handler] { this->onTimeout(admin_handler->attached_request_); });

    timer_->enableTimer(std::chrono::milliseconds(timeout_ms));

  }

}

    const std::weak_ptr<AttachedRequest>& weak_attached_request) {

  dispatcher().post([weak_attached_request] {

    std::shared_ptr<AttachedRequest> attached_request = weak_attached_request.lock();

    if (!attached_request) {

      ENVOY_LOG(debug, "Timer Expiry after admin tap request completion");

      return; // No attached request, flushed already.

    }

    TraceBuffer* trace_buffer = attached_request->traceBuffer();

    if (trace_buffer->flushed()) {

      return;

    }

    std::vector<envoy::data::tap::v3::TraceWrapper> buffer = trace_buffer->flush();

    ENVOY_LOG(debug, "Timer Expiry, admin flushing buffered traces to response");

    for (const auto& trace : buffer) {

      attached_request->streamMsg(trace);

    }

    attached_request->endStream();

  });

}

void AdminHandler::AttachedRequest::endStream() {

  Buffer::OwnedImpl output_buffer;

  stream()->getDecoderFilterCallbacks().encodeData(output_buffer, true);

}

void AdminHandler::AttachedRequest::streamMsg(const Protobuf::Message& message, bool end_stream) {

  std::string output_string;

  switch (format()) {

  case envoy::config::tap::v3::OutputSink::JSON_BODY_AS_STRING:

  case envoy::config::tap::v3::OutputSink::JSON_BODY_AS_BYTES:

    output_string = MessageUtil::getJsonStringFromMessageOrError(message, true, true);

    break;

  case envoy::config::tap::v3::OutputSink::PROTO_BINARY_LENGTH_DELIMITED: {

    Protobuf::io::StringOutputStream stream(&output_string);

    Protobuf::io::CodedOutputStream coded_stream(&stream);

    coded_stream.WriteVarint64(message.ByteSizeLong());

    message.SerializeWithCachedSizes(&coded_stream);

    break;

  }

  }

  Buffer::OwnedImpl output_buffer{output_string};

  stream()->getDecoderFilterCallbacks().encodeData(output_buffer, end_stream);

}

    Server::AdminStream* admin_stream) {

  using ProtoOutputSink = envoy::config::tap::v3::OutputSink;

  const ProtoOutputSink& sink = tap_request.tap_config().output_config().sinks(0);

  switch (sink.output_sink_type_case()) {

  case ProtoOutputSink::kBufferedAdmin:

    return std::make_shared<AttachedRequestBuffered>(admin_handler, tap_request, admin_stream);

  default:

    return std::make_shared<AttachedRequest>(admin_handler, tap_request, admin_stream);

  }

}