include/grpcpp/impl/codegen/server_interface.h

   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 #ifndef GRPCPP_IMPL_CODEGEN_SERVER_INTERFACE_H
  20 #define GRPCPP_IMPL_CODEGEN_SERVER_INTERFACE_H
  21
  22 #include <grpc/impl/codegen/port_platform.h>
  23
  24 #include <grpc/impl/codegen/grpc_types.h>
  25 #include <grpcpp/impl/codegen/byte_buffer.h>
  26 #include <grpcpp/impl/codegen/call.h>
  27 #include <grpcpp/impl/codegen/call_hook.h>
  28 #include <grpcpp/impl/codegen/completion_queue_tag.h>
  29 #include <grpcpp/impl/codegen/core_codegen_interface.h>
  30 #include <grpcpp/impl/codegen/interceptor_common.h>
  31 #include <grpcpp/impl/codegen/rpc_service_method.h>
  32 #include <grpcpp/impl/codegen/server_context.h>
  33
  34 namespace grpc {
  35
  36 class AsyncGenericService;
  37 class Channel;
  38 class CompletionQueue;
  39 class GenericServerContext;
  40 class ServerCompletionQueue;
  41 class ServerCredentials;
  42 class Service;
  43
  44 extern CoreCodegenInterface* g_core_codegen_interface;
  45
  46 /// Models a gRPC server.
  47 ///
  48 /// Servers are configured and started via \a grpc::ServerBuilder.
  49 namespace internal {
  50 class ServerAsyncStreamingInterface;
  51 }  // namespace internal
  52
  53 class CallbackGenericService;
  54
  55 namespace experimental {
  56 class ServerInterceptorFactoryInterface;
  57 }  // namespace experimental
  58
  59 class ServerInterface : public internal::CallHook {
  60  public:
  61   ~ServerInterface() override {}
  62
  63   /// \a Shutdown does the following things:
  64   ///
  65   /// 1. Shutdown the server: deactivate all listening ports, mark it in
  66   ///    "shutdown mode" so that further call Request's or incoming RPC matches
  67   ///    are no longer allowed. Also return all Request'ed-but-not-yet-active
  68   ///    calls as failed (!ok). This refers to calls that have been requested
  69   ///    at the server by the server-side library or application code but that
  70   ///    have not yet been matched to incoming RPCs from the client. Note that
  71   ///    this would even include default calls added automatically by the gRPC
  72   ///    C++ API without the user's input (e.g., "Unimplemented RPC method")
  73   ///
  74   /// 2. Block until all rpc method handlers invoked automatically by the sync
  75   ///    API finish.
  76   ///
  77   /// 3. If all pending calls complete (and all their operations are
  78   ///    retrieved by Next) before \a deadline expires, this finishes
  79   ///    gracefully. Otherwise, forcefully cancel all pending calls associated
  80   ///    with the server after \a deadline expires. In the case of the sync API,
  81   ///    if the RPC function for a streaming call has already been started and
  82   ///    takes a week to complete, the RPC function won't be forcefully
  83   ///    terminated (since that would leave state corrupt and incomplete) and
  84   ///    the method handler will just keep running (which will prevent the
  85   ///    server from completing the "join" operation that it needs to do at
  86   ///    shutdown time).
  87   ///
  88   /// All completion queue associated with the server (for example, for async
  89   /// serving) must be shutdown *after* this method has returned:
  90   /// See \a ServerBuilder::AddCompletionQueue for details.
  91   /// They must also be drained (by repeated Next) after being shutdown.
  92   ///
  93   /// \param deadline How long to wait until pending rpcs are forcefully
  94   /// terminated.
  95   template <class T>
  96   void Shutdown(const T& deadline) {
  97     ShutdownInternal(TimePoint<T>(deadline).raw_time());
  98   }
  99
 100   /// Shutdown the server without a deadline and forced cancellation.
 101   ///
 102   /// All completion queue associated with the server (for example, for async
 103   /// serving) must be shutdown *after* this method has returned:
 104   /// See \a ServerBuilder::AddCompletionQueue for details.
 105   void Shutdown() {
 106     ShutdownInternal(
 107         g_core_codegen_interface->gpr_inf_future(GPR_CLOCK_MONOTONIC));
 108   }
 109
 110   /// Block waiting for all work to complete.
 111   ///
 112   /// \warning The server must be either shutting down or some other thread must
 113   /// call \a Shutdown for this function to ever return.
 114   virtual void Wait() = 0;
 115
 116  protected:
 117   friend class ::grpc::Service;
 118
 119   /// Register a service. This call does not take ownership of the service.
 120   /// The service must exist for the lifetime of the Server instance.
 121   virtual bool RegisterService(const std::string* host, Service* service) = 0;
 122
 123   /// Register a generic service. This call does not take ownership of the
 124   /// service. The service must exist for the lifetime of the Server instance.
 125   virtual void RegisterAsyncGenericService(AsyncGenericService* service) = 0;
 126
 127   /// Register a callback generic service. This call does not take ownership of
 128   /// the  service. The service must exist for the lifetime of the Server
 129   /// instance. May not be abstract since this is a post-1.0 API addition.
 130
 131   virtual void RegisterCallbackGenericService(CallbackGenericService*
 132                                               /*service*/) {}
 133
 134   /// Tries to bind \a server to the given \a addr.
 135   ///
 136   /// It can be invoked multiple times.
 137   ///
 138   /// \param addr The address to try to bind to the server (eg, localhost:1234,
 139   /// 192.168.1.1:31416, [::1]:27182, etc.).
 140   /// \params creds The credentials associated with the server.
 141   ///
 142   /// \return bound port number on success, 0 on failure.
 143   ///
 144   /// \warning It's an error to call this method on an already started server.
 145   virtual int AddListeningPort(const std::string& addr,
 146                                ServerCredentials* creds) = 0;
 147
 148   /// Start the server.
 149   ///
 150   /// \param cqs Completion queues for handling asynchronous services. The
 151   /// caller is required to keep all completion queues live until the server is
 152   /// destroyed.
 153   /// \param num_cqs How many completion queues does \a cqs hold.
 154   virtual void Start(::grpc::ServerCompletionQueue** cqs, size_t num_cqs) = 0;
 155
 156   virtual void ShutdownInternal(gpr_timespec deadline) = 0;
 157
 158   virtual int max_receive_message_size() const = 0;
 159
 160   virtual grpc_server* server() = 0;
 161
 162   void PerformOpsOnCall(internal::CallOpSetInterface* ops,
 163                         internal::Call* call) override = 0;
 164
 165   class BaseAsyncRequest : public internal::CompletionQueueTag {
 166    public:
 167     BaseAsyncRequest(ServerInterface* server, ::grpc::ServerContext* context,
 168                      internal::ServerAsyncStreamingInterface* stream,
 169                      ::grpc::CompletionQueue* call_cq,
 170                      ::grpc::ServerCompletionQueue* notification_cq, void* tag,
 171                      bool delete_on_finalize);
 172     ~BaseAsyncRequest() override;
 173
 174     bool FinalizeResult(void** tag, bool* status) override;
 175
 176    private:
 177     void ContinueFinalizeResultAfterInterception();
 178
 179    protected:
 180     ServerInterface* const server_;
 181     ::grpc::ServerContext* const context_;
 182     internal::ServerAsyncStreamingInterface* const stream_;
 183     ::grpc::CompletionQueue* const call_cq_;
 184     ::grpc::ServerCompletionQueue* const notification_cq_;
 185     void* const tag_;
 186     const bool delete_on_finalize_;
 187     grpc_call* call_;
 188     internal::Call call_wrapper_;
 189     internal::InterceptorBatchMethodsImpl interceptor_methods_;
 190     bool done_intercepting_;
 191   };
 192
 193   /// RegisteredAsyncRequest is not part of the C++ API
 194   class RegisteredAsyncRequest : public BaseAsyncRequest {
 195    public:
 196     RegisteredAsyncRequest(ServerInterface* server,
 197                            ::grpc::ServerContext* context,
 198                            internal::ServerAsyncStreamingInterface* stream,
 199                            ::grpc::CompletionQueue* call_cq,
 200                            ::grpc::ServerCompletionQueue* notification_cq,
 201                            void* tag, const char* name,
 202                            internal::RpcMethod::RpcType type);
 203
 204     bool FinalizeResult(void** tag, bool* status) override {
 205       /* If we are done intercepting, then there is nothing more for us to do */
 206       if (done_intercepting_) {
 207         return BaseAsyncRequest::FinalizeResult(tag, status);
 208       }
 209       call_wrapper_ = ::grpc::internal::Call(
 210           call_, server_, call_cq_, server_->max_receive_message_size(),
 211           context_->set_server_rpc_info(name_, type_,
 212                                         *server_->interceptor_creators()));
 213       return BaseAsyncRequest::FinalizeResult(tag, status);
 214     }
 215
 216    protected:
 217     void IssueRequest(void* registered_method, grpc_byte_buffer** payload,
 218                       ::grpc::ServerCompletionQueue* notification_cq);
 219     const char* name_;
 220     const internal::RpcMethod::RpcType type_;
 221   };
 222
 223   class NoPayloadAsyncRequest final : public RegisteredAsyncRequest {
 224    public:
 225     NoPayloadAsyncRequest(internal::RpcServiceMethod* registered_method,
 226                           ServerInterface* server,
 227                           ::grpc::ServerContext* context,
 228                           internal::ServerAsyncStreamingInterface* stream,
 229                           ::grpc::CompletionQueue* call_cq,
 230                           ::grpc::ServerCompletionQueue* notification_cq,
 231                           void* tag)
 232         : RegisteredAsyncRequest(
 233               server, context, stream, call_cq, notification_cq, tag,
 234               registered_method->name(), registered_method->method_type()) {
 235       IssueRequest(registered_method->server_tag(), nullptr, notification_cq);
 236     }
 237
 238     // uses RegisteredAsyncRequest::FinalizeResult
 239   };
 240
 241   template <class Message>
 242   class PayloadAsyncRequest final : public RegisteredAsyncRequest {
 243    public:
 244     PayloadAsyncRequest(internal::RpcServiceMethod* registered_method,
 245                         ServerInterface* server, ::grpc::ServerContext* context,
 246                         internal::ServerAsyncStreamingInterface* stream,
 247                         ::grpc::CompletionQueue* call_cq,
 248                         ::grpc::ServerCompletionQueue* notification_cq,
 249                         void* tag, Message* request)
 250         : RegisteredAsyncRequest(
 251               server, context, stream, call_cq, notification_cq, tag,
 252               registered_method->name(), registered_method->method_type()),
 253           registered_method_(registered_method),
 254           request_(request) {
 255       IssueRequest(registered_method->server_tag(), payload_.bbuf_ptr(),
 256                    notification_cq);
 257     }
 258
 259     ~PayloadAsyncRequest() override {
 260       payload_.Release();  // We do not own the payload_
 261     }
 262
 263     bool FinalizeResult(void** tag, bool* status) override {
 264       /* If we are done intercepting, then there is nothing more for us to do */
 265       if (done_intercepting_) {
 266         return RegisteredAsyncRequest::FinalizeResult(tag, status);
 267       }
 268       if (*status) {
 269         if (!payload_.Valid() || !SerializationTraits<Message>::Deserialize(
 270                                       payload_.bbuf_ptr(), request_)
 271                                       .ok()) {
 272           // If deserialization fails, we cancel the call and instantiate
 273           // a new instance of ourselves to request another call.  We then
 274           // return false, which prevents the call from being returned to
 275           // the application.
 276           g_core_codegen_interface->grpc_call_cancel_with_status(
 277               call_, GRPC_STATUS_INTERNAL, "Unable to parse request", nullptr);
 278           g_core_codegen_interface->grpc_call_unref(call_);
 279           new PayloadAsyncRequest(registered_method_, server_, context_,
 280                                   stream_, call_cq_, notification_cq_, tag_,
 281                                   request_);
 282           delete this;
 283           return false;
 284         }
 285       }
 286       /* Set interception point for recv message */
 287       interceptor_methods_.AddInterceptionHookPoint(
 288           experimental::InterceptionHookPoints::POST_RECV_MESSAGE);
 289       interceptor_methods_.SetRecvMessage(request_, nullptr);
 290       return RegisteredAsyncRequest::FinalizeResult(tag, status);
 291     }
 292
 293    private:
 294     internal::RpcServiceMethod* const registered_method_;
 295     Message* const request_;
 296     ByteBuffer payload_;
 297   };
 298
 299   class GenericAsyncRequest : public BaseAsyncRequest {
 300    public:
 301     GenericAsyncRequest(ServerInterface* server, GenericServerContext* context,
 302                         internal::ServerAsyncStreamingInterface* stream,
 303                         ::grpc::CompletionQueue* call_cq,
 304                         ::grpc::ServerCompletionQueue* notification_cq,
 305                         void* tag, bool delete_on_finalize);
 306
 307     bool FinalizeResult(void** tag, bool* status) override;
 308
 309    private:
 310     grpc_call_details call_details_;
 311   };
 312
 313   template <class Message>
 314   void RequestAsyncCall(internal::RpcServiceMethod* method,
 315                         ::grpc::ServerContext* context,
 316                         internal::ServerAsyncStreamingInterface* stream,
 317                         ::grpc::CompletionQueue* call_cq,
 318                         ::grpc::ServerCompletionQueue* notification_cq,
 319                         void* tag, Message* message) {
 320     GPR_CODEGEN_ASSERT(method);
 321     new PayloadAsyncRequest<Message>(method, this, context, stream, call_cq,
 322                                      notification_cq, tag, message);
 323   }
 324
 325   void RequestAsyncCall(internal::RpcServiceMethod* method,
 326                         ::grpc::ServerContext* context,
 327                         internal::ServerAsyncStreamingInterface* stream,
 328                         ::grpc::CompletionQueue* call_cq,
 329                         ::grpc::ServerCompletionQueue* notification_cq,
 330                         void* tag) {
 331     GPR_CODEGEN_ASSERT(method);
 332     new NoPayloadAsyncRequest(method, this, context, stream, call_cq,
 333                               notification_cq, tag);
 334   }
 335
 336   void RequestAsyncGenericCall(GenericServerContext* context,
 337                                internal::ServerAsyncStreamingInterface* stream,
 338                                ::grpc::CompletionQueue* call_cq,
 339                                ::grpc::ServerCompletionQueue* notification_cq,
 340                                void* tag) {
 341     new GenericAsyncRequest(this, context, stream, call_cq, notification_cq,
 342                             tag, true);
 343   }
 344
 345  private:
 346   // EXPERIMENTAL
 347   // Getter method for the vector of interceptor factory objects.
 348   // Returns a nullptr (rather than being pure) since this is a post-1.0 method
 349   // and adding a new pure method to an interface would be a breaking change
 350   // (even though this is private and non-API)
 351   virtual std::vector<
 352       std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>*
 353   interceptor_creators() {
 354     return nullptr;
 355   }
 356
 357   // A method to get the callbackable completion queue associated with this
 358   // server. If the return value is nullptr, this server doesn't support
 359   // callback operations.
 360   // TODO(vjpai): Consider a better default like using a global CQ
 361   // Returns nullptr (rather than being pure) since this is a post-1.0 method
 362   // and adding a new pure method to an interface would be a breaking change
 363   // (even though this is private and non-API)
 364   virtual ::grpc::CompletionQueue* CallbackCQ() { return nullptr; }
 365 };
 366
 367 }  // namespace grpc
 368
 369 #endif  // GRPCPP_IMPL_CODEGEN_SERVER_INTERFACE_H