async client: refactor Http::AsyncClient::Callbacks to pass in Http::AsyncClient::Request* for correlation between requests and responses

include/envoy/http/async_client.h

@@ -20,6 +20,19 @@ namespace Http {
  */
 class AsyncClient {
 public:
+  /**
+   * An in-flight HTTP request.
+   */
+  class Request {
+  public:
+    virtual ~Request() = default;
+
+    /**
+     * Signals that the request should be cancelled.
+     */
+    virtual void cancel() PURE;
+  };
+
   /**
    * Async Client failure reasons.
    */
@@ -30,11 +43,42 @@ class AsyncClient {
 
   /**
    * Notifies caller of async HTTP request status.
+   *
+   * To support a use case where a caller makes multiple requests in parallel,
+   * individual callback methods provide request context corresponding to that response.
    */
   class Callbacks {
   public:
     virtual ~Callbacks() = default;
 
+    /**
+     * Called when the async HTTP request succeeds.
+     * @param request  request handle or nullptr if no request could be created.
+     *                 NOTE: request handle is passed for correlation purposes only, e.g.
+     *                 for client code to be able to exclude that handle from a list of
+     *                 requests in progress.
+     * @param response the HTTP response
+     */
+    virtual void onRequestSuccess(const Request* request, ResponseMessagePtr&& response) PURE;
+
+    /**
+     * Called when the async HTTP request fails.
+     * @param request request handle or nullptr if no request could be created.
+     *                NOTE: request handle is passed for correlation purposes only, e.g.
+     *                for client code to be able to exclude that handle from a list of
+     *                requests in progress.
+     * @param reason  failure reason
+     */
+    virtual void onRequestFailure(const Request* request, FailureReason reason) PURE;
+  };
+
+  /**
+   * Notifies caller of async HTTP request status.
+   */
+  class RequestCallbacks : public Callbacks {
+  public:
+    virtual ~RequestCallbacks() override = default;
+
     /**
      * Called when the async HTTP request succeeds.
      * @param response the HTTP response
@@ -43,8 +87,17 @@ class AsyncClient {
 
     /**
      * Called when the async HTTP request fails.
+     * @param reason failure reason
      */
     virtual void onFailure(FailureReason reason) PURE;
+
+    // Callbacks
+
+    void onRequestSuccess(const Request*, ResponseMessagePtr&& response) override {
+      onSuccess(std::forward<ResponseMessagePtr>(response));
+    }
+
+    void onRequestFailure(const Request*, FailureReason reason) override { onFailure(reason); }
   };
 
   /**
@@ -92,19 +145,6 @@ class AsyncClient {
     virtual void onReset() PURE;
   };
 
-  /**
-   * An in-flight HTTP request.
-   */
-  class Request {
-  public:
-    virtual ~Request() = default;
-
-    /**
-     * Signals that the request should be cancelled.
-     */
-    virtual void cancel() PURE;
-  };
-
   /**
    * An in-flight HTTP stream.
    */

source/common/config/remote_data_fetcher.h

@@ -42,14 +42,14 @@ class RemoteDataFetcherCallback {
  * Remote data fetcher.
  */
 class RemoteDataFetcher : public Logger::Loggable<Logger::Id::config>,
-                          public Http::AsyncClient::Callbacks {
+                          public Http::AsyncClient::RequestCallbacks {
 public:
   RemoteDataFetcher(Upstream::ClusterManager& cm, const envoy::config::core::v3::HttpUri& uri,
                     const std::string& content_hash, RemoteDataFetcherCallback& callback);
 
   ~RemoteDataFetcher() override;
 
-  // Http::AsyncClient::Callbacks
+  // Http::AsyncClient::RequestCallbacks
   void onSuccess(Http::ResponseMessagePtr&& response) override;
   void onFailure(Http::AsyncClient::FailureReason reason) override;
 

source/common/http/async_client_impl.cc

@@ -267,7 +267,7 @@ void AsyncRequestImpl::onComplete() {
                                                    response_->trailers(), streamInfo(),
                                                    Tracing::EgressConfig::get());
 
-  callbacks_.onSuccess(std::move(response_));
+  callbacks_.onRequestSuccess(this, std::move(response_));
 }
 
 void AsyncRequestImpl::onHeaders(ResponseHeaderMapPtr&& headers, bool) {
@@ -302,7 +302,7 @@ void AsyncRequestImpl::onReset() {
 
   if (!cancelled_) {
     // In this case we don't have a valid response so we do need to raise a failure.
-    callbacks_.onFailure(AsyncClient::FailureReason::Reset);
+    callbacks_.onRequestFailure(this, AsyncClient::FailureReason::Reset);
   }
 }
 

source/common/http/rest_api_fetcher.h

@@ -15,7 +15,7 @@ namespace Http {
  * A helper base class used to fetch a REST API at a jittered periodic interval. Once initialize()
  * is called, the API will be fetched and events raised.
  */
-class RestApiFetcher : public Http::AsyncClient::Callbacks {
+class RestApiFetcher : public Http::AsyncClient::RequestCallbacks {
 protected:
   RestApiFetcher(Upstream::ClusterManager& cm, const std::string& remote_cluster_name,
                  Event::Dispatcher& dispatcher, Runtime::RandomGenerator& random,
@@ -61,7 +61,7 @@ class RestApiFetcher : public Http::AsyncClient::Callbacks {
   void refresh();
   void requestComplete();
 
-  // Http::AsyncClient::Callbacks
+  // Http::AsyncClient::RequestCallbacks
   void onSuccess(Http::ResponseMessagePtr&& response) override;
   void onFailure(Http::AsyncClient::FailureReason reason) override;
 

source/common/router/shadow_writer_impl.h

@@ -15,15 +15,15 @@ namespace Router {
  */
 class ShadowWriterImpl : Logger::Loggable<Logger::Id::router>,
                          public ShadowWriter,
-                         public Http::AsyncClient::Callbacks {
+                         public Http::AsyncClient::RequestCallbacks {
 public:
   ShadowWriterImpl(Upstream::ClusterManager& cm) : cm_(cm) {}
 
   // Router::ShadowWriter
   void shadow(const std::string& cluster, Http::RequestMessagePtr&& request,
               const Http::AsyncClient::RequestOptions& options) override;
 
-  // Http::AsyncClient::Callbacks
+  // Http::AsyncClient::RequestCallbacks
   void onSuccess(Http::ResponseMessagePtr&&) override {}
   void onFailure(Http::AsyncClient::FailureReason) override {}
 

source/extensions/filters/common/ext_authz/ext_authz_http_impl.cc

@@ -220,7 +220,7 @@ void RawHttpClientImpl::cancel() {
 }
 
 // Client
-void RawHttpClientImpl::check(RequestCallbacks& callbacks,
+void RawHttpClientImpl::check(Filters::Common::ExtAuthz::RequestCallbacks& callbacks,
                               const envoy::service::auth::v3::CheckRequest& request,
                               Tracing::Span& parent_span) {
   ASSERT(callbacks_ == nullptr);

source/extensions/filters/common/ext_authz/ext_authz_http_impl.h

@@ -142,7 +142,7 @@ using ClientConfigSharedPtr = std::shared_ptr<ClientConfig>;
  * setting a path prefix witch is not available for gRPC.
  */
 class RawHttpClientImpl : public Client,
-                          public Http::AsyncClient::Callbacks,
+                          public Http::AsyncClient::RequestCallbacks,
                           Logger::Loggable<Logger::Id::config> {
 public:
   explicit RawHttpClientImpl(Upstream::ClusterManager& cm, ClientConfigSharedPtr config,
@@ -151,10 +151,10 @@ class RawHttpClientImpl : public Client,
 
   // ExtAuthz::Client
   void cancel() override;
-  void check(RequestCallbacks& callbacks, const envoy::service::auth::v3::CheckRequest& request,
-             Tracing::Span&) override;
+  void check(Filters::Common::ExtAuthz::RequestCallbacks& callbacks,
+             const envoy::service::auth::v3::CheckRequest& request, Tracing::Span&) override;
 
-  // Http::AsyncClient::Callbacks
+  // Http::AsyncClient::RequestCallbacks
   void onSuccess(Http::ResponseMessagePtr&& message) override;
   void onFailure(Http::AsyncClient::FailureReason reason) override;
 
@@ -163,7 +163,7 @@ class RawHttpClientImpl : public Client,
   Upstream::ClusterManager& cm_;
   ClientConfigSharedPtr config_;
   Http::AsyncClient::Request* request_{};
-  RequestCallbacks* callbacks_{};
+  Filters::Common::ExtAuthz::RequestCallbacks* callbacks_{};
   TimeSource& time_source_;
   Tracing::SpanPtr span_;
 };

source/extensions/filters/http/common/jwks_fetcher.cc

@@ -16,7 +16,7 @@ namespace {
 
 class JwksFetcherImpl : public JwksFetcher,
                         public Logger::Loggable<Logger::Id::filter>,
-                        public Http::AsyncClient::Callbacks {
+                        public Http::AsyncClient::RequestCallbacks {
 public:
   JwksFetcherImpl(Upstream::ClusterManager& cm) : cm_(cm) { ENVOY_LOG(trace, "{}", __func__); }
 

source/extensions/filters/http/lua/lua_filter.h

@@ -78,7 +78,7 @@ class Filter;
  * The script interacts with Envoy entirely through this handle.
  */
 class StreamHandleWrapper : public Filters::Common::Lua::BaseLuaObject<StreamHandleWrapper>,
-                            public Http::AsyncClient::Callbacks {
+                            public Http::AsyncClient::RequestCallbacks {
 public:
   /**
    * The state machine for a stream handler. In the current implementation everything the filter
@@ -252,7 +252,7 @@ class StreamHandleWrapper : public Filters::Common::Lua::BaseLuaObject<StreamHan
     public_key_wrapper_.reset();
   }
 
-  // Http::AsyncClient::Callbacks
+  // Http::AsyncClient::RequestCallbacks
   void onSuccess(Http::ResponseMessagePtr&&) override;
   void onFailure(Http::AsyncClient::FailureReason) override;
 
@@ -280,9 +280,9 @@ class StreamHandleWrapper : public Filters::Common::Lua::BaseLuaObject<StreamHan
 /**
  * An empty Callbacks client. It will ignore everything, including successes and failures.
  */
-class NoopCallbacks : public Http::AsyncClient::Callbacks {
+class NoopCallbacks : public Http::AsyncClient::RequestCallbacks {
 public:
-  // Http::AsyncClient::Callbacks
+  // Http::AsyncClient::RequestCallbacks
   void onSuccess(Http::ResponseMessagePtr&&) override {}
   void onFailure(Http::AsyncClient::FailureReason) override {}
 };

source/extensions/filters/http/squash/squash_filter.h

@@ -56,12 +56,12 @@ class SquashFilterConfig : protected Logger::Loggable<Logger::Id::config> {
 
 using SquashFilterConfigSharedPtr = std::shared_ptr<SquashFilterConfig>;
 
-class AsyncClientCallbackShim : public Http::AsyncClient::Callbacks {
+class AsyncClientCallbackShim : public Http::AsyncClient::RequestCallbacks {
 public:
   AsyncClientCallbackShim(std::function<void(Http::ResponseMessagePtr&&)>&& on_success,
                           std::function<void(Http::AsyncClient::FailureReason)>&& on_fail)
       : on_success_(on_success), on_fail_(on_fail) {}
-  // Http::AsyncClient::Callbacks
+  // Http::AsyncClient::RequestCallbacks
   void onSuccess(Http::ResponseMessagePtr&& m) override {
     on_success_(std::forward<Http::ResponseMessagePtr>(m));
   }

source/extensions/tracers/datadog/datadog_tracer_impl.h

@@ -94,7 +94,7 @@ class Driver : public Common::Ot::OpenTracingDriver {
  * The timer interval can be controlled with the setting
  * tracing.datadog.flush_interval_ms, and defaults to 2000ms.
  */
-class TraceReporter : public Http::AsyncClient::Callbacks,
+class TraceReporter : public Http::AsyncClient::RequestCallbacks,
                       protected Logger::Loggable<Logger::Id::tracing> {
 public:
   /**
@@ -106,7 +106,7 @@ class TraceReporter : public Http::AsyncClient::Callbacks,
    */
   TraceReporter(TraceEncoderSharedPtr encoder, Driver& driver, Event::Dispatcher& dispatcher);
 
-  // Http::AsyncClient::Callbacks.
+  // Http::AsyncClient::RequestCallbacks.
   void onSuccess(Http::ResponseMessagePtr&&) override;
   void onFailure(Http::AsyncClient::FailureReason) override;
 

source/extensions/tracers/lightstep/lightstep_tracer_impl.h

@@ -75,7 +75,8 @@ class LightStepDriver : public Common::Ot::OpenTracingDriver {
   PropagationMode propagationMode() const override { return propagation_mode_; }
 
 private:
-  class LightStepTransporter : public lightstep::AsyncTransporter, Http::AsyncClient::Callbacks {
+  class LightStepTransporter : public lightstep::AsyncTransporter,
+                               Http::AsyncClient::RequestCallbacks {
   public:
     explicit LightStepTransporter(LightStepDriver& driver);
 
@@ -87,7 +88,7 @@ class LightStepDriver : public Common::Ot::OpenTracingDriver {
     void Send(std::unique_ptr<lightstep::BufferChain>&& message,
               Callback& callback) noexcept override;
 
-    // Http::AsyncClient::Callbacks
+    // Http::AsyncClient::RequestCallbacks
     void onSuccess(Http::ResponseMessagePtr&& response) override;
     void onFailure(Http::AsyncClient::FailureReason failure_reason) override;
 

source/extensions/tracers/zipkin/zipkin_tracer_impl.h

@@ -170,7 +170,7 @@ struct CollectorInfo {
  *
  * The default values for the runtime parameters are 5 spans and 5000ms.
  */
-class ReporterImpl : public Reporter, Http::AsyncClient::Callbacks {
+class ReporterImpl : public Reporter, Http::AsyncClient::RequestCallbacks {
 public:
   /**
    * Constructor.
@@ -192,7 +192,7 @@ class ReporterImpl : public Reporter, Http::AsyncClient::Callbacks {
    */
   void reportSpan(Span&& span) override;
 
-  // Http::AsyncClient::Callbacks.
+  // Http::AsyncClient::RequestCallbacks.
   // The callbacks below record Zipkin-span-related stats.
   void onSuccess(Http::ResponseMessagePtr&&) override;
   void onFailure(Http::AsyncClient::FailureReason) override;