/* * * 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 #include #include #include #include #include "helloworld.grpc.pb.h" using grpc::Server; using grpc::ServerAsyncResponseWriter; using grpc::ServerBuilder; using grpc::ServerContext; using grpc::ServerCompletionQueue; using grpc::Status; using helloworld::HelloRequest; using helloworld::HelloReply; using helloworld::Greeter; class ServerImpl final { public: ~ServerImpl() { server_->Shutdown(); // Always shutdown the completion queue after the server. cq_->Shutdown(); } // There is no shutdown handling in this code. void Run() { std::string server_address("0.0.0.0:50051"); ServerBuilder builder; // Listen on the given address without any authentication mechanism. builder.AddListeningPort(server_address, grpc::InsecureServerCredentials()); // Register "service_" as the instance through which we'll communicate with // clients. In this case it corresponds to an *asynchronous* service. builder.RegisterService(&service_); // Get hold of the completion queue used for the asynchronous communication // with the gRPC runtime. cq_ = builder.AddCompletionQueue(); // Finally assemble the server. server_ = builder.BuildAndStart(); std::cout << "Server listening on " << server_address << std::endl; // Proceed to the server's main loop. HandleRpcs(); } private: // Class encompasing the state and logic needed to serve a request. class CallData { public: // Take in the "service" instance (in this case representing an asynchronous // server) and the completion queue "cq" used for asynchronous communication // with the gRPC runtime. CallData(Greeter::AsyncService* service, ServerCompletionQueue* cq) : service_(service), cq_(cq), responder_(&ctx_), status_(CREATE) { // Invoke the serving logic right away. Proceed(); } void Proceed() { if (status_ == CREATE) { // Make this instance progress to the PROCESS state. status_ = PROCESS; // As part of the initial CREATE state, we *request* that the system // start processing SayHello requests. In this request, "this" acts are // the tag uniquely identifying the request (so that different CallData // instances can serve different requests concurrently), in this case // the memory address of this CallData instance. service_->RequestSayHello(&ctx_, &request_, &responder_, cq_, cq_, this); } else if (status_ == PROCESS) { // Spawn a new CallData instance to serve new clients while we process // the one for this CallData. The instance will deallocate itself as // part of its FINISH state. new CallData(service_, cq_); // The actual processing. std::string prefix("Hello "); reply_.set_message(prefix + request_.name()); // And we are done! Let the gRPC runtime know we've finished, using the // memory address of this instance as the uniquely identifying tag for // the event. status_ = FINISH; responder_.Finish(reply_, Status::OK, this); } else { GPR_ASSERT(status_ == FINISH); // Once in the FINISH state, deallocate ourselves (CallData). delete this; } } private: // The means of communication with the gRPC runtime for an asynchronous // server. Greeter::AsyncService* service_; // The producer-consumer queue where for asynchronous server notifications. ServerCompletionQueue* cq_; // Context for the rpc, allowing to tweak aspects of it such as the use // of compression, authentication, as well as to send metadata back to the // client. ServerContext ctx_; // What we get from the client. HelloRequest request_; // What we send back to the client. HelloReply reply_; // The means to get back to the client. ServerAsyncResponseWriter responder_; // Let's implement a tiny state machine with the following states. enum CallStatus { CREATE, PROCESS, FINISH }; CallStatus status_; // The current serving state. }; // This can be run in multiple threads if needed. void HandleRpcs() { // Spawn a new CallData instance to serve new clients. new CallData(&service_, cq_.get()); void* tag; // uniquely identifies a request. bool ok; while (true) { // Block waiting to read the next event from the completion queue. The // event is uniquely identified by its tag, which in this case is the // memory address of a CallData instance. cq_->Next(&tag, &ok); GPR_ASSERT(ok); static_cast(tag)->Proceed(); } } std::unique_ptr cq_; Greeter::AsyncService service_; std::unique_ptr server_; }; int main(int argc, char** argv) { ServerImpl server; server.Run(); return 0; }