Skip to content
Snippets Groups Projects
Commit 0010cdae authored by Chen Wang's avatar Chen Wang
Browse files

Add Pull method to subscriber

parent 04f1aa80
No related branches found
No related tags found
No related merge requests found
...@@ -81,6 +81,29 @@ Status Subscriber::GetSubscription(const grpc::string& name, ...@@ -81,6 +81,29 @@ Status Subscriber::GetSubscription(const grpc::string& name,
return s; return s;
} }
Status Subscriber::Pull(const grpc::string& name,
grpc::string* data) {
tech::pubsub::PullRequest request;
tech::pubsub::PullResponse response;
ClientContext context;
request.set_subscription(name);
Status s = stub_->Pull(&context, request, &response);
if (s.IsOk()) {
tech::pubsub::PubsubEvent event = response.pubsub_event();
if (event.has_message()) {
*data = event.message().data();
}
tech::pubsub::AcknowledgeRequest ack;
proto2::Empty empty;
ClientContext ack_context;
ack.set_subscription(name);
ack.add_ack_id(response.ack_id());
stub_->Acknowledge(&ack_context, ack, &empty);
}
return s;
}
} // namespace tips } // namespace tips
} // namespace examples } // namespace examples
} // namespace grpc } // namespace grpc
...@@ -53,6 +53,8 @@ class Subscriber { ...@@ -53,6 +53,8 @@ class Subscriber {
Status GetSubscription(const grpc::string& name, grpc::string* topic); Status GetSubscription(const grpc::string& name, grpc::string* topic);
Status Pull(const grpc::string& name, grpc::string* data);
private: private:
std::unique_ptr<tech::pubsub::SubscriberService::Stub> stub_; std::unique_ptr<tech::pubsub::SubscriberService::Stub> stub_;
}; };
......
...@@ -53,6 +53,7 @@ namespace { ...@@ -53,6 +53,7 @@ namespace {
const char kTopic[] = "test topic"; const char kTopic[] = "test topic";
const char kSubscriptionName[] = "subscription name"; const char kSubscriptionName[] = "subscription name";
const char kData[] = "Message data";
class SubscriberServiceImpl : public tech::pubsub::SubscriberService::Service { class SubscriberServiceImpl : public tech::pubsub::SubscriberService::Service {
public: public:
...@@ -72,6 +73,21 @@ class SubscriberServiceImpl : public tech::pubsub::SubscriberService::Service { ...@@ -72,6 +73,21 @@ class SubscriberServiceImpl : public tech::pubsub::SubscriberService::Service {
return Status::OK; return Status::OK;
} }
Status Pull(ServerContext* context,
const tech::pubsub::PullRequest* request,
tech::pubsub::PullResponse* response) override {
EXPECT_EQ(request->subscription(), kSubscriptionName);
response->set_ack_id("1");
response->mutable_pubsub_event()->mutable_message()->set_data(kData);
return Status::OK;
}
Status Acknowledge(ServerContext* context,
const tech::pubsub::AcknowledgeRequest* request,
proto2::Empty* response) override {
return Status::OK;
}
}; };
class SubscriberTest : public ::testing::Test { class SubscriberTest : public ::testing::Test {
...@@ -108,10 +124,15 @@ TEST_F(SubscriberTest, TestSubscriber) { ...@@ -108,10 +124,15 @@ TEST_F(SubscriberTest, TestSubscriber) {
EXPECT_TRUE(subscriber_->CreateSubscription(kTopic, EXPECT_TRUE(subscriber_->CreateSubscription(kTopic,
kSubscriptionName).IsOk()); kSubscriptionName).IsOk());
grpc::string topic; grpc::string topic;
EXPECT_TRUE(subscriber_->GetSubscription(kSubscriptionName, EXPECT_TRUE(subscriber_->GetSubscription(kSubscriptionName,
&topic).IsOk()); &topic).IsOk());
EXPECT_EQ(topic, kTopic); EXPECT_EQ(topic, kTopic);
grpc::string data;
EXPECT_TRUE(subscriber_->Pull(kSubscriptionName,
&data).IsOk());
} }
} // namespace } // namespace
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment