1 /*
2 *
3 * Copyright 2015 gRPC authors.
4 *
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 *
17 */
18
19 #include <memory>
20 #include <iostream>
21 #include <string>
22 #include <thread>
23
24 #include <grpcpp/grpcpp.h>
25 #include <grpc/support/log.h>
26
27 #ifdef BAZEL_BUILD
28 #include "examples/protos/helloworld.grpc.pb.h"
29 #else
30 #include "helloworld.grpc.pb.h"
31 #endif
32
33 using grpc::Server;
34 using grpc::ServerAsyncResponseWriter;
35 using grpc::ServerBuilder;
36 using grpc::ServerContext;
37 using grpc::ServerCompletionQueue;
38 using grpc::Status;
39 using helloworld::HelloRequest;
40 using helloworld::HelloReply;
41 using helloworld::Greeter;
42
43 class ServerImpl final {
44 public:
~ServerImpl()45 ~ServerImpl() {
46 server_->Shutdown();
47 // Always shutdown the completion queue after the server.
48 cq_->Shutdown();
49 }
50
51 // There is no shutdown handling in this code.
Run()52 void Run() {
53 std::string server_address("0.0.0.0:50051");
54
55 ServerBuilder builder;
56 // Listen on the given address without any authentication mechanism.
57 builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
58 // Register "service_" as the instance through which we'll communicate with
59 // clients. In this case it corresponds to an *asynchronous* service.
60 builder.RegisterService(&service_);
61 // Get hold of the completion queue used for the asynchronous communication
62 // with the gRPC runtime.
63 cq_ = builder.AddCompletionQueue();
64 // Finally assemble the server.
65 server_ = builder.BuildAndStart();
66 std::cout << "Server listening on " << server_address << std::endl;
67
68 // Proceed to the server's main loop.
69 HandleRpcs();
70 }
71
72 private:
73 // Class encompasing the state and logic needed to serve a request.
74 class CallData {
75 public:
76 // Take in the "service" instance (in this case representing an asynchronous
77 // server) and the completion queue "cq" used for asynchronous communication
78 // with the gRPC runtime.
CallData(Greeter::AsyncService * service,ServerCompletionQueue * cq)79 CallData(Greeter::AsyncService* service, ServerCompletionQueue* cq)
80 : service_(service), cq_(cq), responder_(&ctx_), status_(CREATE) {
81 // Invoke the serving logic right away.
82 Proceed();
83 }
84
Proceed()85 void Proceed() {
86 if (status_ == CREATE) {
87 // Make this instance progress to the PROCESS state.
88 status_ = PROCESS;
89
90 // As part of the initial CREATE state, we *request* that the system
91 // start processing SayHello requests. In this request, "this" acts are
92 // the tag uniquely identifying the request (so that different CallData
93 // instances can serve different requests concurrently), in this case
94 // the memory address of this CallData instance.
95 service_->RequestSayHello(&ctx_, &request_, &responder_, cq_, cq_,
96 this);
97 } else if (status_ == PROCESS) {
98 // Spawn a new CallData instance to serve new clients while we process
99 // the one for this CallData. The instance will deallocate itself as
100 // part of its FINISH state.
101 new CallData(service_, cq_);
102
103 // The actual processing.
104 std::string prefix("Hello ");
105 reply_.set_message(prefix + request_.name());
106
107 // And we are done! Let the gRPC runtime know we've finished, using the
108 // memory address of this instance as the uniquely identifying tag for
109 // the event.
110 status_ = FINISH;
111 responder_.Finish(reply_, Status::OK, this);
112 } else {
113 GPR_ASSERT(status_ == FINISH);
114 // Once in the FINISH state, deallocate ourselves (CallData).
115 delete this;
116 }
117 }
118
119 private:
120 // The means of communication with the gRPC runtime for an asynchronous
121 // server.
122 Greeter::AsyncService* service_;
123 // The producer-consumer queue where for asynchronous server notifications.
124 ServerCompletionQueue* cq_;
125 // Context for the rpc, allowing to tweak aspects of it such as the use
126 // of compression, authentication, as well as to send metadata back to the
127 // client.
128 ServerContext ctx_;
129
130 // What we get from the client.
131 HelloRequest request_;
132 // What we send back to the client.
133 HelloReply reply_;
134
135 // The means to get back to the client.
136 ServerAsyncResponseWriter<HelloReply> responder_;
137
138 // Let's implement a tiny state machine with the following states.
139 enum CallStatus { CREATE, PROCESS, FINISH };
140 CallStatus status_; // The current serving state.
141 };
142
143 // This can be run in multiple threads if needed.
HandleRpcs()144 void HandleRpcs() {
145 // Spawn a new CallData instance to serve new clients.
146 new CallData(&service_, cq_.get());
147 void* tag; // uniquely identifies a request.
148 bool ok;
149 while (true) {
150 // Block waiting to read the next event from the completion queue. The
151 // event is uniquely identified by its tag, which in this case is the
152 // memory address of a CallData instance.
153 // The return value of Next should always be checked. This return value
154 // tells us whether there is any kind of event or cq_ is shutting down.
155 GPR_ASSERT(cq_->Next(&tag, &ok));
156 GPR_ASSERT(ok);
157 static_cast<CallData*>(tag)->Proceed();
158 }
159 }
160
161 std::unique_ptr<ServerCompletionQueue> cq_;
162 Greeter::AsyncService service_;
163 std::unique_ptr<Server> server_;
164 };
165
main(int argc,char ** argv)166 int main(int argc, char** argv) {
167 ServerImpl server;
168 server.Run();
169
170 return 0;
171 }
172