gRPC streaming client.

include/envoy/grpc/BUILD

@@ -8,6 +8,17 @@ load(
 
 envoy_package()
 
+envoy_cc_library(
+    name = "async_client_interface",
+    hdrs = ["async_client.h"],
+    external_deps = ["protobuf"],
+    deps = [
+        ":status",
+        "//include/envoy/common:optional",
+        "//include/envoy/http:header_map_interface",
+    ],
+)
+
 envoy_cc_library(
     name = "rpc_channel_interface",
     hdrs = ["rpc_channel.h"],

include/envoy/grpc/async_client.h

@@ -0,0 +1,113 @@
+#pragma once
+
+#include <chrono>
+
+#include "envoy/common/optional.h"
+#include "envoy/common/pure.h"
+#include "envoy/grpc/status.h"
+#include "envoy/http/header_map.h"
+
+#include "google/protobuf/descriptor.h"
+
+namespace Envoy {
+namespace Grpc {
+
+/**
+ * An in-flight gRPC stream.
+ */
+template <class RequestType> class AsyncClientStream {
+public:
+  virtual ~AsyncClientStream() {}
+
+  /**
+   * Send request message to the stream.
+   * @param request protobuf serializable message.
+   */
+  virtual void sendMessage(const RequestType& request) PURE;
+
+  /**
+   * Close the stream locally. No further methods may be invoked on the
+   * stream object, but callbacks may still be received until the stream is closed remotely.
+   */
+  virtual void close() PURE;
+
+  /**
+   * Close the stream locally and remotely (as needed). No further methods may be invoked on the
+   * stream object and no further callbacks will be invoked.
+   */
+  virtual void reset() PURE;
+};
+
+/**
+ * Notifies caller of async gRPC stream status.
+ * Note the gRPC stream is full-duplex, even if the local to remote stream has been ended by
+ * AsyncClientStream.close(), AsyncClientCallbacks can continue to receive events until the remote
+ * to local stream is closed (onRemoteClose), and vice versa. Once the stream is closed remotely, no
+ * further callbacks will be invoked.
+ */
+template <class ResponseType> class AsyncClientCallbacks {
+public:
+  virtual ~AsyncClientCallbacks() {}
+
+  /**
+   * Called when populating the headers to send with initial metadata.
+   * @param metadata initial metadata reference.
+   */
+  virtual void onCreateInitialMetadata(Http::HeaderMap& metadata) PURE;
+
+  /**
+   * Called when initial metadata is recevied.
+   * @param metadata initial metadata reference.
+   */
+  virtual void onReceiveInitialMetadata(Http::HeaderMapPtr&& metadata) PURE;
+
+  /**
+   * Called when an async gRPC message is received.
+   * @param response the gRPC message.
+   */
+  virtual void onReceiveMessage(std::unique_ptr<ResponseType>&& message) PURE;
+
+  /**
+   * Called when trailing metadata is recevied.
+   * @param metadata trailing metadata reference.
+   */
+  virtual void onReceiveTrailingMetadata(Http::HeaderMapPtr&& metadata) PURE;
+
+  /**
+   * Called when the remote closes or an error occurs on the gRPC stream. The stream is
+   * considered remotely closed after this invocation and no further callbacks will be
+   * invoked. A non-Ok status implies that stream is also locally closed and that no
+   * further stream operations are permitted.
+   * @param status the gRPC status.
+   */
+  virtual void onRemoteClose(Status::GrpcStatus status) PURE;
+};
+
+/**
+ * Supports sending gRPC requests and receiving responses asynchronously. This can be used to
+ * implement either plain gRPC or streaming gRPC calls.
+ */
+template <class RequestType, class ResponseType> class AsyncClient {
+public:
+  virtual ~AsyncClient() {}
+
+  /**
+   * Start a gRPC stream asynchronously.
+   * @param service_method protobuf descriptor of gRPC service method.
+   * @param callbacks the callbacks to be notified of stream status.
+   * @param timeout supplies the stream timeout, measured since when the frame with end_stream
+   *        flag is sent until when the first frame is received.
+   * @return a stream handle or nullptr if no stream could be started. NOTE: In this case
+   *         onRemoteClose() has already been called inline. The client owns the stream and
+   *         the handle can be used to send more messages or finish the stream. It is expected that
+   *         finish() is invoked by the caller to notify the client that the stream resources may
+   *         be reclaimed.
+   */
+  virtual AsyncClientStream<RequestType>*
+  start(const google::protobuf::MethodDescriptor& service_method,
+        AsyncClientCallbacks<ResponseType>& callbacks,
+        const Optional<std::chrono::milliseconds>& timeout) PURE;
+};
+
+} // Grpc
+} // Envoy

source/common/grpc/BUILD

@@ -8,6 +8,17 @@ load(
 
 envoy_package()
 
+envoy_cc_library(
+    name = "async_client_lib",
+    hdrs = ["async_client_impl.h"],
+    deps = [
+        ":codec_lib",
+        ":common_lib",
+        "//include/envoy/grpc:async_client_interface",
+        "//source/common/http:async_client_lib",
+    ],
+)
+
 envoy_cc_library(
     name = "codec_lib",
     srcs = ["codec.cc"],

source/common/grpc/async_client_impl.h

@@ -0,0 +1,200 @@
+#pragma once
+
+#include "envoy/grpc/async_client.h"
+
+#include "common/common/enum_to_int.h"
+#include "common/common/linked_object.h"
+#include "common/grpc/codec.h"
+#include "common/grpc/common.h"
+#include "common/http/async_client_impl.h"
+#include "common/http/header_map_impl.h"
+#include "common/http/utility.h"
+
+namespace Envoy {
+namespace Grpc {
+
+template <class RequestType, class ResponseType> class AsyncClientStreamImpl;
+
+template <class RequestType, class ResponseType>
+class AsyncClientImpl final : public AsyncClient<RequestType, ResponseType> {
+public:
+  AsyncClientImpl(Upstream::ClusterManager& cm, const std::string& remote_cluster_name)
+      : cm_(cm), remote_cluster_name_(remote_cluster_name) {}
+
+  ~AsyncClientImpl() override { ASSERT(active_streams_.empty()); }
+
+  // Grpc::AsyncClient
+  AsyncClientStream<RequestType>*
+  start(const google::protobuf::MethodDescriptor& service_method,
+        AsyncClientCallbacks<ResponseType>& callbacks,
+        const Optional<std::chrono::milliseconds>& timeout) override {
+    std::unique_ptr<AsyncClientStreamImpl<RequestType, ResponseType>> grpc_stream{
+        new AsyncClientStreamImpl<RequestType, ResponseType>(*this, callbacks)};
+    Http::AsyncClient::Stream* http_stream =
+        cm_.httpAsyncClientForCluster(remote_cluster_name_)
+            .start(*grpc_stream, Optional<std::chrono::milliseconds>(timeout));
+
+    if (http_stream == nullptr) {
+      callbacks.onRemoteClose(Status::GrpcStatus::Unavailable);
+      return nullptr;
+    }
+
+    grpc_stream->set_stream(http_stream);
+
+    Http::MessagePtr message = Common::prepareHeaders(
+        remote_cluster_name_, service_method.service()->full_name(), service_method.name());
+    callbacks.onCreateInitialMetadata(message->headers());
+
+    http_stream->sendHeaders(message->headers(), false);
+    grpc_stream->moveIntoList(std::move(grpc_stream), active_streams_);
+
+    return active_streams_.front().get();
+  }
+
+private:
+  Upstream::ClusterManager& cm_;
+  const std::string remote_cluster_name_;
+  std::list<std::unique_ptr<AsyncClientStreamImpl<RequestType, ResponseType>>> active_streams_;
+
+  friend class AsyncClientStreamImpl<RequestType, ResponseType>;
+};
+
+template <class RequestType, class ResponseType>
+class AsyncClientStreamImpl : public AsyncClientStream<RequestType>,
+                              Http::AsyncClient::StreamCallbacks,
+                              LinkedObject<AsyncClientStreamImpl<RequestType, ResponseType>> {
+public:
+  AsyncClientStreamImpl(AsyncClientImpl<RequestType, ResponseType>& parent,
+                        AsyncClientCallbacks<ResponseType>& callbacks)
+      : parent_(parent), callbacks_(callbacks) {}
+
+  // Http::AsyncClient::StreamCallbacks
+  void onHeaders(Http::HeaderMapPtr&& headers, bool end_stream) override {
+    ASSERT(!remote_closed_);
+    if (Http::Utility::getResponseStatus(*headers) != enumToInt(Http::Code::OK)) {
+      streamError(Status::GrpcStatus::Internal);
+      return;
+    }
+    if (end_stream) {
+      onTrailers(std::move(headers));
+      return;
+    }
+    callbacks_.onReceiveInitialMetadata(std::move(headers));
+  }
+
+  void onData(Buffer::Instance& data, bool end_stream) override {
+    ASSERT(!remote_closed_);
+    std::vector<Frame> frames;
+    if (!decoder_.decode(data, frames)) {
+      streamError(Status::GrpcStatus::Internal);
+      return;
+    }
+
+    for (const auto& frame : frames) {
+      std::unique_ptr<ResponseType> response(new ResponseType());
+      // TODO(htuch): We can avoid linearizing the buffer here when Buffer::Instance implements
+      // protobuf ZeroCopyInputStream.
+      // TODO(htuch): Need to add support for compressed responses as well here.
+      if (frame.flags_ != GRPC_FH_DEFAULT ||
+          !response->ParseFromArray(frame.data_->linearize(frame.data_->length()),
+                                    frame.data_->length())) {
+        streamError(Status::GrpcStatus::Internal);
+        return;
+      }
+      callbacks_.onReceiveMessage(std::move(response));
+    }
+    if (end_stream) {
+      streamError(Status::GrpcStatus::Internal);
+      return;
+    }
+  }
+
+  void onTrailers(Http::HeaderMapPtr&& trailers) override {
+    ASSERT(!remote_closed_);
+    callbacks_.onReceiveTrailingMetadata(std::move(trailers));
+
+    const Optional<Status::GrpcStatus> grpc_status = Common::getGrpcStatus(*trailers);
+    if (!grpc_status.valid()) {
+      streamError(Status::GrpcStatus::Internal);
+      return;
+    }
+    if (grpc_status.value() != Status::GrpcStatus::Ok) {
+      streamError(grpc_status.value());
+      return;
+    }
+
+    callbacks_.onRemoteClose(Status::GrpcStatus::Ok);
+    closeRemote();
+  }
+
+  void onReset() override {
+    if (http_reset_) {
+      return;
+    }
+
+    streamError(Status::GrpcStatus::Internal);
+  }
+
+  // Grpc::AsyncClientStream
+  void sendMessage(const RequestType& request) override {
+    stream_->sendData(*Common::serializeBody(request), false);
+  }
+
+  void close() override { closeLocal(); }
+
+  void reset() override {
+    closeLocal();
+    closeRemote();
+  }
+
+  void set_stream(Http::AsyncClient::Stream* stream) { stream_ = stream; }
+
+private:
+  void streamError(Status::GrpcStatus grpc_status) {
+    callbacks_.onRemoteClose(grpc_status);
+    reset();
+  }
+
+  void cleanup() {
+    if (!http_reset_) {
+      http_reset_ = true;
+      stream_->reset();
+    }
+
+    // This will destroy us, but only do so if we are actually in a list. This does not happen in
+    // the immediate failure case.
+    if (LinkedObject<AsyncClientStreamImpl<RequestType, ResponseType>>::inserted()) {
+      LinkedObject<AsyncClientStreamImpl<RequestType, ResponseType>>::removeFromList(
+          parent_.active_streams_);
+    }
+  }
+
+  void closeLocal() {
+    local_closed_ |= true;
+    if (complete()) {
+      cleanup();
+    }
+  }
+
+  void closeRemote() {
+    remote_closed_ |= true;
+    if (complete()) {
+      cleanup();
+    }
+  }
+
+  bool complete() const { return local_closed_ && remote_closed_; }
+
+  AsyncClientImpl<RequestType, ResponseType>& parent_;
+  AsyncClientCallbacks<ResponseType>& callbacks_;
+  bool local_closed_{};
+  bool remote_closed_{};
+  bool http_reset_{};
+  Http::AsyncClient::Stream* stream_{};
+  Decoder decoder_;
+
+  friend class AsyncClientImpl<RequestType, ResponseType>;
+};
+
+} // namespace Grpc
+} // namespace Envoy

test/common/grpc/BUILD

@@ -8,6 +8,20 @@ load(
 
 envoy_package()
 
+envoy_cc_test(
+    name = "async_client_impl_test",
+    srcs = ["async_client_impl_test.cc"],
+    deps = [
+        "//source/common/grpc:async_client_lib",
+        "//test/mocks/buffer:buffer_mocks",
+        "//test/mocks/grpc:grpc_mocks",
+        "//test/mocks/http:http_mocks",
+        "//test/mocks/upstream:upstream_mocks",
+        "//test/proto:helloworld_proto",
+        "//test/test_common:utility_lib",
+    ],
+)
+
 envoy_cc_test(
     name = "codec_test",
     srcs = ["codec_test.cc"],