Implement both Publisher and Subscriber.
diff --git a/examples/tips/client_main.cc b/examples/tips/main.cc
similarity index 71%
rename from examples/tips/client_main.cc
rename to examples/tips/main.cc
index 5a3a0da..94a0bc7 100644
--- a/examples/tips/client_main.cc
+++ b/examples/tips/main.cc
@@ -46,7 +46,8 @@
#include <grpc++/credentials.h>
#include <grpc++/status.h>
-#include "examples/tips/client.h"
+#include "examples/tips/publisher.h"
+#include "examples/tips/subscriber.h"
#include "test/cpp/util/create_test_channel.h"
DEFINE_int32(server_port, 443, "Server port.");
@@ -54,7 +55,17 @@
"pubsub-staging.googleapis.com", "Server host to connect to");
DEFINE_string(service_account_key_file, "",
"Path to service account json key file.");
-DEFINE_string(oauth_scope, "", "Scope for OAuth tokens.");
+DEFINE_string(oauth_scope,
+ "https://www.googleapis.com/auth/cloud-platform",
+ "Scope for OAuth tokens.");
+
+namespace {
+
+const char kTopic[] = "/topics/stoked-keyword-656/testtopics";
+const char kSubscriptionName[] = "stoked-keyword-656/testsubscription";
+const char kMessageData[] = "Message Data";
+
+} // namespace
grpc::string GetServiceAccountJsonKey() {
static grpc::string json_key;
@@ -94,20 +105,40 @@
true, // use prod roots
creds));
- grpc::examples::tips::Client client(channel);
+ grpc::examples::tips::Publisher publisher(channel);
+ grpc::examples::tips::Subscriber subscriber(channel);
- grpc::Status s = client.CreateTopic("/topics/stoked-keyword-656/testtopics");
- gpr_log(GPR_INFO, "return code %d, %s", s.code(), s.details().c_str());
+ grpc::string topic = kTopic;
+
+ if (publisher.GetTopic(topic).IsOk()) publisher.DeleteTopic(topic);
+
+ grpc::Status s = publisher.CreateTopic(topic);
+ gpr_log(GPR_INFO, "Create topic returns code %d, %s",
+ s.code(), s.details().c_str());
GPR_ASSERT(s.IsOk());
- s = client.GetTopic("/topics/stoked-keyword-656/testtopics");
- gpr_log(GPR_INFO, "return code %d, %s", s.code(), s.details().c_str());
+ s = publisher.GetTopic(topic);
+ gpr_log(GPR_INFO, "Get topic returns code %d, %s",
+ s.code(), s.details().c_str());
GPR_ASSERT(s.IsOk());
- s = client.DeleteTopic("/topics/stoked-keyword-656/testtopics");
- gpr_log(GPR_INFO, "return code %d, %s", s.code(), s.details().c_str());
+ s = publisher.Publish(topic, kMessageData);
+ gpr_log(GPR_INFO, "Publish returns code %d, %s",
+ s.code(), s.details().c_str());
GPR_ASSERT(s.IsOk());
+ s = subscriber.CreateSubscription(kTopic, kSubscriptionName);
+ gpr_log(GPR_INFO, "create subscrption returns code %d, %s",
+ s.code(), s.details().c_str());
+ GPR_ASSERT(s.IsOk());
+
+ s = publisher.DeleteTopic(kTopic);
+ gpr_log(GPR_INFO, "Delete topic returns code %d, %s",
+ s.code(), s.details().c_str());
+ GPR_ASSERT(s.IsOk());
+
+ subscriber.Shutdown();
+ publisher.Shutdown();
channel.reset();
grpc_shutdown();
return 0;
diff --git a/examples/tips/client.cc b/examples/tips/publisher.cc
similarity index 78%
rename from examples/tips/client.cc
rename to examples/tips/publisher.cc
index f9d5319..238ea59 100644
--- a/examples/tips/client.cc
+++ b/examples/tips/publisher.cc
@@ -33,7 +33,7 @@
#include <grpc++/client_context.h>
-#include "examples/tips/client.h"
+#include "examples/tips/publisher.h"
using tech::pubsub::Topic;
using tech::pubsub::DeleteTopicRequest;
@@ -41,16 +41,22 @@
using tech::pubsub::PublisherService;
using tech::pubsub::ListTopicsRequest;
using tech::pubsub::ListTopicsResponse;
+using tech::pubsub::PublishRequest;
+using tech::pubsub::PubsubMessage;
namespace grpc {
namespace examples {
namespace tips {
-Client::Client(std::shared_ptr<ChannelInterface> channel)
+Publisher::Publisher(std::shared_ptr<ChannelInterface> channel)
: stub_(PublisherService::NewStub(channel)) {
}
-Status Client::CreateTopic(grpc::string topic) {
+void Publisher::Shutdown() {
+ stub_.reset();
+}
+
+Status Publisher::CreateTopic(grpc::string topic) {
Topic request;
Topic response;
request.set_name(topic);
@@ -59,7 +65,7 @@
return stub_->CreateTopic(&context, request, &response);
}
-Status Client::ListTopics() {
+Status Publisher::ListTopics() {
ListTopicsRequest request;
ListTopicsResponse response;
ClientContext context;
@@ -67,7 +73,7 @@
return stub_->ListTopics(&context, request, &response);
}
-Status Client::GetTopic(grpc::string topic) {
+Status Publisher::GetTopic(grpc::string topic) {
GetTopicRequest request;
Topic response;
ClientContext context;
@@ -77,7 +83,7 @@
return stub_->GetTopic(&context, request, &response);
}
-Status Client::DeleteTopic(grpc::string topic) {
+Status Publisher::DeleteTopic(grpc::string topic) {
DeleteTopicRequest request;
proto2::Empty response;
ClientContext context;
@@ -87,6 +93,18 @@
return stub_->DeleteTopic(&context, request, &response);
}
+Status Publisher::Publish(const grpc::string& topic,
+ const grpc::string& data) {
+ PublishRequest request;
+ proto2::Empty response;
+ ClientContext context;
+
+ request.mutable_message()->set_data(data);
+ request.set_topic(topic);
+
+ return stub_->Publish(&context, request, &response);
+}
+
} // namespace tips
} // namespace examples
} // namespace grpc
diff --git a/examples/tips/client.h b/examples/tips/publisher.h
similarity index 86%
rename from examples/tips/client.h
rename to examples/tips/publisher.h
index 661ee5c..a97dbe6 100644
--- a/examples/tips/client.h
+++ b/examples/tips/publisher.h
@@ -31,8 +31,8 @@
*
*/
-#ifndef __GRPCPP_EXAMPLES_TIPS_CLIENT_H_
-#define __GRPCPP_EXAMPLES_TIPS_CLIENT_H_
+#ifndef __GRPCPP_EXAMPLES_TIPS_PUBLISHER_H_
+#define __GRPCPP_EXAMPLES_TIPS_PUBLISHER_H_
#include <grpc++/channel_interface.h>
#include <grpc++/status.h>
@@ -43,14 +43,18 @@
namespace examples {
namespace tips {
-class Client {
+class Publisher {
public:
- Client(std::shared_ptr<grpc::ChannelInterface> channel);
+ Publisher(std::shared_ptr<grpc::ChannelInterface> channel);
+ void Shutdown();
+
Status CreateTopic(grpc::string topic);
Status GetTopic(grpc::string topic);
Status DeleteTopic(grpc::string topic);
Status ListTopics();
+ Status Publish(const grpc::string& topic, const grpc::string& data);
+
private:
std::unique_ptr<tech::pubsub::PublisherService::Stub> stub_;
};
@@ -59,4 +63,4 @@
} // namespace examples
} // namespace grpc
-#endif // __GRPCPP_EXAMPLES_TIPS_CLIENT_H_
+#endif // __GRPCPP_EXAMPLES_TIPS_PUBLISHER_H_
diff --git a/examples/tips/client_test.cc b/examples/tips/publisher_test.cc
similarity index 64%
copy from examples/tips/client_test.cc
copy to examples/tips/publisher_test.cc
index 69238f2..7f845fe 100644
--- a/examples/tips/client_test.cc
+++ b/examples/tips/publisher_test.cc
@@ -41,7 +41,7 @@
#include <grpc++/status.h>
#include <gtest/gtest.h>
-#include "examples/tips/client.h"
+#include "examples/tips/publisher.h"
#include "test/core/util/port.h"
#include "test/core/util/test_config.h"
@@ -52,8 +52,9 @@
namespace {
const char kTopic[] = "test topic";
+const char kMessageData[] = "test message data";
-class PublishServiceImpl : public tech::pubsub::PublisherService::Service {
+class PublisherServiceImpl : public tech::pubsub::PublisherService::Service {
public:
Status CreateTopic(::grpc::ServerContext* context,
const ::tech::pubsub::Topic* request,
@@ -61,34 +62,71 @@
EXPECT_EQ(request->name(), kTopic);
return Status::OK;
}
+
+ Status Publish(ServerContext* context,
+ const ::tech::pubsub::PublishRequest* request,
+ ::proto2::Empty* response) override {
+ EXPECT_EQ(request->message().data(), kMessageData);
+ return Status::OK;
+ }
+
+ Status GetTopic(ServerContext* context,
+ const ::tech::pubsub::GetTopicRequest* request,
+ ::tech::pubsub::Topic* response) override {
+ EXPECT_EQ(request->topic(), kTopic);
+ return Status::OK;
+ }
+
+ Status ListTopics(ServerContext* context,
+ const ::tech::pubsub::ListTopicsRequest* request,
+ ::tech::pubsub::ListTopicsResponse* response) override {
+ return Status::OK;
+ }
+
+ Status DeleteTopic(ServerContext* context,
+ const ::tech::pubsub::DeleteTopicRequest* request,
+ ::proto2::Empty* response) override {
+ EXPECT_EQ(request->topic(), kTopic);
+ return Status::OK;
+ }
+
};
-class End2endTest : public ::testing::Test {
+class PublisherTest : public ::testing::Test {
protected:
+ // Setup a server and a client for PublisherService.
void SetUp() override {
int port = grpc_pick_unused_port_or_die();
server_address_ << "localhost:" << port;
- // Setup server
ServerBuilder builder;
builder.AddPort(server_address_.str());
builder.RegisterService(service_.service());
server_ = builder.BuildAndStart();
channel_ = CreateChannel(server_address_.str(), ChannelArguments());
+
+ publisher_.reset(new grpc::examples::tips::Publisher(channel_));
}
- void TearDown() override { server_->Shutdown(); }
+ void TearDown() override {
+ server_->Shutdown();
+ publisher_->Shutdown();
+ }
- std::unique_ptr<Server> server_;
std::ostringstream server_address_;
- PublishServiceImpl service_;
+ std::unique_ptr<Server> server_;
+ PublisherServiceImpl service_;
std::shared_ptr<ChannelInterface> channel_;
+
+ std::unique_ptr<grpc::examples::tips::Publisher> publisher_;
};
-TEST_F(End2endTest, CreateTopic) {
- grpc::examples::tips::Client client(channel_);
- client.CreateTopic(kTopic);
+TEST_F(PublisherTest, TestPublisher) {
+ EXPECT_TRUE(publisher_->CreateTopic(kTopic).IsOk());
+ EXPECT_TRUE(publisher_->Publish(kTopic, kMessageData).IsOk());
+ EXPECT_TRUE(publisher_->GetTopic(kTopic).IsOk());
+ EXPECT_TRUE(publisher_->ListTopics().IsOk());
}
} // namespace
diff --git a/examples/tips/client.cc b/examples/tips/subscriber.cc
similarity index 67%
copy from examples/tips/client.cc
copy to examples/tips/subscriber.cc
index f9d5319..a482ad6 100644
--- a/examples/tips/client.cc
+++ b/examples/tips/subscriber.cc
@@ -33,58 +33,52 @@
#include <grpc++/client_context.h>
-#include "examples/tips/client.h"
+#include "examples/tips/subscriber.h"
using tech::pubsub::Topic;
using tech::pubsub::DeleteTopicRequest;
using tech::pubsub::GetTopicRequest;
-using tech::pubsub::PublisherService;
+using tech::pubsub::SubscriberService;
using tech::pubsub::ListTopicsRequest;
using tech::pubsub::ListTopicsResponse;
+using tech::pubsub::PublishRequest;
+using tech::pubsub::PubsubMessage;
namespace grpc {
namespace examples {
namespace tips {
-Client::Client(std::shared_ptr<ChannelInterface> channel)
- : stub_(PublisherService::NewStub(channel)) {
+Subscriber::Subscriber(std::shared_ptr<ChannelInterface> channel)
+ : stub_(SubscriberService::NewStub(channel)) {
}
-Status Client::CreateTopic(grpc::string topic) {
- Topic request;
- Topic response;
- request.set_name(topic);
- ClientContext context;
-
- return stub_->CreateTopic(&context, request, &response);
+void Subscriber::Shutdown() {
+ stub_.reset();
}
-Status Client::ListTopics() {
- ListTopicsRequest request;
- ListTopicsResponse response;
- ClientContext context;
-
- return stub_->ListTopics(&context, request, &response);
-}
-
-Status Client::GetTopic(grpc::string topic) {
- GetTopicRequest request;
- Topic response;
+Status Subscriber::CreateSubscription(const grpc::string& topic,
+ const grpc::string& name) {
+ tech::pubsub::Subscription request;
+ tech::pubsub::Subscription response;
ClientContext context;
request.set_topic(topic);
+ request.set_name(name);
- return stub_->GetTopic(&context, request, &response);
+ return stub_->CreateSubscription(&context, request, &response);
}
-Status Client::DeleteTopic(grpc::string topic) {
- DeleteTopicRequest request;
- proto2::Empty response;
+Status Subscriber::GetSubscription(const grpc::string& name,
+ grpc::string* topic) {
+ tech::pubsub::GetSubscriptionRequest request;
+ tech::pubsub::Subscription response;
ClientContext context;
- request.set_topic(topic);
+ request.set_subscription(name);
- return stub_->DeleteTopic(&context, request, &response);
+ Status s = stub_->GetSubscription(&context, request, &response);
+ *topic = response.topic();
+ return s;
}
} // namespace tips
diff --git a/examples/tips/client.h b/examples/tips/subscriber.h
similarity index 78%
copy from examples/tips/client.h
copy to examples/tips/subscriber.h
index 661ee5c..e0491ff 100644
--- a/examples/tips/client.h
+++ b/examples/tips/subscriber.h
@@ -31,8 +31,8 @@
*
*/
-#ifndef __GRPCPP_EXAMPLES_TIPS_CLIENT_H_
-#define __GRPCPP_EXAMPLES_TIPS_CLIENT_H_
+#ifndef __GRPCPP_EXAMPLES_TIPS_SUBSCRIBER_H_
+#define __GRPCPP_EXAMPLES_TIPS_SUBSCRIBER_H_
#include <grpc++/channel_interface.h>
#include <grpc++/status.h>
@@ -43,20 +43,22 @@
namespace examples {
namespace tips {
-class Client {
+class Subscriber {
public:
- Client(std::shared_ptr<grpc::ChannelInterface> channel);
- Status CreateTopic(grpc::string topic);
- Status GetTopic(grpc::string topic);
- Status DeleteTopic(grpc::string topic);
- Status ListTopics();
+ Subscriber(std::shared_ptr<grpc::ChannelInterface> channel);
+ void Shutdown();
+
+ Status CreateSubscription(const grpc::string& topic,
+ const grpc::string& name);
+
+ Status GetSubscription(const grpc::string& name, grpc::string* topic);
private:
- std::unique_ptr<tech::pubsub::PublisherService::Stub> stub_;
+ std::unique_ptr<tech::pubsub::SubscriberService::Stub> stub_;
};
} // namespace tips
} // namespace examples
} // namespace grpc
-#endif // __GRPCPP_EXAMPLES_TIPS_CLIENT_H_
+#endif // __GRPCPP_EXAMPLES_TIPS_SUBSCRIBER_H_
diff --git a/examples/tips/client_test.cc b/examples/tips/subscriber_test.cc
similarity index 66%
rename from examples/tips/client_test.cc
rename to examples/tips/subscriber_test.cc
index 69238f2..4894814 100644
--- a/examples/tips/client_test.cc
+++ b/examples/tips/subscriber_test.cc
@@ -41,7 +41,7 @@
#include <grpc++/status.h>
#include <gtest/gtest.h>
-#include "examples/tips/client.h"
+#include "examples/tips/subscriber.h"
#include "test/core/util/port.h"
#include "test/core/util/test_config.h"
@@ -52,43 +52,66 @@
namespace {
const char kTopic[] = "test topic";
+const char kSubscriptionName[] = "subscription name";
-class PublishServiceImpl : public tech::pubsub::PublisherService::Service {
+class SubscriberServiceImpl : public tech::pubsub::SubscriberService::Service {
public:
- Status CreateTopic(::grpc::ServerContext* context,
- const ::tech::pubsub::Topic* request,
- ::tech::pubsub::Topic* response) override {
- EXPECT_EQ(request->name(), kTopic);
+ Status CreateSubscription(ServerContext* context,
+ const tech::pubsub::Subscription* request,
+ tech::pubsub::Subscription* response) override {
+ EXPECT_EQ(request->topic(), kTopic);
+ EXPECT_EQ(request->name(), kSubscriptionName);
return Status::OK;
}
+
+ Status GetSubscription(ServerContext* context,
+ const tech::pubsub::GetSubscriptionRequest* request,
+ tech::pubsub::Subscription* response) override {
+ EXPECT_EQ(request->subscription(), kSubscriptionName);
+ response->set_topic(kTopic);
+ return Status::OK;
+ }
+
};
-class End2endTest : public ::testing::Test {
+class SubscriberTest : public ::testing::Test {
protected:
+ // Setup a server and a client for SubscriberService.
void SetUp() override {
int port = grpc_pick_unused_port_or_die();
server_address_ << "localhost:" << port;
- // Setup server
ServerBuilder builder;
builder.AddPort(server_address_.str());
builder.RegisterService(service_.service());
server_ = builder.BuildAndStart();
channel_ = CreateChannel(server_address_.str(), ChannelArguments());
+
+ subscriber_.reset(new grpc::examples::tips::Subscriber(channel_));
}
- void TearDown() override { server_->Shutdown(); }
+ void TearDown() override {
+ server_->Shutdown();
+ subscriber_->Shutdown();
+ }
- std::unique_ptr<Server> server_;
std::ostringstream server_address_;
- PublishServiceImpl service_;
+ std::unique_ptr<Server> server_;
+ SubscriberServiceImpl service_;
std::shared_ptr<ChannelInterface> channel_;
+
+ std::unique_ptr<grpc::examples::tips::Subscriber> subscriber_;
};
-TEST_F(End2endTest, CreateTopic) {
- grpc::examples::tips::Client client(channel_);
- client.CreateTopic(kTopic);
+TEST_F(SubscriberTest, TestSubscriber) {
+ EXPECT_TRUE(subscriber_->CreateSubscription(kTopic,
+ kSubscriptionName).IsOk());
+
+ grpc::string topic;
+ EXPECT_TRUE(subscriber_->GetSubscription(kSubscriptionName,
+ &topic).IsOk());
+ EXPECT_EQ(topic, kTopic);
}
} // namespace