Server side cancellation receive support

3 files changed, 7 insertions, 27 deletions

diff --git a/include/grpc++/completion_queue.h b/include/grpc++/completion_queue.h
index 9a4fa9f2e1..cec0ef0a8d 100644
--- a/include/grpc++/completion_queue.h
+++ b/include/grpc++/completion_queue.h
@@ -114,7 +114,7 @@ class CompletionQueue {
   bool Pluck(CompletionQueueTag *tag);
 
   // Does a single polling pluck on tag
-  void TryPluck(CompletionQueueTag *tag);
+  void TryPluck(CompletionQueueTag *tag, bool forever);
 
   grpc_completion_queue *cq_;  // owned
 };
diff --git a/include/grpc++/server_context.h b/include/grpc++/server_context.h
index e2e14d9ef7..81dcb21d13 100644
--- a/include/grpc++/server_context.h
+++ b/include/grpc++/server_context.h
@@ -34,8 +34,6 @@
 #ifndef __GRPCPP_SERVER_CONTEXT_H_
 #define __GRPCPP_SERVER_CONTEXT_H_
 
-#include <grpc++/completion_queue.h>
-
 #include <chrono>
 #include <map>
 #include <mutex>
@@ -63,7 +61,9 @@ class ServerWriter;
 template <class R, class W>
 class ServerReaderWriter;
 
+class Call;
 class CallOpBuffer;
+class CompletionQueue;
 class Server;
 
 // Interface of server side rpc context.
@@ -79,7 +79,7 @@ class ServerContext final {
   void AddInitialMetadata(const grpc::string& key, const grpc::string& value);
   void AddTrailingMetadata(const grpc::string& key, const grpc::string& value);
 
-  bool IsCancelled() { return completion_op_.CheckCancelled(cq_); }
+  bool IsCancelled();
 
   std::multimap<grpc::string, grpc::string> client_metadata() {
     return client_metadata_;
@@ -102,22 +102,14 @@ class ServerContext final {
   template <class R, class W>
   friend class ::grpc::ServerReaderWriter;
 
-  class CompletionOp final : public CompletionQueueTag {
-   public:
-    bool FinalizeResult(void** tag, bool* status) override;
-
-    bool CheckCancelled(CompletionQueue* cq);
+  class CompletionOp;
 
-   private:
-    std::mutex mu_;
-    bool finalized_ = false;
-    int cancelled_ = 0;
-  };
+  void BeginCompletionOp(Call* call);
 
   ServerContext(gpr_timespec deadline, grpc_metadata* metadata,
                 size_t metadata_count);
 
-  CompletionOp completion_op_;
+  CompletionOp* completion_op_ = nullptr;
 
   std::chrono::system_clock::time_point deadline_;
   grpc_call* call_ = nullptr;
diff --git a/include/grpc++/stream.h b/include/grpc++/stream.h
index 20ba3fb790..a37062b42d 100644
--- a/include/grpc++/stream.h
+++ b/include/grpc++/stream.h
@@ -576,8 +576,6 @@ class ServerAsyncResponseWriter final : public ServerAsyncStreamingInterface {
     if (status.IsOk()) {
       finish_buf_.AddSendMessage(msg);
     }
-    bool cancelled = false;
-    finish_buf_.AddServerRecvClose(&cancelled);
     finish_buf_.AddServerSendStatus(&ctx_->trailing_metadata_, status);
     call_.PerformOps(&finish_buf_);
   }
@@ -589,8 +587,6 @@ class ServerAsyncResponseWriter final : public ServerAsyncStreamingInterface {
       finish_buf_.AddSendInitialMetadata(&ctx_->initial_metadata_);
       ctx_->sent_initial_metadata_ = true;
     }
-    bool cancelled = false;
-    finish_buf_.AddServerRecvClose(&cancelled);
     finish_buf_.AddServerSendStatus(&ctx_->trailing_metadata_, status);
     call_.PerformOps(&finish_buf_);
   }
@@ -636,8 +632,6 @@ class ServerAsyncReader : public ServerAsyncStreamingInterface,
     if (status.IsOk()) {
       finish_buf_.AddSendMessage(msg);
     }
-    bool cancelled = false;
-    finish_buf_.AddServerRecvClose(&cancelled);
     finish_buf_.AddServerSendStatus(&ctx_->trailing_metadata_, status);
     call_.PerformOps(&finish_buf_);
   }
@@ -649,8 +643,6 @@ class ServerAsyncReader : public ServerAsyncStreamingInterface,
       finish_buf_.AddSendInitialMetadata(&ctx_->initial_metadata_);
       ctx_->sent_initial_metadata_ = true;
     }
-    bool cancelled = false;
-    finish_buf_.AddServerRecvClose(&cancelled);
     finish_buf_.AddServerSendStatus(&ctx_->trailing_metadata_, status);
     call_.PerformOps(&finish_buf_);
   }
@@ -697,8 +689,6 @@ class ServerAsyncWriter : public ServerAsyncStreamingInterface,
       finish_buf_.AddSendInitialMetadata(&ctx_->initial_metadata_);
       ctx_->sent_initial_metadata_ = true;
     }
-    bool cancelled = false;
-    finish_buf_.AddServerRecvClose(&cancelled);
     finish_buf_.AddServerSendStatus(&ctx_->trailing_metadata_, status);
     call_.PerformOps(&finish_buf_);
   }
@@ -753,8 +743,6 @@ class ServerAsyncReaderWriter : public ServerAsyncStreamingInterface,
       finish_buf_.AddSendInitialMetadata(&ctx_->initial_metadata_);
       ctx_->sent_initial_metadata_ = true;
     }
-    bool cancelled = false;
-    finish_buf_.AddServerRecvClose(&cancelled);
     finish_buf_.AddServerSendStatus(&ctx_->trailing_metadata_, status);
     call_.PerformOps(&finish_buf_);
   }