grpc async server toy example

greeter_async_server.cc

// Class encompasing the state and logic needed to serve a request.
class CallDataBase {
 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.
  CallDataBase(Greeter::AsyncService* service, ServerCompletionQueue* cq)
      : service_(service), cq_(cq), status_(PROCESS) {}
  virtual ~CallDataBase() {}

  void Proceed() {
    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.
      status_ = FINISH;
      Process();
    } else {
      GPR_ASSERT(status_ == FINISH);
      // Once in the FINISH state, deallocate ourselves (CallData).
      delete this;
    }
  }

 protected:
  virtual void Process() = 0;
  // 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_;

 private:
  // Let's implement a tiny state machine with the following states.
  enum CallStatus { PROCESS, FINISH };
  CallStatus status_;  // The current serving state.
};

class HelloCallData : public CallDataBase {
 public:
  HelloCallData(Greeter::AsyncService* service, ServerCompletionQueue* cq) : CallDataBase(service, cq) {
    service->RequestSayHello(&ctx, &request_, &responder_, cq, cq, this);
  }
  
  void Process() override {
    new HelloCallData(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.
    responder_.Finish(reply_, Status::OK, this);
  }

 private:
  ServerContext ctx_;
  HelloRequest request_;
  HelloReply reply_;
  ServerAsyncResponseWriter<HelloReply> responder_;  
};

class ByeCallData : public CallDataBase {
  // The same as HelloCallData but with SayBye types and logic.
};

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:
  // This can be run in multiple threads if needed.
  void HandleRpcs() {
    new HelloCallData(&service_, cq_.get());
    new ByeCallData(&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.
      // The return value of Next should always be checked. This return value
      // tells us whether there is any kind of event or cq_ is shutting down.
      GPR_ASSERT(cq_->Next(&tag, &ok));
      GPR_ASSERT(ok);
      static_cast<CallDataBase*>(tag)->Proceed();
    }
  }

  std::unique_ptr<ServerCompletionQueue> cq_;
  Greeter::AsyncService service_;
  std::unique_ptr<Server> server_;
};