Remove "final" keyword and make methods protected.

This adds extensibility to the API and makes custom implementation
of the server possible.
pull/14517/head
makdharma 7 years ago
parent 5699cfcd45
commit 8065000697
  1. 10
      include/grpcpp/impl/codegen/completion_queue.h
  2. 97
      include/grpcpp/server.h
  3. 36
      include/grpcpp/server_builder.h
  4. 20
      src/cpp/server/server_cc.cc

@ -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

@ -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<Channel> 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<std::unique_ptr<SyncRequestThreadManager>> 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<grpc::string> services_;
bool has_generic_service_;
// Pointer to the wrapped grpc_server.
grpc_server* server_;
std::unique_ptr<ServerInitializer> server_initializer_;
std::unique_ptr<HealthCheckServiceInterface> health_check_service_;

@ -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<Server> BuildAndStart();
virtual std::unique_ptr<Server> 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<ServerCredentials> creds;
int* selected_port;
};
typedef std::unique_ptr<grpc::string> 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<std::unique_ptr<ServerBuilderOption>> options_;
std::vector<std::unique_ptr<NamedService>> services_;
std::vector<Port> 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<grpc::string> 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<std::unique_ptr<ServerBuilderOption>> options_;
std::vector<std::unique_ptr<NamedService>> services_;
std::vector<Port> ports_;
SyncServerSettings sync_server_settings_;

@ -368,13 +368,13 @@ Server::Server(
std::shared_ptr<std::vector<std::unique_ptr<ServerCompletionQueue>>>
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 {

Loading…
Cancel
Save