Add Pull method to subscriber
diff --git a/examples/tips/subscriber.cc b/examples/tips/subscriber.cc
index a482ad6..2e2370e 100644
--- a/examples/tips/subscriber.cc
+++ b/examples/tips/subscriber.cc
@@ -81,6 +81,29 @@
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 examples
} // namespace grpc
diff --git a/examples/tips/subscriber.h b/examples/tips/subscriber.h
index e0491ff..38345c0 100644
--- a/examples/tips/subscriber.h
+++ b/examples/tips/subscriber.h
@@ -53,6 +53,8 @@
Status GetSubscription(const grpc::string& name, grpc::string* topic);
+ Status Pull(const grpc::string& name, grpc::string* data);
+
private:
std::unique_ptr<tech::pubsub::SubscriberService::Stub> stub_;
};
diff --git a/examples/tips/subscriber_test.cc b/examples/tips/subscriber_test.cc
index 4894814..4ff9364 100644
--- a/examples/tips/subscriber_test.cc
+++ b/examples/tips/subscriber_test.cc
@@ -53,6 +53,7 @@
const char kTopic[] = "test topic";
const char kSubscriptionName[] = "subscription name";
+const char kData[] = "Message data";
class SubscriberServiceImpl : public tech::pubsub::SubscriberService::Service {
public:
@@ -72,6 +73,21 @@
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 {
@@ -108,10 +124,15 @@
EXPECT_TRUE(subscriber_->CreateSubscription(kTopic,
kSubscriptionName).IsOk());
+
grpc::string topic;
EXPECT_TRUE(subscriber_->GetSubscription(kSubscriptionName,
&topic).IsOk());
EXPECT_EQ(topic, kTopic);
+
+ grpc::string data;
+ EXPECT_TRUE(subscriber_->Pull(kSubscriptionName,
+ &data).IsOk());
}
} // namespace