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

pull/501/head
Yang Gao 10 years ago
parent 984b09087f
commit 5705fe3fca
  1. 42
      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_;

Loading…
Cancel
Save