From 8065000697a6225d76ebbcc8b7eca9f01a9be474 Mon Sep 17 00:00:00 2001 From: makdharma <> Date: Fri, 23 Feb 2018 14:51:46 -0800 Subject: Remove "final" keyword and make methods protected. This adds extensibility to the API and makes custom implementation of the server possible. --- include/grpcpp/impl/codegen/completion_queue.h | 10 +-- include/grpcpp/server.h | 97 +++++++++++++------------- include/grpcpp/server_builder.h | 36 +++++----- src/cpp/server/server_cc.cc | 20 +++--- 4 files changed, 86 insertions(+), 77 deletions(-) diff --git a/include/grpcpp/impl/codegen/completion_queue.h b/include/grpcpp/impl/codegen/completion_queue.h index 80c7c41982..5ba70e2e66 100644 --- a/include/grpcpp/impl/codegen/completion_queue.h +++ b/include/grpcpp/impl/codegen/completion_queue.h @@ -165,7 +165,7 @@ class CompletionQueue : private GrpcLibraryCodegen { /// /// \return true if got an event, false if the queue is fully drained and /// shut down. - bool Next(void** tag, bool* ok) { + virtual bool Next(void** tag, bool* ok) { return (AsyncNextInternal(tag, ok, g_core_codegen_interface->gpr_inf_future( GPR_CLOCK_REALTIME)) != SHUTDOWN); @@ -365,9 +365,7 @@ class ServerCompletionQueue : public CompletionQueue { public: bool IsFrequentlyPolled() { return polling_type_ != GRPC_CQ_NON_LISTENING; } - private: - grpc_cq_polling_type polling_type_; - friend class ServerBuilder; + protected: /// \param is_frequently_polled Informs the GRPC library about whether the /// server completion queue would be actively polled (by calling Next() or /// AsyncNext()). By default all server completion queues are assumed to be @@ -376,6 +374,10 @@ class ServerCompletionQueue : public CompletionQueue { : CompletionQueue(grpc_completion_queue_attributes{ GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, polling_type}), polling_type_(polling_type) {} + + private: + grpc_cq_polling_type polling_type_; + friend class ServerBuilder; }; } // namespace grpc diff --git a/include/grpcpp/server.h b/include/grpcpp/server.h index f99a6c26b4..c75b492734 100644 --- a/include/grpcpp/server.h +++ b/include/grpcpp/server.h @@ -49,7 +49,7 @@ class ServerInitializer; /// /// Use a \a grpc::ServerBuilder to create, configure, and start /// \a Server instances. -class Server final : public ServerInterface, private GrpcLibraryCodegen { +class Server : public ServerInterface, private GrpcLibraryCodegen { public: ~Server(); @@ -98,24 +98,26 @@ class Server final : public ServerInterface, private GrpcLibraryCodegen { /// Establish a channel for in-process communication std::shared_ptr InProcessChannel(const ChannelArguments& args); - private: - friend class AsyncGenericService; - friend class ServerBuilder; - friend class ServerInitializer; - - class SyncRequest; - class AsyncRequest; - class ShutdownRequest; - - /// SyncRequestThreadManager is an implementation of ThreadManager. This class - /// is responsible for polling for incoming RPCs and calling the RPC handlers. - /// This is only used in case of a Sync server (i.e a server exposing a sync - /// interface) - class SyncRequestThreadManager; +protected: + /// Register a service. This call does not take ownership of the service. + /// The service must exist for the lifetime of the Server instance. + bool RegisterService(const grpc::string* host, Service* service) override; - class UnimplementedAsyncRequestContext; - class UnimplementedAsyncRequest; - class UnimplementedAsyncResponse; + /// Try binding the server to the given \a addr endpoint + /// (port, and optionally including IP address to bind to). + /// + /// It can be invoked multiple times. Should be used before + /// starting the server. + /// + /// \param addr The address to try to bind to the server (eg, localhost:1234, + /// 192.168.1.1:31416, [::1]:27182, etc.). + /// \param creds The credentials associated with the server. + /// + /// \return bound port number on success, 0 on failure. + /// + /// \warning It is an error to call this method on an already started server. + int AddListeningPort(const grpc::string& addr, + ServerCredentials* creds) override; /// Server constructors. To be used by \a ServerBuilder only. /// @@ -143,30 +145,6 @@ class Server final : public ServerInterface, private GrpcLibraryCodegen { sync_server_cqs, int min_pollers, int max_pollers, int sync_cq_timeout_msec); - /// Register a service. This call does not take ownership of the service. - /// The service must exist for the lifetime of the Server instance. - bool RegisterService(const grpc::string* host, Service* service) override; - - /// Register a generic service. This call does not take ownership of the - /// service. The service must exist for the lifetime of the Server instance. - void RegisterAsyncGenericService(AsyncGenericService* service) override; - - /// Try binding the server to the given \a addr endpoint - /// (port, and optionally including IP address to bind to). - /// - /// It can be invoked multiple times. Should be used before - /// starting the server. - /// - /// \param addr The address to try to bind to the server (eg, localhost:1234, - /// 192.168.1.1:31416, [::1]:27182, etc.). - /// \param creds The credentials associated with the server. - /// - /// \return bound port number on success, 0 on failure. - /// - /// \warning It is an error to call this method on an already started server. - int AddListeningPort(const grpc::string& addr, - ServerCredentials* creds) override; - /// Start the server. /// /// \param cqs Completion queues for handling asynchronous services. The @@ -175,6 +153,35 @@ class Server final : public ServerInterface, private GrpcLibraryCodegen { /// \param num_cqs How many completion queues does \a cqs hold. void Start(ServerCompletionQueue** cqs, size_t num_cqs) override; + // Pointer to the wrapped grpc_server. + grpc_server* server_; + + // Server status + bool started_; + + private: + friend class AsyncGenericService; + friend class ServerBuilder; + friend class ServerInitializer; + + class SyncRequest; + class AsyncRequest; + class ShutdownRequest; + + /// SyncRequestThreadManager is an implementation of ThreadManager. This class + /// is responsible for polling for incoming RPCs and calling the RPC handlers. + /// This is only used in case of a Sync server (i.e a server exposing a sync + /// interface) + class SyncRequestThreadManager; + + class UnimplementedAsyncRequestContext; + class UnimplementedAsyncRequest; + class UnimplementedAsyncResponse; + + /// Register a generic service. This call does not take ownership of the + /// service. The service must exist for the lifetime of the Server instance. + void RegisterAsyncGenericService(AsyncGenericService* service) override; + void PerformOpsOnCall(internal::CallOpSetInterface* ops, internal::Call* call) override; @@ -200,9 +207,8 @@ class Server final : public ServerInterface, private GrpcLibraryCodegen { /// the \a sync_server_cqs) std::vector> sync_req_mgrs_; - // Sever status + // Server status std::mutex mu_; - bool started_; bool shutdown_; bool shutdown_notified_; // Was notify called on the shutdown_cv_ @@ -213,9 +219,6 @@ class Server final : public ServerInterface, private GrpcLibraryCodegen { std::vector services_; bool has_generic_service_; - // Pointer to the wrapped grpc_server. - grpc_server* server_; - std::unique_ptr server_initializer_; std::unique_ptr health_check_service_; diff --git a/include/grpcpp/server_builder.h b/include/grpcpp/server_builder.h index c35a6cf98a..8f148bcd06 100644 --- a/include/grpcpp/server_builder.h +++ b/include/grpcpp/server_builder.h @@ -52,7 +52,7 @@ class ServerBuilderPluginTest; class ServerBuilder { public: ServerBuilder(); - ~ServerBuilder(); + virtual ~ServerBuilder(); ////////////////////////////////////////////////////////////////////////////// // Primary API's @@ -65,7 +65,7 @@ class ServerBuilder { /// traffic (via AddListeningPort) /// 3. [for async api only] completion queues have been added via /// AddCompletionQueue - std::unique_ptr BuildAndStart(); + virtual std::unique_ptr BuildAndStart(); /// Register a service. This call does not take ownership of the service. /// The service must exist for the lifetime of the \a Server instance returned @@ -210,15 +210,29 @@ class ServerBuilder { /// doc/workarounds.md. ServerBuilder& EnableWorkaround(grpc_workaround_list id); - private: - friend class ::grpc::testing::ServerBuilderPluginTest; - + protected: struct Port { grpc::string addr; std::shared_ptr creds; int* selected_port; }; + typedef std::unique_ptr HostString; + struct NamedService { + explicit NamedService(Service* s) : service(s) {} + NamedService(const grpc::string& h, Service* s) + : host(new grpc::string(h)), service(s) {} + HostString host; + Service* service; + }; + + std::vector> options_; + std::vector> services_; + std::vector ports_; + + private: + friend class ::grpc::testing::ServerBuilderPluginTest; + struct SyncServerSettings { SyncServerSettings() : num_cqs(1), min_pollers(1), max_pollers(2), cq_timeout_msec(10000) {} @@ -238,20 +252,8 @@ class ServerBuilder { int cq_timeout_msec; }; - typedef std::unique_ptr HostString; - struct NamedService { - explicit NamedService(Service* s) : service(s) {} - NamedService(const grpc::string& h, Service* s) - : host(new grpc::string(h)), service(s) {} - HostString host; - Service* service; - }; - int max_receive_message_size_; int max_send_message_size_; - std::vector> options_; - std::vector> services_; - std::vector ports_; SyncServerSettings sync_server_settings_; diff --git a/src/cpp/server/server_cc.cc b/src/cpp/server/server_cc.cc index 5638636e67..067f292b6e 100644 --- a/src/cpp/server/server_cc.cc +++ b/src/cpp/server/server_cc.cc @@ -368,13 +368,13 @@ Server::Server( std::shared_ptr>> sync_server_cqs, int min_pollers, int max_pollers, int sync_cq_timeout_msec) - : max_receive_message_size_(max_receive_message_size), - sync_server_cqs_(sync_server_cqs), + : server_(nullptr), started_(false), + max_receive_message_size_(max_receive_message_size), + sync_server_cqs_(sync_server_cqs), shutdown_(false), shutdown_notified_(false), has_generic_service_(false), - server_(nullptr), server_initializer_(new ServerInitializer(this)), health_check_service_disabled_(false) { g_gli_initializer.summon(); @@ -382,11 +382,13 @@ Server::Server( global_callbacks_ = g_callbacks; global_callbacks_->UpdateArguments(args); - for (auto it = sync_server_cqs_->begin(); it != sync_server_cqs_->end(); - it++) { - sync_req_mgrs_.emplace_back(new SyncRequestThreadManager( - this, (*it).get(), global_callbacks_, min_pollers, max_pollers, - sync_cq_timeout_msec)); + if (sync_server_cqs_ != nullptr) { + for (auto it = sync_server_cqs_->begin(); it != sync_server_cqs_->end(); + it++) { + sync_req_mgrs_.emplace_back(new SyncRequestThreadManager( + this, (*it).get(), global_callbacks_, min_pollers, max_pollers, + sync_cq_timeout_msec)); + } } grpc_channel_args channel_args; @@ -525,7 +527,7 @@ void Server::Start(ServerCompletionQueue** cqs, size_t num_cqs) { // explicit one. if (health_check_service_ == nullptr && !health_check_service_disabled_ && DefaultHealthCheckServiceEnabled()) { - if (sync_server_cqs_->empty()) { + if (sync_server_cqs_ == nullptr || sync_server_cqs_->empty()) { gpr_log(GPR_INFO, "Default health check service disabled at async-only server."); } else { -- cgit v1.2.3