Async helloworld client and server

cpp/helloworld/Makefile

@@ -41,7 +41,7 @@ PROTOS_PATH = ../../protos
 
 vpath %.proto $(PROTOS_PATH)
 
-all: system-check greeter_client greeter_server
+all: system-check greeter_client greeter_server greeter_async_client greeter_async_server
 
 greeter_client: helloworld.pb.o helloworld.grpc.pb.o greeter_client.o
 	$(CXX) $^ $(LDFLAGS) -o $@
@@ -49,6 +49,12 @@ greeter_client: helloworld.pb.o helloworld.grpc.pb.o greeter_client.o
 greeter_server: helloworld.pb.o helloworld.grpc.pb.o greeter_server.o
 	$(CXX) $^ $(LDFLAGS) -o $@
 
+greeter_async_client: helloworld.pb.o helloworld.grpc.pb.o greeter_async_client.o
+	$(CXX) $^ $(LDFLAGS) -o $@
+
+greeter_async_server: helloworld.pb.o helloworld.grpc.pb.o greeter_async_server.o
+	$(CXX) $^ $(LDFLAGS) -o $@
+
 %.grpc.pb.cc: %.proto
 	$(PROTOC) -I $(PROTOS_PATH) --grpc_out=. --plugin=protoc-gen-grpc=$(GRPC_CPP_PLUGIN_PATH) $<
 
@@ -56,7 +62,7 @@ greeter_server: helloworld.pb.o helloworld.grpc.pb.o greeter_server.o
 	$(PROTOC) -I $(PROTOS_PATH) --cpp_out=. $<
 
 clean:
-	rm -f *.o *.pb.cc *.pb.h greeter_client greeter_server
+	rm -f *.o *.pb.cc *.pb.h greeter_client greeter_server greeter_async_client greeter_async_server
 
 
 # The following is to test your system and ensure a smoother experience.

cpp/helloworld/greeter_async_client.cc

@@ -0,0 +1,111 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include <iostream>
+#include <memory>
+#include <string>
+
+#include <grpc/grpc.h>
+#include <grpc/support/log.h>
+#include <grpc++/async_unary_call.h>
+#include <grpc++/channel_arguments.h>
+#include <grpc++/channel_interface.h>
+#include <grpc++/client_context.h>
+#include <grpc++/completion_queue.h>
+#include <grpc++/create_channel.h>
+#include <grpc++/credentials.h>
+#include <grpc++/status.h>
+#include "helloworld.grpc.pb.h"
+
+using grpc::ChannelArguments;
+using grpc::ChannelInterface;
+using grpc::ClientAsyncResponseReader;
+using grpc::ClientContext;
+using grpc::CompletionQueue;
+using grpc::Status;
+using helloworld::HelloRequest;
+using helloworld::HelloReply;
+using helloworld::Greeter;
+
+class GreeterClient {
+ public:
+  GreeterClient(std::shared_ptr<ChannelInterface> channel)
+      : stub_(Greeter::NewStub(channel)) {}
+
+  std::string SayHello(const std::string& user) {
+    HelloRequest request;
+    request.set_name(user);
+    HelloReply reply;
+    ClientContext context;
+    CompletionQueue cq;
+    Status status;
+
+    std::unique_ptr<ClientAsyncResponseReader<HelloReply> > rpc(
+        stub_->AsyncSayHello(&context, request, &cq, (void*)1));
+    void* got_tag;
+    bool ok;
+    cq.Next(&got_tag, &ok);
+    GPR_ASSERT(ok);
+    GPR_ASSERT(got_tag == (void*)1);
+
+    rpc->Finish(&reply, &status, (void*)2);
+    cq.Next(&got_tag, &ok);
+    GPR_ASSERT(ok);
+    GPR_ASSERT(got_tag == (void*)2);
+
+    if (status.IsOk()) {
+      return reply.message();
+    } else {
+      return "Rpc failed";
+    }
+  }
+
+  void Shutdown() { stub_.reset(); }
+
+ private:
+  std::unique_ptr<Greeter::Stub> stub_;
+};
+
+int main(int argc, char** argv) {
+  grpc_init();
+
+  GreeterClient greeter(grpc::CreateChannel(
+      "localhost:50051", grpc::InsecureCredentials(), ChannelArguments()));
+  std::string user("world");
+  std::string reply = greeter.SayHello(user);
+  std::cout << "Greeter received: " << reply << std::endl;
+
+  greeter.Shutdown();
+
+  grpc_shutdown();
+}

cpp/helloworld/greeter_async_server.cc

@@ -0,0 +1,134 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include <memory>
+#include <iostream>
+#include <string>
+#include <thread>
+
+#include <grpc/grpc.h>
+#include <grpc/support/log.h>
+#include <grpc++/async_unary_call.h>
+#include <grpc++/completion_queue.h>
+#include <grpc++/server.h>
+#include <grpc++/server_builder.h>
+#include <grpc++/server_context.h>
+#include <grpc++/server_credentials.h>
+#include <grpc++/status.h>
+#include "helloworld.grpc.pb.h"
+
+using grpc::CompletionQueue;
+using grpc::Server;
+using grpc::ServerAsyncResponseWriter;
+using grpc::ServerBuilder;
+using grpc::ServerContext;
+using grpc::Status;
+using helloworld::HelloRequest;
+using helloworld::HelloReply;
+using helloworld::Greeter;
+
+static bool got_sigint = false;
+
+class ServerImpl final {
+ public:
+  ServerImpl() : service_(&service_cq_) {}
+
+  ~ServerImpl() {
+    server_->Shutdown();
+    rpc_cq_.Shutdown();
+    service_cq_.Shutdown();
+  }
+
+  // There is no shutdown handling in this code.
+  void Run() {
+    std::string server_address("0.0.0.0:50051");
+
+    ServerBuilder builder;
+    builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
+    builder.RegisterAsyncService(&service_);
+    server_ = builder.BuildAndStart();
+    std::cout << "Server listening on " << server_address << std::endl;
+
+    while (true) {
+      CallData* rpc = new CallData();
+      service_.RequestSayHello(&rpc->ctx, &rpc->request, &rpc->responder,
+                               &rpc_cq_, rpc);
+      void* got_tag;
+      bool ok;
+      service_cq_.Next(&got_tag, &ok);
+      GPR_ASSERT(ok);
+      GPR_ASSERT(got_tag == rpc);
+
+      std::thread t(&ServerImpl::HandleRpc, this, rpc);
+      t.detach();
+    }
+  }
+
+ private:
+  struct CallData {
+    CallData() : responder(&ctx) {}
+    ServerContext ctx;
+    HelloRequest request;
+    HelloReply reply;
+    ServerAsyncResponseWriter<HelloReply> responder;
+  };
+
+  // Runs in a detached thread, processes rpc then deletes data.
+  void HandleRpc(CallData* rpc) {
+    std::string prefix("Hello ");
+    rpc->reply.set_message(prefix + rpc->request.name());
+    rpc->responder.Finish(rpc->reply, Status::OK, &rpc->ctx);
+    void* got_tag;
+    bool ok;
+    rpc_cq_.Next(&got_tag, &ok);
+    GPR_ASSERT(ok);
+    GPR_ASSERT(got_tag == &rpc->ctx);
+
+    delete rpc;
+  }
+
+  CompletionQueue service_cq_;
+  CompletionQueue rpc_cq_;
+  Greeter::AsyncService service_;
+  std::unique_ptr<Server> server_;
+};
+
+int main(int argc, char** argv) {
+  grpc_init();
+
+  ServerImpl server;
+  server.Run();
+
+  grpc_shutdown();
+  return 0;
+}