grpc: add a raw (Buffer::Instance) interface

include/envoy/grpc/BUILD

@@ -14,8 +14,10 @@ envoy_cc_library(
     external_deps = ["abseil_optional"],
     deps = [
         ":status",
+        "//include/envoy/buffer:buffer_interface",
         "//include/envoy/http:header_map_interface",
         "//include/envoy/tracing:http_tracer_interface",
+        "//source/common/common:assert_lib",
         "//source/common/protobuf",
     ],
 )

include/envoy/grpc/async_client.h

@@ -2,11 +2,13 @@
 
 #include <chrono>
 
+#include "envoy/buffer/buffer.h"
 #include "envoy/common/pure.h"
 #include "envoy/grpc/status.h"
 #include "envoy/http/header_map.h"
 #include "envoy/tracing/http_tracer.h"
 
+#include "common/common/assert.h"
 #include "common/protobuf/protobuf.h"
 
 #include "absl/types/optional.h"
@@ -36,12 +38,12 @@ class AsyncStream {
 
   /**
    * Send request message to the stream.
-   * @param request protobuf serializable message.
+   * @param request serializalized message.
    * @param end_stream 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 sendMessage(const Protobuf::Message& request, bool end_stream) PURE;
+  virtual void sendMessage(Buffer::InstancePtr&& request, bool end_stream) PURE;
 
   /**
    * Close the stream locally and send an empty DATA frame to the remote. No further methods may be
@@ -55,6 +57,11 @@ class AsyncStream {
    * stream object and no further callbacks will be invoked.
    */
   virtual void resetStream() PURE;
+
+  /**
+   * @return true if the underlying protocol requires messages to be serialized with a gRPC header.
+   */
+  virtual bool isGrpcHeaderRequired() PURE;
 };
 
 class AsyncRequestCallbacks {
@@ -67,19 +74,12 @@ class AsyncRequestCallbacks {
    */
   virtual void onCreateInitialMetadata(Http::HeaderMap& metadata) PURE;
 
-  /**
-   * Factory for empty response messages.
-   * @return ProtobufTypes::MessagePtr a Protobuf::Message with the response
-   *         type for the request.
-   */
-  virtual ProtobufTypes::MessagePtr createEmptyResponse() PURE;
-
   /**
    * Called when the async gRPC request succeeds. No further callbacks will be invoked.
-   * @param response the gRPC response.
+   * @param response the gRPC response bytes.
    * @param span a tracing span to fill with extra tags.
    */
-  virtual void onSuccessUntyped(ProtobufTypes::MessagePtr&& response, Tracing::Span& span) PURE;
+  virtual void onSuccess(Buffer::InstancePtr&& response, Tracing::Span& span) PURE;
 
   /**
    * Called when the async gRPC request fails. No further callbacks will be invoked.
@@ -91,20 +91,6 @@ class AsyncRequestCallbacks {
                          Tracing::Span& span) PURE;
 };
 
-// Templatized variant of AsyncRequestCallbacks.
-template <class ResponseType> class TypedAsyncRequestCallbacks : public AsyncRequestCallbacks {
-public:
-  ProtobufTypes::MessagePtr createEmptyResponse() override {
-    return std::make_unique<ResponseType>();
-  }
-
-  virtual void onSuccess(std::unique_ptr<ResponseType>&& response, Tracing::Span& span) PURE;
-
-  void onSuccessUntyped(ProtobufTypes::MessagePtr&& response, Tracing::Span& span) override {
-    onSuccess(std::unique_ptr<ResponseType>(dynamic_cast<ResponseType*>(response.release())), span);
-  }
-};
-
 /**
  * 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
@@ -116,13 +102,6 @@ class AsyncStreamCallbacks {
 public:
   virtual ~AsyncStreamCallbacks() {}
 
-  /**
-   * Factory for empty response messages.
-   * @return ProtobufTypes::MessagePtr a Protobuf::Message with the response
-   *          type for the stream.
-   */
-  virtual ProtobufTypes::MessagePtr createEmptyResponse() PURE;
-
   /**
    * Called when populating the headers to send with initial metadata.
    * @param metadata initial metadata reference.
@@ -139,8 +118,10 @@ class AsyncStreamCallbacks {
   /**
    * Called when an async gRPC message is received.
    * @param response the gRPC message.
+   * @return bool which is true if the message well formed and false otherwise which will cause
+              the stream to shutdown with an INTERNAL error.
    */
-  virtual void onReceiveMessageUntyped(ProtobufTypes::MessagePtr&& message) PURE;
+  virtual bool onReceiveMessage(Buffer::InstancePtr&& response) PURE;
 
   /**
    * Called when trailing metadata is received. This will also be called on non-Ok grpc-status
@@ -152,27 +133,13 @@ class AsyncStreamCallbacks {
   /**
    * 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. In addition, no further stream operations are permitted.
+  { * invoked. In addition, no further stream operations are permitted.
    * @param status the gRPC status.
    * @param message the gRPC status message or empty string if not present.
    */
   virtual void onRemoteClose(Status::GrpcStatus status, const std::string& message) PURE;
 };
 
-// Templatized variant of AsyncStreamCallbacks.
-template <class ResponseType> class TypedAsyncStreamCallbacks : public AsyncStreamCallbacks {
-public:
-  ProtobufTypes::MessagePtr createEmptyResponse() override {
-    return std::make_unique<ResponseType>();
-  }
-
-  virtual void onReceiveMessage(std::unique_ptr<ResponseType>&& message) PURE;
-
-  void onReceiveMessageUntyped(ProtobufTypes::MessagePtr&& message) override {
-    onReceiveMessage(std::unique_ptr<ResponseType>(dynamic_cast<ResponseType*>(message.release())));
-  }
-};
-
 /**
  * Supports sending gRPC requests and receiving responses asynchronously. This can be used to
  * implement either plain gRPC or streaming gRPC calls.
@@ -183,33 +150,40 @@ class AsyncClient {
 
   /**
    * Start a gRPC unary RPC asynchronously.
-   * @param service_method protobuf descriptor of gRPC service method.
-   * @param request protobuf serializable message.
+   * @param service_full_name full name of the service (i.e. service_method.service()->full_name()).
+   * @param method_name name of the method (i.e. service_method.name()).
+   * @param request serialized message.
    * @param callbacks the callbacks to be notified of RPC status.
    * @param parent_span the current parent tracing context.
    * @param timeout supplies the request timeout.
    * @return a request handle or nullptr if no request could be started. NOTE: In this case
    *         onFailure() has already been called inline. The client owns the request and the
    *         handle should just be used to cancel.
    */
-  virtual AsyncRequest* send(const Protobuf::MethodDescriptor& service_method,
-                             const Protobuf::Message& request, AsyncRequestCallbacks& callbacks,
+  virtual AsyncRequest* send(absl::string_view service_full_name, absl::string_view method_name,
+                             Buffer::InstancePtr&& request, AsyncRequestCallbacks& callbacks,
                              Tracing::Span& parent_span,
                              const absl::optional<std::chrono::milliseconds>& timeout) PURE;
 
   /**
    * Start a gRPC stream asynchronously.
    * TODO(mattklein123): Determine if tracing should be added to streaming requests.
-   * @param service_method protobuf descriptor of gRPC service method.
+   * @param service_full_name full name of the service (i.e. service_method.service()->full_name()).
+   * @param method_name name of the method (i.e. service_method.name()).
    * @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
    *         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
    *         closeStream() is invoked by the caller to notify the client that the stream resources
    *         may be reclaimed.
    */
-  virtual AsyncStream* start(const Protobuf::MethodDescriptor& service_method,
+  virtual AsyncStream* start(absl::string_view service_full_name, absl::string_view method_name,
                              AsyncStreamCallbacks& callbacks) PURE;
+
+  /**
+   * @return true if the underlying protocol requires messages to be serialized with a gRPC header.
+   */
+  virtual bool isGrpcHeaderRequired() PURE;
 };
 
 typedef std::unique_ptr<AsyncClient> AsyncClientPtr;

source/common/buffer/buffer_impl.cc

@@ -130,8 +130,10 @@ void OwnedImpl::commit(RawSlice* iovecs, uint64_t num_iovecs) {
       slice_index--;
     }
     if (slice_index < 0) {
-      // There was no slice containing any data, so rewind the iterator at the first slice.
       slice_index = 0;
+      if (!slices_[0]) {
+        return;
+      }
     }
 
     // Next, scan forward and attempt to match the slices against iovecs.

source/common/config/BUILD

@@ -125,6 +125,7 @@ envoy_cc_library(
         "//source/common/common:backoff_lib",
         "//source/common/common:minimal_logger_lib",
         "//source/common/common:token_bucket_impl_lib",
+        "//source/common/grpc:async_client_lib",
         "//source/common/protobuf",
     ],
 )

source/common/config/grpc_stream.h

@@ -8,6 +8,7 @@
 #include "common/common/backoff_strategy.h"
 #include "common/common/token_bucket_impl.h"
 #include "common/config/utility.h"
+#include "common/grpc/typed_async_client.h"
 
 namespace Envoy {
 namespace Config {
@@ -59,7 +60,7 @@ class GrpcStream : public Grpc::TypedAsyncStreamCallbacks<ResponseProto>,
 
   void establishNewStream() {
     ENVOY_LOG(debug, "Establishing new gRPC bidi stream for {}", service_method_.DebugString());
-    stream_ = async_client_->start(service_method_, *this);
+    stream_ = async_client_->startTyped(service_method_, *this);
     if (stream_ == nullptr) {
       ENVOY_LOG(warn, "Unable to establish new stream");
       handleEstablishmentFailure();
@@ -72,7 +73,7 @@ class GrpcStream : public Grpc::TypedAsyncStreamCallbacks<ResponseProto>,
 
   bool grpcStreamAvailable() const { return stream_ != nullptr; }
 
-  void sendMessage(const RequestProto& request) { stream_->sendMessage(request, false); }
+  void sendMessage(const RequestProto& request) { stream_->sendMessageTyped(request, false); }
 
   // Grpc::AsyncStreamCallbacks
   void onCreateInitialMetadata(Http::HeaderMap& metadata) override {
@@ -83,7 +84,7 @@ class GrpcStream : public Grpc::TypedAsyncStreamCallbacks<ResponseProto>,
     UNREFERENCED_PARAMETER(metadata);
   }
 
-  void onReceiveMessage(std::unique_ptr<ResponseProto>&& message) override {
+  void onReceiveMessageTyped(std::unique_ptr<ResponseProto>&& message) override {
     // Reset here so that it starts with fresh backoff interval on next disconnect.
     backoff_strategy_->reset();
     // Some times during hot restarts this stat's value becomes inconsistent and will continue to
@@ -150,8 +151,8 @@ class GrpcStream : public Grpc::TypedAsyncStreamCallbacks<ResponseProto>,
   const uint32_t RETRY_INITIAL_DELAY_MS = 500;
   const uint32_t RETRY_MAX_DELAY_MS = 30000; // Do not cross more than 30s
 
-  Grpc::AsyncClientPtr async_client_;
-  Grpc::AsyncStream* stream_{};
+  Grpc::TypedAsyncClient<RequestProto, ResponseProto> async_client_;
+  Grpc::TypedAsyncStream<RequestProto> stream_{};
   const Protobuf::MethodDescriptor& service_method_;
   ControlPlaneStats control_plane_stats_;
 

source/common/grpc/BUILD

@@ -9,13 +9,27 @@ load(
 
 envoy_package()
 
+envoy_cc_library(
+    name = "typed_async_client_lib",
+    srcs = ["typed_async_client.cc"],
+    hdrs = ["typed_async_client.h"],
+    deps = [
+        ":codec_lib",
+        ":common_lib",
+        "//include/envoy/grpc:async_client_interface",
+        "//source/common/buffer:zero_copy_input_stream_lib",
+        "//source/common/http:async_client_lib",
+    ],
+)
+
 envoy_cc_library(
     name = "async_client_lib",
     srcs = ["async_client_impl.cc"],
     hdrs = ["async_client_impl.h"],
     deps = [
         ":codec_lib",
         ":common_lib",
+        ":typed_async_client_lib",
         "//include/envoy/grpc:async_client_interface",
         "//source/common/buffer:zero_copy_input_stream_lib",
         "//source/common/http:async_client_lib",
@@ -28,6 +42,7 @@ envoy_cc_library(
     hdrs = ["async_client_manager_impl.h"],
     deps = [
         ":async_client_lib",
+        ":common_lib",
         "//include/envoy/grpc:async_client_manager_interface",
         "//include/envoy/singleton:manager_interface",
         "//include/envoy/thread_local:thread_local_interface",
@@ -61,7 +76,10 @@ envoy_cc_library(
     name = "common_lib",
     srcs = ["common.cc"],
     hdrs = ["common.h"],
-    external_deps = ["abseil_optional"],
+    external_deps = [
+        "abseil_optional",
+        "grpc",
+    ],
     deps = [
         "//include/envoy/http:header_map_interface",
         "//include/envoy/http:message_interface",
@@ -91,7 +109,10 @@ envoy_cc_library(
         "grpc",
     ],
     deps = [
+        ":async_client_lib",
+        ":common_lib",
         ":google_grpc_creds_lib",
+        ":typed_async_client_lib",
         "//include/envoy/api:api_interface",
         "//include/envoy/grpc:google_grpc_creds_interface",
         "//include/envoy/thread:thread_interface",