Add a cq argument in ClientAsync ctor and give it to the Call

1 files changed, 20 insertions, 22 deletions

diff --git a/include/grpc++/stream.h b/include/grpc++/stream.h
index 2f37cc4075..4bc540cb57 100644
--- a/include/grpc++/stream.h
+++ b/include/grpc++/stream.h
@@ -394,10 +394,10 @@ class ClientAsyncReader final : public ClientAsyncStreamingInterface,
                            public AsyncReaderInterface<R> {
  public:
   // Create a stream and write the first request out.
-  ClientAsyncReader(ChannelInterface *channel, const RpcMethod &method,
-               ClientContext *context,
+  ClientAsyncReader(ChannelInterface *channel, CompletionQueue* cq,
+                    const RpcMethod &method, ClientContext *context,
                const google::protobuf::Message &request, void* tag)
-      : context_(context), call_(channel->CreateCall(method, context, &cq_)) {
+      : context_(context), call_(channel->CreateCall(method, context, cq)) {
     init_buf_.Reset(tag);
     init_buf_.AddSendInitialMetadata(&context->send_initial_metadata_);
     init_buf_.AddSendMessage(request);
@@ -408,10 +408,9 @@ class ClientAsyncReader final : public ClientAsyncStreamingInterface,
   void ReadInitialMetadata(void* tag) override {
     GPR_ASSERT(!context_->initial_metadata_received_);
 
-    CallOpBuffer buf;
-    buf.Reset(tag);
-    buf.AddRecvInitialMetadata(&context_->recv_initial_metadata_);
-    call_.PerformOps(&buf);
+    meta_buf_.Reset(tag);
+    meta_buf_.AddRecvInitialMetadata(&context_->recv_initial_metadata_);
+    call_.PerformOps(&meta_buf_);
     context_->initial_metadata_received_ = true;
   }
 
@@ -437,9 +436,9 @@ class ClientAsyncReader final : public ClientAsyncStreamingInterface,
 
  private:
   ClientContext* context_ = nullptr;
-  CompletionQueue cq_;
   Call call_;
   CallOpBuffer init_buf_;
+  CallOpBuffer meta_buf_;
   CallOpBuffer read_buf_;
   CallOpBuffer finish_buf_;
 };
@@ -448,11 +447,11 @@ template <class W>
 class ClientAsyncWriter final : public ClientAsyncStreamingInterface,
                            public WriterInterface<W> {
  public:
-  ClientAsyncWriter(ChannelInterface *channel, const RpcMethod &method,
-               ClientContext *context,
-               google::protobuf::Message *response, void* tag)
+  ClientAsyncWriter(ChannelInterface *channel, CompletionQueue* cq,
+                    const RpcMethod &method, ClientContext *context,
+                    google::protobuf::Message *response, void* tag)
       : context_(context), response_(response),
-        call_(channel->CreateCall(method, context, &cq_)) {
+        call_(channel->CreateCall(method, context, cq)) {
     init_buf_.Reset(tag);
     init_buf_.AddSendInitialMetadata(&context->send_initial_metadata_);
     call_.PerformOps(&init_buf_);
@@ -461,10 +460,9 @@ class ClientAsyncWriter final : public ClientAsyncStreamingInterface,
   void ReadInitialMetadata(void* tag) override {
     GPR_ASSERT(!context_->initial_metadata_received_);
 
-    CallOpBuffer buf;
-    buf.Reset(tag);
-    buf.AddRecvInitialMetadata(&context_->recv_initial_metadata_);
-    call_.PerformOps(&buf);
+    meta_buf_.Reset(tag);
+    meta_buf_.AddRecvInitialMetadata(&context_->recv_initial_metadata_);
+    call_.PerformOps(&meta_buf_);
     context_->initial_metadata_received_ = true;
   }
 
@@ -495,9 +493,9 @@ class ClientAsyncWriter final : public ClientAsyncStreamingInterface,
   ClientContext* context_ = nullptr;
   google::protobuf::Message *const response_;
   bool got_message_;
-  CompletionQueue cq_;
   Call call_;
   CallOpBuffer init_buf_;
+  CallOpBuffer meta_buf_;
   CallOpBuffer write_buf_;
   CallOpBuffer writes_done_buf_;
   CallOpBuffer finish_buf_;
@@ -509,7 +507,7 @@ class ClientAsyncReaderWriter final : public ClientAsyncStreamingInterface,
                                  public AsyncWriterInterface<W>,
                                  public AsyncReaderInterface<R> {
  public:
-  ClientAsyncReaderWriter(ChannelInterface *channel,
+  ClientAsyncReaderWriter(ChannelInterface *channel, CompletionQueue* cq,
                      const RpcMethod &method, ClientContext *context, void* tag)
       : context_(context), call_(channel->CreateCall(method, context, &cq_)) {
     init_buf_.Reset(tag);
@@ -520,10 +518,9 @@ class ClientAsyncReaderWriter final : public ClientAsyncStreamingInterface,
   void ReadInitialMetadata(void* tag) override {
     GPR_ASSERT(!context_->initial_metadata_received_);
 
-    CallOpBuffer buf;
-    buf.Reset(tag);
-    buf.AddRecvInitialMetadata(&context_->recv_initial_metadata_);
-    call_.PerformOps(&buf);
+    meta_buf_.Reset(tag);
+    meta_buf_.AddRecvInitialMetadata(&context_->recv_initial_metadata_);
+    call_.PerformOps(&meta_buf_);
     context_->initial_metadata_received_ = true;
   }
 
@@ -564,6 +561,7 @@ class ClientAsyncReaderWriter final : public ClientAsyncStreamingInterface,
   CompletionQueue cq_;
   Call call_;
   CallOpBuffer init_buf_;
+  CallOpBuffer meta_buf_;
   CallOpBuffer read_buf_;
   CallOpBuffer write_buf_;
   CallOpBuffer writes_done_buf_;