text: #pragma once
  
  #include <memory>
  
  #include "envoy/grpc/async_client.h"
  #include "envoy/local_info/local_info.h"
  #include "envoy/network/connection.h"
  #include "envoy/service/metrics/v3/metrics_service.pb.h"
  #include "envoy/singleton/instance.h"
  #include "envoy/stats/histogram.h"
  #include "envoy/stats/sink.h"
  #include "envoy/stats/stats.h"
  #include "envoy/upstream/cluster_manager.h"
  
  #include "common/buffer/buffer_impl.h"
  #include "common/grpc/typed_async_client.h"
  
  namespace Envoy {
  namespace Extensions {
  namespace StatSinks {
  namespace MetricsService {
  
  /**
   * Interface for metrics streamer.
   */
  class GrpcMetricsStreamer
      : public Grpc::AsyncStreamCallbacks<envoy::service::metrics::v3::StreamMetricsResponse> {
  public:
    ~GrpcMetricsStreamer() override = default;
  
    /**
     * Send Metrics Message.
     * @param message supplies the metrics to send.
     */
    virtual void send(envoy::service::metrics::v3::StreamMetricsMessage& message) PURE;
  
    // Grpc::AsyncStreamCallbacks
    void onCreateInitialMetadata(Http::RequestHeaderMap&) override {}
    void onReceiveInitialMetadata(Http::ResponseHeaderMapPtr&&) override {}
    void
    onReceiveMessage(std::unique_ptr<envoy::service::metrics::v3::StreamMetricsResponse>&&) override {
    }
    void onReceiveTrailingMetadata(Http::ResponseTrailerMapPtr&&) override {}
    void onRemoteClose(Grpc::Status::GrpcStatus, const std::string&) override{};
  };
  
  using GrpcMetricsStreamerSharedPtr = std::shared_ptr<GrpcMetricsStreamer>;
  
  /**
   * Production implementation of GrpcMetricsStreamer
   */
  class GrpcMetricsStreamerImpl : public Singleton::Instance, public GrpcMetricsStreamer {
  public:
    GrpcMetricsStreamerImpl(Grpc::AsyncClientFactoryPtr&& factory,
                            const LocalInfo::LocalInfo& local_info,
                            envoy::config::core::v3::ApiVersion transport_api_version);
  
    // GrpcMetricsStreamer
    void send(envoy::service::metrics::v3::StreamMetricsMessage& message) override;
  
    // Grpc::AsyncStreamCallbacks
    void onRemoteClose(Grpc::Status::GrpcStatus, const std::string&) override { stream_ = nullptr; }
  
  private:
    Grpc::AsyncStream<envoy::service::metrics::v3::StreamMetricsMessage> stream_{};
    Grpc::AsyncClient<envoy::service::metrics::v3::StreamMetricsMessage,
                      envoy::service::metrics::v3::StreamMetricsResponse>
        client_;
    const LocalInfo::LocalInfo& local_info_;
    const Protobuf::MethodDescriptor& service_method_;
    const envoy::config::core::v3::ApiVersion transport_api_version_;
  };
  
  using GrpcMetricsStreamerImplPtr = std::unique_ptr<GrpcMetricsStreamerImpl>;
  
  /**
   * Stat Sink implementation of Metrics Service.
   */
  class MetricsServiceSink : public Stats::Sink {
  public:
    // MetricsService::Sink
    MetricsServiceSink(const GrpcMetricsStreamerSharedPtr& grpc_metrics_streamer,
                       const bool report_counters_as_deltas);
    void flush(Stats::MetricSnapshot& snapshot) override;
    void onHistogramComplete(const Stats::Histogram&, uint64_t) override {}
  
    void flushCounter(const Stats::MetricSnapshot::CounterSnapshot& counter_snapshot,
                      int64_t snapshot_time_ms);
    void flushGauge(const Stats::Gauge& gauge, int64_t snapshot_time_ms);
    void flushHistogram(const Stats::ParentHistogram& envoy_histogram, int64_t snapshot_time_ms);
  
  private:
    GrpcMetricsStreamerSharedPtr grpc_metrics_streamer_;
    envoy::service::metrics::v3::StreamMetricsMessage message_;
    const bool report_counters_as_deltas_;
  };
  
  } // namespace MetricsService
  } // namespace StatSinks
  } // namespace Extensions
  } // namespace Envoy
  