Add WaitForState

pull/2811/head
yang-g 10 years ago
parent 36f5965247
commit 8708dd76c1
  1. 10
      include/grpc++/channel_interface.h
  2. 36
      src/cpp/client/channel.cc
  3. 9
      src/cpp/client/channel.h
  4. 4
      test/cpp/end2end/end2end_test.cc

@ -69,10 +69,15 @@ class ChannelInterface : public CallHook,
gpr_timespec deadline, gpr_timespec deadline,
CompletionQueue* cq, void* tag) = 0; CompletionQueue* cq, void* tag) = 0;
// Blocking wait for channel state change or deadline expires. // Blocking wait for channel state change or deadline expiration.
// GetState needs to called to get the current state. // GetState needs to called to get the current state.
virtual bool WaitForStateChange(grpc_connectivity_state last_observed, virtual bool WaitForStateChange(grpc_connectivity_state last_observed,
gpr_timespec deadline) = 0; gpr_timespec deadline) = 0;
// Blocking wait for target state or deadline expriration.
virtual bool WaitForState(grpc_connectivity_state target_state,
gpr_timespec deadline) = 0;
#ifndef GRPC_CXX0X_NO_CHRONO #ifndef GRPC_CXX0X_NO_CHRONO
virtual void NotifyOnStateChange( virtual void NotifyOnStateChange(
grpc_connectivity_state last_observed, grpc_connectivity_state last_observed,
@ -81,6 +86,9 @@ class ChannelInterface : public CallHook,
virtual bool WaitForStateChange( virtual bool WaitForStateChange(
grpc_connectivity_state last_observed, grpc_connectivity_state last_observed,
const std::chrono::system_clock::time_point& deadline) = 0; const std::chrono::system_clock::time_point& deadline) = 0;
virtual bool WaitForState(
grpc_connectivity_state target_state,
const std::chrono::system_clock::time_point& deadline) = 0;
#endif // !GRPC_CXX0X_NO_CHRONO #endif // !GRPC_CXX0X_NO_CHRONO
}; };

