Merge pull request #11454 from markdroth/server_cq_ok_semantics

Do not return calls on server when request proto fails to deserialize.

3 files changed, 42 insertions, 7 deletions

diff --git a/include/grpc++/impl/codegen/core_codegen.h b/include/grpc++/impl/codegen/core_codegen.h
index 504c79cff9..2b15a01845 100644
--- a/include/grpc++/impl/codegen/core_codegen.h
+++ b/include/grpc++/impl/codegen/core_codegen.h
@@ -60,6 +60,10 @@ class CoreCodegen final : public CoreCodegenInterface {
   void gpr_cv_signal(gpr_cv* cv) override;
   void gpr_cv_broadcast(gpr_cv* cv) override;
 
+  grpc_call_error grpc_call_cancel_with_status(grpc_call* call,
+                                               grpc_status_code status,
+                                               const char* description,
+                                               void* reserved) override;
   void grpc_call_ref(grpc_call* call) override;
   void grpc_call_unref(grpc_call* call) override;
   virtual void* grpc_call_arena_alloc(grpc_call* call, size_t length) override;
diff --git a/include/grpc++/impl/codegen/core_codegen_interface.h b/include/grpc++/impl/codegen/core_codegen_interface.h
index 930b116dda..b4c771ac93 100644
--- a/include/grpc++/impl/codegen/core_codegen_interface.h
+++ b/include/grpc++/impl/codegen/core_codegen_interface.h
@@ -89,6 +89,10 @@ class CoreCodegenInterface {
   virtual grpc_slice grpc_slice_new_with_user_data(void* p, size_t len,
                                                    void (*destroy)(void*),
                                                    void* user_data) = 0;
+  virtual grpc_call_error grpc_call_cancel_with_status(grpc_call* call,
+                                                       grpc_status_code status,
+                                                       const char* description,
+                                                       void* reserved) = 0;
   virtual void grpc_call_ref(grpc_call* call) = 0;
   virtual void grpc_call_unref(grpc_call* call) = 0;
   virtual void* grpc_call_arena_alloc(grpc_call* call, size_t length) = 0;
diff --git a/include/grpc++/impl/codegen/server_interface.h b/include/grpc++/impl/codegen/server_interface.h
index cc9bf0a108..3bcf4c87e7 100644
--- a/include/grpc++/impl/codegen/server_interface.h
+++ b/include/grpc++/impl/codegen/server_interface.h
@@ -177,22 +177,49 @@ class ServerInterface : public internal::CallHook {
                         ServerCompletionQueue* notification_cq, void* tag,
                         Message* request)
         : RegisteredAsyncRequest(server, context, stream, call_cq, tag),
+          registered_method_(registered_method),
+          server_(server),
+          context_(context),
+          stream_(stream),
+          call_cq_(call_cq),
+          notification_cq_(notification_cq),
+          tag_(tag),
           request_(request) {
       IssueRequest(registered_method, &payload_, notification_cq);
     }
 
     bool FinalizeResult(void** tag, bool* status) override {
-      bool serialization_status =
-          *status && payload_ &&
-          SerializationTraits<Message>::Deserialize(payload_, request_).ok();
-      bool ret = RegisteredAsyncRequest::FinalizeResult(tag, status);
-      *status = serialization_status && *status;
-      return ret;
+      if (*status) {
+        if (payload_ == nullptr ||
+            !SerializationTraits<Message>::Deserialize(payload_, request_)
+                 .ok()) {
+          // If deserialization fails, we cancel the call and instantiate
+          // a new instance of ourselves to request another call.  We then
+          // return false, which prevents the call from being returned to
+          // the application.
+          g_core_codegen_interface->grpc_call_cancel_with_status(
+              call_, GRPC_STATUS_INTERNAL, "Unable to parse request", nullptr);
+          g_core_codegen_interface->grpc_call_unref(call_);
+          new PayloadAsyncRequest(registered_method_, server_, context_,
+                                  stream_, call_cq_, notification_cq_, tag_,
+                                  request_);
+          delete this;
+          return false;
+        }
+      }
+      return RegisteredAsyncRequest::FinalizeResult(tag, status);
     }
 
    private:
-    grpc_byte_buffer* payload_;
+    void* 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_;
+    grpc_byte_buffer* payload_;
   };
 
   class GenericAsyncRequest : public BaseAsyncRequest {