Extend AsyncClient to support streaming requests / responses

include/envoy/http/async_client.h

@@ -37,6 +37,39 @@ class AsyncClient {
     virtual void onFailure(FailureReason reason) PURE;
   };
 
+  /**
+   * Notifies caller of async HTTP stream status.
+   */
+  class StreamCallbacks {
+  public:
+    virtual ~StreamCallbacks() {}
+
+    /**
+     * Called when the async HTTP stream receives headers.
+     * @param headers the headers received
+     * @param end_stream whether the response is header only
+     */
+    virtual void onHeaders(HeaderMapPtr&& headers, bool end_stream) PURE;
+
+    /**
+     * Called when the async HTTP stream receives data.
+     * @param data the data received
+     * @param end_stream whether the data is the last data frame
+     */
+    virtual void onData(Buffer::Instance& data, bool end_stream) PURE;
+
+    /**
+     * Called when the async HTTP stream receives trailers.
+     * @param trailers the trailers received.
+     */
+    virtual void onTrailers(HeaderMapPtr&& trailers) PURE;
+
+    /**
+     * Called when the async HTTP stream is reset.
+     */
+    virtual void onResetStream() PURE;
+  };
+
   /**
    * An in-flight HTTP request
    */
@@ -50,6 +83,39 @@ class AsyncClient {
     virtual void cancel() PURE;
   };
 
+  /**
+   * An in-flight HTTP stream
+   */
+  class Stream {
+  public:
+    virtual ~Stream() {}
+
+    /***
+     * Send headers to the stream
+     * @param headers supplies the headers to send
+     * @param end_stream supplies whether this is a header only request
+     */
+    virtual void sendHeaders(HeaderMap& headers, bool end_stream) PURE;
+
+    /***
+     * Send data to the stream
+     * @param data supplies the data to send
+     * @param end_stream supplies whether this is the last data
+     */
+    virtual void sendData(Buffer::Instance& data, bool end_stream) PURE;
+
+    /***
+     * Send trailers. This implicitly ends the stream.
+     * @param trailers supplies the trailers to send
+     */
+    virtual void sendTrailers(HeaderMap& trailers) PURE;
+
+    /***
+     * Close the stream.
+     */
+    virtual void close() PURE;
+  };
+
   virtual ~AsyncClient() {}
 
   /**
@@ -62,6 +128,16 @@ class AsyncClient {
    */
   virtual Request* send(MessagePtr&& request, Callbacks& callbacks,
                         const Optional<std::chrono::milliseconds>& timeout) PURE;
+
+  /**
+   * Start an HTTP stream asynchronously
+   * @param callbacks the callbacks to be notified of stream status.
+   * @return a stream handle or nullptr if no stream could be started. NOTE: In this case
+   *         onResetStream() has already been called inline. The client owns the stream and
+   *         the handle can be used to send more messages or close the stream.
+   */
+  virtual Stream* start(StreamCallbacks& callbacks,
+                        const Optional<std::chrono::milliseconds>& timeout) PURE;
 };
 
 typedef std::unique_ptr<AsyncClient> AsyncClientPtr;

source/common/http/async_client_impl.cc

@@ -1,15 +1,14 @@
 #include "async_client_impl.h"
