From 5705fe3fca7c9ed042b78dee3c2b89364fe183f8 Mon Sep 17 00:00:00 2001 From: Yang Gao Date: Thu, 12 Feb 2015 14:30:34 -0800 Subject: Add a cq argument in ClientAsync ctor and give it to the Call --- include/grpc++/stream.h | 42 ++++++++++++++++++++---------------------- 1 file 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 { 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 ClientAsyncWriter final : public ClientAsyncStreamingInterface, public WriterInterface { 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, public AsyncReaderInterface { 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_; -- cgit v1.2.3