@ -136,6 +136,30 @@ bool WaitForStateChangeShared(grpc_channel* channel,
return ok; return ok;
} }
template <typename T>
bool WaitForStateShared(grpc_channel* channel,
grpc_connectivity_state target_state,
const T& deadline) {
grpc_connectivity_state current_state =
grpc_channel_check_connectivity_state(channel, 0);
if (current_state == target_state) {
return true;
}
TimePoint<T> deadline_tp(deadline);
CompletionQueue cq;
bool ok = false;
void* tag = NULL;
while (current_state != target_state) {
NotifyOnStateChangeShared(channel, current_state, deadline_tp.raw_time(),
&cq, NULL);
cq.Next(&tag, &ok);
if (!ok) {
return false;
}
current_state = grpc_channel_check_connectivity_state(channel, 0);
}
return true;
}
} // namespace } // namespace
void Channel::NotifyOnStateChange(grpc_connectivity_state last_observed, void Channel::NotifyOnStateChange(grpc_connectivity_state last_observed,
@ -149,6 +173,11 @@ bool Channel::WaitForStateChange(grpc_connectivity_state last_observed,
return WaitForStateChangeShared(c_channel_, last_observed, deadline); return WaitForStateChangeShared(c_channel_, last_observed, deadline);
} }
bool Channel::WaitForState(grpc_connectivity_state target_state,
gpr_timespec deadline) {
return WaitForStateShared(c_channel_, target_state, deadline);
}
#ifndef GRPC_CXX0X_NO_CHRONO #ifndef GRPC_CXX0X_NO_CHRONO
void Channel::NotifyOnStateChange( void Channel::NotifyOnStateChange(
grpc_connectivity_state last_observed, grpc_connectivity_state last_observed,
@ -162,5 +191,12 @@ bool Channel::WaitForStateChange(
const std::chrono::system_clock::time_point& deadline) { const std::chrono::system_clock::time_point& deadline) {
return WaitForStateChangeShared(c_channel_, last_observed, deadline); return WaitForStateChangeShared(c_channel_, last_observed, deadline);
} }
bool Channel::WaitForState(
grpc_connectivity_state target_state,
const std::chrono::system_clock::time_point& deadline) {
return WaitForStateShared(c_channel_, target_state, deadline);
}
#endif // !GRPC_CXX0X_NO_CHRONO #endif // !GRPC_CXX0X_NO_CHRONO
} // namespace grpc } // namespace grpc

@ -71,15 +71,24 @@ class Channel GRPC_FINAL : public GrpcLibrary, public ChannelInterface {
bool WaitForStateChange(grpc_connectivity_state last_observed, bool WaitForStateChange(grpc_connectivity_state last_observed,
gpr_timespec deadline) GRPC_OVERRIDE; gpr_timespec deadline) GRPC_OVERRIDE;
bool WaitForState(grpc_connectivity_state target_state,
gpr_timespec deadline) GRPC_OVERRIDE;
#ifndef GRPC_CXX0X_NO_CHRONO #ifndef GRPC_CXX0X_NO_CHRONO
void NotifyOnStateChange( void NotifyOnStateChange(
grpc_connectivity_state last_observed, grpc_connectivity_state last_observed,
const std::chrono::system_clock::time_point& deadline, const std::chrono::system_clock::time_point& deadline,
CompletionQueue* cq, void* tag) GRPC_OVERRIDE; CompletionQueue* cq, void* tag) GRPC_OVERRIDE;
bool WaitForStateChange( bool WaitForStateChange(
grpc_connectivity_state last_observed, grpc_connectivity_state last_observed,
const std::chrono::system_clock::time_point& deadline) GRPC_OVERRIDE; const std::chrono::system_clock::time_point& deadline) GRPC_OVERRIDE;
bool WaitForState(grpc_connectivity_state target_state,
const std::chrono::system_clock::time_point& deadline)
GRPC_OVERRIDE;
#endif // !GRPC_CXX0X_NO_CHRONO #endif // !GRPC_CXX0X_NO_CHRONO
private: private:
const grpc::string host_; const grpc::string host_;
grpc_channel* const c_channel_; // owned grpc_channel* const c_channel_; // owned

@ -876,10 +876,10 @@ TEST_F(End2endTest, ChannelState) {
// Start IDLE // Start IDLE
EXPECT_EQ(GRPC_CHANNEL_IDLE, channel_->GetState(false)); EXPECT_EQ(GRPC_CHANNEL_IDLE, channel_->GetState(false));
// Did not ask to connect, no state change.
CompletionQueue cq; CompletionQueue cq;
std::chrono::system_clock::time_point deadline = std::chrono::system_clock::time_point deadline =
std::chrono::system_clock::now() + std::chrono::milliseconds(10); std::chrono::system_clock::now() + std::chrono::milliseconds(10);
// No state change.
channel_->NotifyOnStateChange(GRPC_CHANNEL_IDLE, deadline, &cq, NULL); channel_->NotifyOnStateChange(GRPC_CHANNEL_IDLE, deadline, &cq, NULL);
void* tag; void* tag;
bool ok = true; bool ok = true;
@ -890,6 +890,8 @@ TEST_F(End2endTest, ChannelState) {
EXPECT_TRUE(channel_->WaitForStateChange( EXPECT_TRUE(channel_->WaitForStateChange(
GRPC_CHANNEL_IDLE, gpr_inf_future(GPR_CLOCK_REALTIME))); GRPC_CHANNEL_IDLE, gpr_inf_future(GPR_CLOCK_REALTIME)));
EXPECT_EQ(GRPC_CHANNEL_CONNECTING, channel_->GetState(false)); EXPECT_EQ(GRPC_CHANNEL_CONNECTING, channel_->GetState(false));
EXPECT_TRUE(channel_->WaitForState(GRPC_CHANNEL_READY,
gpr_inf_future(GPR_CLOCK_REALTIME)));
} }
} // namespace testing } // namespace testing

Loading…
Cancel
Save