Merge branch 'master' into config-isolation

1 files changed, 78 insertions, 18 deletions

diff --git a/include/grpcpp/impl/codegen/server_interface.h b/include/grpcpp/impl/codegen/server_interface.h
index 237991cde6..e0e2629827 100644
--- a/include/grpcpp/impl/codegen/server_interface.h
+++ b/include/grpcpp/impl/codegen/server_interface.h
@@ -21,10 +21,12 @@
 
 #include <grpc/impl/codegen/grpc_types.h>
 #include <grpcpp/impl/codegen/byte_buffer.h>
+#include <grpcpp/impl/codegen/call.h>
 #include <grpcpp/impl/codegen/call_hook.h>
 #include <grpcpp/impl/codegen/completion_queue_tag.h>
 #include <grpcpp/impl/codegen/core_codegen_interface.h>
 #include <grpcpp/impl/codegen/rpc_service_method.h>
+#include <grpcpp/impl/codegen/server_context.h>
 
 namespace grpc {
 
@@ -148,44 +150,69 @@ class ServerInterface : public internal::CallHook {
    public:
     BaseAsyncRequest(ServerInterface* server, ServerContext* context,
                      internal::ServerAsyncStreamingInterface* stream,
-                     CompletionQueue* call_cq, void* tag,
+                     CompletionQueue* call_cq,
+                     ServerCompletionQueue* notification_cq, void* tag,
                      bool delete_on_finalize);
     virtual ~BaseAsyncRequest();
 
     bool FinalizeResult(void** tag, bool* status) override;
 
+   private:
+    void ContinueFinalizeResultAfterInterception();
+
    protected:
     ServerInterface* const server_;
     ServerContext* const context_;
     internal::ServerAsyncStreamingInterface* const stream_;
     CompletionQueue* const call_cq_;
+    ServerCompletionQueue* const notification_cq_;
     void* const tag_;
     const bool delete_on_finalize_;
     grpc_call* call_;
+    internal::Call call_wrapper_;
+    internal::InterceptorBatchMethodsImpl interceptor_methods_;
+    bool done_intercepting_;
   };
 
+  /// RegisteredAsyncRequest is not part of the C++ API
   class RegisteredAsyncRequest : public BaseAsyncRequest {
    public:
     RegisteredAsyncRequest(ServerInterface* server, ServerContext* context,
                            internal::ServerAsyncStreamingInterface* stream,
-                           CompletionQueue* call_cq, void* tag);
-
-    // uses BaseAsyncRequest::FinalizeResult
+                           CompletionQueue* call_cq,
+                           ServerCompletionQueue* notification_cq, void* tag,
+                           const char* name, internal::RpcMethod::RpcType type);
+
+    virtual bool FinalizeResult(void** tag, bool* status) override {
+      /* If we are done intercepting, then there is nothing more for us to do */
+      if (done_intercepting_) {
+        return BaseAsyncRequest::FinalizeResult(tag, status);
+      }
+      call_wrapper_ = internal::Call(
+          call_, server_, call_cq_, server_->max_receive_message_size(),
+          context_->set_server_rpc_info(name_, type_,
+                                        *server_->interceptor_creators()));
+      return BaseAsyncRequest::FinalizeResult(tag, status);
+    }
 
    protected:
     void IssueRequest(void* registered_method, grpc_byte_buffer** payload,
                       ServerCompletionQueue* notification_cq);
+    const char* name_;
+    const internal::RpcMethod::RpcType type_;
   };
 
   class NoPayloadAsyncRequest final : public RegisteredAsyncRequest {
    public:
-    NoPayloadAsyncRequest(void* registered_method, ServerInterface* server,
-                          ServerContext* context,
+    NoPayloadAsyncRequest(internal::RpcServiceMethod* registered_method,
+                          ServerInterface* server, ServerContext* context,
                           internal::ServerAsyncStreamingInterface* stream,
                           CompletionQueue* call_cq,
                           ServerCompletionQueue* notification_cq, void* tag)
-        : RegisteredAsyncRequest(server, context, stream, call_cq, tag) {
-      IssueRequest(registered_method, nullptr, notification_cq);
+        : RegisteredAsyncRequest(
+              server, context, stream, call_cq, notification_cq, tag,
+              registered_method->name(), registered_method->method_type()) {
+      IssueRequest(registered_method->server_tag(), nullptr, notification_cq);
     }
 
     // uses RegisteredAsyncRequest::FinalizeResult
@@ -194,13 +221,15 @@ class ServerInterface : public internal::CallHook {
   template <class Message>
   class PayloadAsyncRequest final : public RegisteredAsyncRequest {
    public:
-    PayloadAsyncRequest(void* registered_method, ServerInterface* server,
-                        ServerContext* context,
+    PayloadAsyncRequest(internal::RpcServiceMethod* registered_method,
+                        ServerInterface* server, ServerContext* context,
                         internal::ServerAsyncStreamingInterface* stream,
                         CompletionQueue* call_cq,
                         ServerCompletionQueue* notification_cq, void* tag,
                         Message* request)
-        : RegisteredAsyncRequest(server, context, stream, call_cq, tag),
+        : RegisteredAsyncRequest(
+              server, context, stream, call_cq, notification_cq, tag,
+              registered_method->name(), registered_method->method_type()),
           registered_method_(registered_method),
           server_(server),
           context_(context),
@@ -209,7 +238,8 @@ class ServerInterface : public internal::CallHook {
           notification_cq_(notification_cq),
           tag_(tag),
           request_(request) {
-      IssueRequest(registered_method, payload_.bbuf_ptr(), notification_cq);
+      IssueRequest(registered_method->server_tag(), payload_.bbuf_ptr(),
+                   notification_cq);
     }
 
     ~PayloadAsyncRequest() {
@@ -217,6 +247,10 @@ class ServerInterface : public internal::CallHook {
     }
 
     bool FinalizeResult(void** tag, bool* status) override {
+      /* If we are done intercepting, then there is nothing more for us to do */
+      if (done_intercepting_) {
+        return RegisteredAsyncRequest::FinalizeResult(tag, status);
+      }
       if (*status) {
         if (!payload_.Valid() || !SerializationTraits<Message>::Deserialize(
                                       payload_.bbuf_ptr(), request_)
@@ -235,15 +269,20 @@ class ServerInterface : public internal::CallHook {
           return false;
         }
       }
+      /* Set interception point for recv message */
+      interceptor_methods_.AddInterceptionHookPoint(
+          experimental::InterceptionHookPoints::POST_RECV_MESSAGE);
+      interceptor_methods_.SetRecvMessage(request_);
       return RegisteredAsyncRequest::FinalizeResult(tag, status);
     }
 
    private:
-    void* const registered_method_;
+    internal::RpcServiceMethod* const registered_method_;
     ServerInterface* const server_;
     ServerContext* const context_;
     internal::ServerAsyncStreamingInterface* const stream_;
     CompletionQueue* const call_cq_;
+
     ServerCompletionQueue* const notification_cq_;
     void* const tag_;
     Message* const request_;
@@ -272,9 +311,8 @@ class ServerInterface : public internal::CallHook {
                         ServerCompletionQueue* notification_cq, void* tag,
                         Message* message) {
     GPR_CODEGEN_ASSERT(method);
-    new PayloadAsyncRequest<Message>(method->server_tag(), this, context,
-                                     stream, call_cq, notification_cq, tag,
-                                     message);
+    new PayloadAsyncRequest<Message>(method, this, context, stream, call_cq,
+                                     notification_cq, tag, message);
   }
 
   void RequestAsyncCall(internal::RpcServiceMethod* method,
@@ -283,8 +321,8 @@ class ServerInterface : public internal::CallHook {
                         CompletionQueue* call_cq,
                         ServerCompletionQueue* notification_cq, void* tag) {
     GPR_CODEGEN_ASSERT(method);
-    new NoPayloadAsyncRequest(method->server_tag(), this, context, stream,
-                              call_cq, notification_cq, tag);
+    new NoPayloadAsyncRequest(method, this, context, stream, call_cq,
+                              notification_cq, tag);
   }
 
   void RequestAsyncGenericCall(GenericServerContext* context,
@@ -295,6 +333,28 @@ class ServerInterface : public internal::CallHook {
     new GenericAsyncRequest(this, context, stream, call_cq, notification_cq,
                             tag, true);
   }
+
+ private:
+  // EXPERIMENTAL
+  // Getter method for the vector of interceptor factory objects.
+  // Returns a nullptr (rather than being pure) since this is a post-1.0 method
+  // and adding a new pure method to an interface would be a breaking change
+  // (even though this is private and non-API)
+  virtual std::vector<
+      std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>*
+  interceptor_creators() {
+    return nullptr;
+  }
+
+  // EXPERIMENTAL
+  // A method to get the callbackable completion queue associated with this
+  // server. If the return value is nullptr, this server doesn't support
+  // callback operations.
+  // TODO(vjpai): Consider a better default like using a global CQ
+  // Returns nullptr (rather than being pure) since this is a post-1.0 method
+  // and adding a new pure method to an interface would be a breaking change
+  // (even though this is private and non-API)
+  virtual CompletionQueue* CallbackCQ() { return nullptr; }
 };
 
 }  // namespace grpc