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