-#include "headers.h"
 
 namespace Http {
 
 const std::vector<std::reference_wrapper<const Router::RateLimitPolicyEntry>>
-    AsyncRequestImpl::NullRateLimitPolicy::rate_limit_policy_entry_;
-const AsyncRequestImpl::NullRateLimitPolicy AsyncRequestImpl::RouteEntryImpl::rate_limit_policy_;
-const AsyncRequestImpl::NullRetryPolicy AsyncRequestImpl::RouteEntryImpl::retry_policy_;
-const AsyncRequestImpl::NullShadowPolicy AsyncRequestImpl::RouteEntryImpl::shadow_policy_;
-const AsyncRequestImpl::NullVirtualHost AsyncRequestImpl::RouteEntryImpl::virtual_host_;
-const AsyncRequestImpl::NullRateLimitPolicy AsyncRequestImpl::NullVirtualHost::rate_limit_policy_;
+    AsyncStreamImpl::NullRateLimitPolicy::rate_limit_policy_entry_;
+const AsyncStreamImpl::NullRateLimitPolicy AsyncStreamImpl::RouteEntryImpl::rate_limit_policy_;
+const AsyncStreamImpl::NullRetryPolicy AsyncStreamImpl::RouteEntryImpl::retry_policy_;
+const AsyncStreamImpl::NullShadowPolicy AsyncStreamImpl::RouteEntryImpl::shadow_policy_;
+const AsyncStreamImpl::NullVirtualHost AsyncStreamImpl::RouteEntryImpl::virtual_host_;
+const AsyncStreamImpl::NullRateLimitPolicy AsyncStreamImpl::NullVirtualHost::rate_limit_policy_;
 
 AsyncClientImpl::AsyncClientImpl(const Upstream::ClusterInfo& cluster, Stats::Store& stats_store,
                                  Event::Dispatcher& dispatcher,
@@ -29,92 +28,92 @@ AsyncClientImpl::~AsyncClientImpl() {
 
 AsyncClient::Request* AsyncClientImpl::send(MessagePtr&& request, AsyncClient::Callbacks& callbacks,
                                             const Optional<std::chrono::milliseconds>& timeout) {
-  std::unique_ptr<AsyncRequestImpl> new_request{
-      new AsyncRequestImpl(std::move(request), *this, callbacks, timeout)};
+  AsyncRequestImpl* async_request =
+      new AsyncRequestImpl(std::move(request), *this, callbacks, timeout);
+  std::unique_ptr<AsyncStreamImpl> new_request{async_request};
 
   // The request may get immediately failed. If so, we will return nullptr.
   if (!new_request->complete_) {
     new_request->moveIntoList(std::move(new_request), active_requests_);
-    return active_requests_.front().get();
+    return async_request;
   } else {
     return nullptr;
   }
 }
 
-AsyncRequestImpl::AsyncRequestImpl(MessagePtr&& request, AsyncClientImpl& parent,
-                                   AsyncClient::Callbacks& callbacks,
-                                   const Optional<std::chrono::milliseconds>& timeout)
-    : request_(std::move(request)), parent_(parent), callbacks_(callbacks),
-      stream_id_(parent.config_.random_.random()), router_(parent.config_),
-      request_info_(Protocol::Http11), route_(parent_.cluster_.name(), timeout) {
+AsyncClient::Stream* AsyncClientImpl::start(AsyncClient::StreamCallbacks& callbacks,
+                                            const Optional<std::chrono::milliseconds>& timeout) {
+  AsyncStreamImpl* async_request = new AsyncStreamImpl(*this, callbacks, timeout);
+  std::unique_ptr<AsyncStreamImpl> new_request{async_request};
 
-  router_.setDecoderFilterCallbacks(*this);
-  request_->headers().insertEnvoyInternalRequest().value(
-      Headers::get().EnvoyInternalRequestValues.True);
-  request_->headers().insertForwardedFor().value(parent_.config_.local_info_.address());
-  router_.decodeHeaders(request_->headers(), !request_->body());
-  if (!complete_ && request_->body()) {
-    router_.decodeData(*request_->body(), true);
+  // The request may get immediately failed. If so, we will return nullptr.
+  if (!new_request->complete_) {
+    new_request->moveIntoList(std::move(new_request), active_requests_);
+    return async_request;
+  } else {
+    return nullptr;
   }
+}
 
-  // TODO: Support request trailers.
+AsyncStreamImpl::AsyncStreamImpl(AsyncClientImpl& parent, AsyncClient::StreamCallbacks& callbacks,
+                                 const Optional<std::chrono::milliseconds>& timeout)
+    : parent_(parent), stream_callbacks_(callbacks), stream_id_(parent.config_.random_.random()),
+      router_(parent.config_), request_info_(Protocol::Http11),
+      route_(parent_.cluster_.name(), timeout) {
+
+  router_.setDecoderFilterCallbacks(*this);
   // TODO: Correctly set protocol in request info when we support access logging.
 }
 
-AsyncRequestImpl::~AsyncRequestImpl() { ASSERT(!reset_callback_); }
+AsyncStreamImpl::~AsyncStreamImpl() { ASSERT(!reset_callback_); }
 
-void AsyncRequestImpl::encodeHeaders(HeaderMapPtr&& headers, bool end_stream) {
-  response_.reset(new ResponseMessageImpl(std::move(headers)));
+void AsyncStreamImpl::encodeHeaders(HeaderMapPtr&& headers, bool end_stream) {
 #ifndef NDEBUG
   log_debug("async http request response headers (end_stream={}):", end_stream);
-  response_->headers().iterate([](const HeaderEntry& header, void*) -> void {
+  headers->iterate([](const HeaderEntry& header, void*) -> void {
     log_debug("  '{}':'{}'", header.key().c_str(), header.value().c_str());
   }, nullptr);
 #endif
 
-  if (end_stream) {
-    onComplete();
-  }
+  stream_callbacks_.onHeaders(std::move(headers), end_stream);
 }
 
-void AsyncRequestImpl::encodeData(Buffer::Instance& data, bool end_stream) {
+void AsyncStreamImpl::encodeData(Buffer::Instance& data, bool end_stream) {
   log_trace("async http request response data (length={} end_stream={})", data.length(),
             end_stream);
-  if (!response_->body()) {
-    response_->body(Buffer::InstancePtr{new Buffer::OwnedImpl()});
-  }
-  response_->body()->move(data);
-
-  if (end_stream) {
-    onComplete();
-  }
+  stream_callbacks_.onData(data, end_stream);
 }
 
-void AsyncRequestImpl::encodeTrailers(HeaderMapPtr&& trailers) {
-  response_->trailers(std::move(trailers));
+void AsyncStreamImpl::encodeTrailers(HeaderMapPtr&& trailers) {
 #ifndef NDEBUG
   log_debug("async http request response trailers:");
-  response_->trailers()->iterate([](const HeaderEntry& header, void*) -> void {
+  trailers->iterate([](const HeaderEntry& header, void*) -> void {
     log_debug("  '{}':'{}'", header.key().c_str(), header.value().c_str());
   }, nullptr);
 #endif
 
-  onComplete();
+  stream_callbacks_.onTrailers(std::move(trailers));
 }
 
-void AsyncRequestImpl::cancel() {
-  reset_callback_();
-  cleanup();
+void AsyncStreamImpl::sendHeaders(HeaderMap& headers, bool end_stream) {
+  headers.insertEnvoyInternalRequest().value(Headers::get().EnvoyInternalRequestValues.True);
+  headers.insertForwardedFor().value(parent_.config_.local_info_.address());
+  router_.decodeHeaders(headers, end_stream);
+}
+void AsyncStreamImpl::sendData(Buffer::Instance& data, bool end_stream) {
+  router_.decodeData(data, end_stream);
+  decoding_buffer_ = &data;
 }
 
-void AsyncRequestImpl::onComplete() {
-  complete_ = true;
-  callbacks_.onSuccess(std::move(response_));
+void AsyncStreamImpl::sendTrailers(HeaderMap& trailers) { router_.decodeTrailers(trailers); }
+
+void AsyncStreamImpl::close() {
+  reset_callback_();
   cleanup();
 }
 
-void AsyncRequestImpl::cleanup() {
-  response_.reset();
+void AsyncStreamImpl::cleanup() {
+  // response_.reset();
   reset_callback_ = nullptr;
 
   // This will destroy us, but only do so if we are actually in a list. This does not happen in
@@ -124,18 +123,67 @@ void AsyncRequestImpl::cleanup() {
   }
 }
 
-void AsyncRequestImpl::resetStream() {
-  // In this case we don't have a valid response so we do need to raise a failure.
-  callbacks_.onFailure(AsyncClient::FailureReason::Reset);
+void AsyncStreamImpl::resetStream() {
+  stream_callbacks_.onResetStream();
   cleanup();
 }
 
-void AsyncRequestImpl::failDueToClientDestroy() {
+void AsyncStreamImpl::failDueToClientDestroy() {
   // In this case we are going away because the client is being destroyed. We need to both reset
   // the stream as well as raise a failure callback.
   reset_callback_();
-  callbacks_.onFailure(AsyncClient::FailureReason::Reset);
+  resetStream();
+}
+
+AsyncRequestImpl::AsyncRequestImpl(MessagePtr&& request, AsyncClientImpl& parent,
+                                   AsyncClient::Callbacks& callbacks,
+                                   const Optional<std::chrono::milliseconds>& timeout)
+    : AsyncStreamImpl(parent, *this, timeout), request_(std::move(request)), callbacks_(callbacks) {
+
+  sendHeaders(request_->headers(), !request_->body());
+  if (!complete_ && request_->body()) {
+    sendData(*request_->body(), true);
+  }
+  // TODO: Support request trailers.
+}
+
+AsyncRequestImpl::~AsyncRequestImpl() {}
+
+void AsyncRequestImpl::onComplete() {
+  complete_ = true;
+  callbacks_.onSuccess(std::move(response_));
   cleanup();
 }
 
+void AsyncRequestImpl::onHeaders(HeaderMapPtr&& headers, bool end_stream) {
+  response_.reset(new ResponseMessageImpl(std::move(headers)));
+
+  if (end_stream) {
+    onComplete();
+  }
+}
+
+void AsyncRequestImpl::onData(Buffer::Instance& data, bool end_stream) {
+  if (!response_->body()) {
+    response_->body(Buffer::InstancePtr{new Buffer::OwnedImpl()});
+  }
+  response_->body()->move(data);
+
+  if (end_stream) {
+    onComplete();
+  }
+}
+
+void AsyncRequestImpl::onTrailers(HeaderMapPtr&& trailers) {
+  response_->trailers(std::move(trailers));
+  onComplete();
+}
+
+void AsyncRequestImpl::onResetStream() {
+  // In this case we don't have a valid response so we do need to raise a failure.
+  callbacks_.onFailure(AsyncClient::FailureReason::Reset);
+}
+
+void AsyncRequestImpl::cancel() { close(); }
+
 } // Http