• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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