Add client side WaitForInitialMetadata for streaming.

pull/501/head
Yang Gao 10 years ago
parent 1fe817fffd
commit fd7199f64e
  1. 9
      include/grpc++/impl/call.h
  2. 29
      include/grpc++/stream.h
  3. 6
      src/cpp/common/call.cc

@ -134,7 +134,16 @@ class Call final {
grpc_call *call() { return call_; } grpc_call *call() { return call_; }
CompletionQueue *cq() { return cq_; } CompletionQueue *cq() { return cq_; }
// TODO(yangg) change it to a general state query function.
bool initial_metadata_received() {
return initial_metadata_received_;
}
void set_initial_metadata_received() {
initial_metadata_received_ = true;
}
private: private:
bool initial_metadata_received_ = false;
CallHook *call_hook_; CallHook *call_hook_;
CompletionQueue *cq_; CompletionQueue *cq_;
grpc_call* call_; grpc_call* call_;

@ -98,7 +98,22 @@ class ClientReader final : public ClientStreamingInterface,
cq_.Pluck(&buf); cq_.Pluck(&buf);
} }
// Blocking wait for initial metadata from server. The received metadata
// can only be accessed after this call returns. Calling this method is
// optional as it will be called internally before the first Read.
void WaitForInitialMetadata() {
if (!call_.initial_metadata_received()) {
CallOpBuffer buf;
buf.AddRecvInitialMetadata(&context_->recv_initial_metadata_);
call_.PerformOps(&buf);
GPR_ASSERT(cq_.Pluck(&buf));
call_.set_initial_metadata_received();
}
}
virtual bool Read(R *msg) override { virtual bool Read(R *msg) override {
WaitForInitialMetadata();
CallOpBuffer buf; CallOpBuffer buf;
bool got_message; bool got_message;
buf.AddRecvMessage(msg, &got_message); buf.AddRecvMessage(msg, &got_message);
@ -186,7 +201,21 @@ class ClientReaderWriter final : public ClientStreamingInterface,
GPR_ASSERT(cq_.Pluck(&buf)); GPR_ASSERT(cq_.Pluck(&buf));
} }
// Blocking wait for initial metadata from server. The received metadata
// can only be accessed after this call returns. Calling this method is
// optional as it will be called internally before the first Read.
void WaitForInitialMetadata() {
if (!call_.initial_metadata_received()) {
CallOpBuffer buf;
buf.AddRecvInitialMetadata(&context_->recv_initial_metadata_);
call_.PerformOps(&buf);
GPR_ASSERT(cq_.Pluck(&buf));
call_.set_initial_metadata_received();
}
}
virtual bool Read(R *msg) override { virtual bool Read(R *msg) override {
WaitForInitialMetadata();
CallOpBuffer buf; CallOpBuffer buf;
bool got_message; bool got_message;
buf.AddRecvMessage(msg, &got_message); buf.AddRecvMessage(msg, &got_message);

@ -121,6 +121,12 @@ void CallOpBuffer::AddSendInitialMetadata(
initial_metadata_ = FillMetadataArray(metadata); initial_metadata_ = FillMetadataArray(metadata);
} }
void CallOpBuffer::AddRecvInitialMetadata(
std::multimap<grpc::string, grpc::string>* metadata) {
recv_initial_metadata_ = metadata;
}
void CallOpBuffer::AddSendInitialMetadata(ClientContext *ctx) { void CallOpBuffer::AddSendInitialMetadata(ClientContext *ctx) {
AddSendInitialMetadata(&ctx->send_initial_metadata_); AddSendInitialMetadata(&ctx->send_initial_metadata_);
} }

Loading…
Cancel
Save