diff options
Diffstat (limited to 'include/grpc++')
-rw-r--r-- | include/grpc++/create_channel.h | 1 | ||||
-rw-r--r-- | include/grpc++/ext/reflection.grpc.pb.h | 2 | ||||
-rw-r--r-- | include/grpc++/grpc++.h | 4 | ||||
-rw-r--r-- | include/grpc++/impl/codegen/call.h | 6 | ||||
-rw-r--r-- | include/grpc++/impl/codegen/client_context.h | 6 | ||||
-rw-r--r-- | include/grpc++/impl/codegen/completion_queue.h | 10 | ||||
-rw-r--r-- | include/grpc++/impl/codegen/method_handler_impl.h | 64 | ||||
-rw-r--r-- | include/grpc++/impl/codegen/rpc_method.h | 3 | ||||
-rw-r--r-- | include/grpc++/impl/codegen/rpc_service_method.h | 1 | ||||
-rw-r--r-- | include/grpc++/impl/codegen/server_context.h | 14 | ||||
-rw-r--r-- | include/grpc++/impl/codegen/service_type.h | 11 | ||||
-rw-r--r-- | include/grpc++/impl/codegen/sync_stream.h | 138 | ||||
-rw-r--r-- | include/grpc++/impl/server_builder_plugin.h | 5 |
13 files changed, 226 insertions, 39 deletions
diff --git a/include/grpc++/create_channel.h b/include/grpc++/create_channel.h index e9ccb51503..0537695ed2 100644 --- a/include/grpc++/create_channel.h +++ b/include/grpc++/create_channel.h @@ -48,7 +48,6 @@ namespace grpc { /// \param target The URI of the endpoint to connect to. /// \param creds Credentials to use for the created channel. If it does not hold /// an object or is invalid, a lame channel is returned. -/// \param args Options for channel creation. std::shared_ptr<Channel> CreateChannel( const grpc::string& target, const std::shared_ptr<ChannelCredentials>& creds); diff --git a/include/grpc++/ext/reflection.grpc.pb.h b/include/grpc++/ext/reflection.grpc.pb.h index 0b4ef86147..064117e303 100644 --- a/include/grpc++/ext/reflection.grpc.pb.h +++ b/include/grpc++/ext/reflection.grpc.pb.h @@ -74,6 +74,7 @@ #include <grpc++/impl/codegen/async_stream.h> #include <grpc++/impl/codegen/async_unary_call.h> +#include <grpc++/impl/codegen/method_handler_impl.h> #include <grpc++/impl/codegen/proto_utils.h> #include <grpc++/impl/codegen/rpc_method.h> #include <grpc++/impl/codegen/service_type.h> @@ -174,6 +175,7 @@ class ServerReflection GRPC_FINAL { return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); } }; + typedef Service StreamedUnaryService; }; } // namespace v1alpha diff --git a/include/grpc++/grpc++.h b/include/grpc++/grpc++.h index ef07e19975..afb1c555bb 100644 --- a/include/grpc++/grpc++.h +++ b/include/grpc++/grpc++.h @@ -51,6 +51,9 @@ #ifndef GRPCXX_GRPCXX_H #define GRPCXX_GRPCXX_H +// Pragma for http://include-what-you-use.org/ tool, tells that following +// headers are not private for grpc++.h and are part of its interface. +// IWYU pragma: begin_exports #include <grpc/grpc.h> #include <grpc++/channel.h> @@ -62,5 +65,6 @@ #include <grpc++/server_builder.h> #include <grpc++/server_context.h> #include <grpc++/server_posix.h> +// IWYU pragma: end_exports #endif // GRPCXX_GRPCXX_H diff --git a/include/grpc++/impl/codegen/call.h b/include/grpc++/impl/codegen/call.h index dfac177970..df225d362b 100644 --- a/include/grpc++/impl/codegen/call.h +++ b/include/grpc++/impl/codegen/call.h @@ -662,10 +662,10 @@ class Call GRPC_FINAL { call_hook_->PerformOpsOnCall(ops, this); } - grpc_call* call() { return call_; } - CompletionQueue* cq() { return cq_; } + grpc_call* call() const { return call_; } + CompletionQueue* cq() const { return cq_; } - int max_message_size() { return max_message_size_; } + int max_message_size() const { return max_message_size_; } private: CallHook* call_hook_; diff --git a/include/grpc++/impl/codegen/client_context.h b/include/grpc++/impl/codegen/client_context.h index 012bcc2bbe..d77ca4c396 100644 --- a/include/grpc++/impl/codegen/client_context.h +++ b/include/grpc++/impl/codegen/client_context.h @@ -271,7 +271,7 @@ class ClientContext { /// Set \a algorithm to be the compression algorithm used for the client call. /// - /// \param algorith The compression algorithm used for the client call. + /// \param algorithm The compression algorithm used for the client call. void set_compression_algorithm(grpc_compression_algorithm algorithm); /// Return the peer uri in a string. @@ -307,6 +307,10 @@ class ClientContext { }; static void SetGlobalCallbacks(GlobalCallbacks* callbacks); + // Should be used for framework-level extensions only. + // Applications never need to call this method. + grpc_call* c_call() { return call_; } + private: // Disallow copy and assign. ClientContext(const ClientContext&); diff --git a/include/grpc++/impl/codegen/completion_queue.h b/include/grpc++/impl/codegen/completion_queue.h index 03009e0561..78bc5ca481 100644 --- a/include/grpc++/impl/codegen/completion_queue.h +++ b/include/grpc++/impl/codegen/completion_queue.h @@ -68,8 +68,10 @@ template <class R> class ServerReader; template <class W> class ServerWriter; +namespace internal { template <class W, class R> -class ServerReaderWriter; +class ServerReaderWriterBody; +} template <class ServiceType, class RequestType, class ResponseType> class RpcMethodHandler; template <class ServiceType, class RequestType, class ResponseType> @@ -178,15 +180,15 @@ class CompletionQueue : private GrpcLibraryCodegen { template <class W> friend class ::grpc::ServerWriter; template <class W, class R> - friend class ::grpc::ServerReaderWriter; + friend class ::grpc::internal::ServerReaderWriterBody; template <class ServiceType, class RequestType, class ResponseType> friend class RpcMethodHandler; template <class ServiceType, class RequestType, class ResponseType> friend class ClientStreamingHandler; template <class ServiceType, class RequestType, class ResponseType> friend class ServerStreamingHandler; - template <class ServiceType, class RequestType, class ResponseType> - friend class BidiStreamingHandler; + template <class Streamer, bool WriteNeeded> + friend class TemplatedBidiStreamingHandler; friend class UnknownMethodHandler; friend class ::grpc::Server; friend class ::grpc::ServerContext; diff --git a/include/grpc++/impl/codegen/method_handler_impl.h b/include/grpc++/impl/codegen/method_handler_impl.h index 2f4be644ba..d989263252 100644 --- a/include/grpc++/impl/codegen/method_handler_impl.h +++ b/include/grpc++/impl/codegen/method_handler_impl.h @@ -167,20 +167,22 @@ class ServerStreamingHandler : public MethodHandler { }; // A wrapper class of an application provided bidi-streaming handler. -template <class ServiceType, class RequestType, class ResponseType> -class BidiStreamingHandler : public MethodHandler { +// This also applies to server-streamed implementation of a unary method +// with the additional requirement that such methods must have done a +// write for status to be ok +// Since this is used by more than 1 class, the service is not passed in. +// Instead, it is expected to be an implicitly-captured argument of func +// (through bind or something along those lines) +template <class Streamer, bool WriteNeeded> +class TemplatedBidiStreamingHandler : public MethodHandler { public: - BidiStreamingHandler( - std::function<Status(ServiceType*, ServerContext*, - ServerReaderWriter<ResponseType, RequestType>*)> - func, - ServiceType* service) - : func_(func), service_(service) {} + TemplatedBidiStreamingHandler( + std::function<Status(ServerContext*, Streamer*)> func) + : func_(func), write_needed_(WriteNeeded) {} void RunHandler(const HandlerParameter& param) GRPC_FINAL { - ServerReaderWriter<ResponseType, RequestType> stream(param.call, - param.server_context); - Status status = func_(service_, param.server_context, &stream); + Streamer stream(param.call, param.server_context); + Status status = func_(param.server_context, &stream); CallOpSet<CallOpSendInitialMetadata, CallOpServerSendStatus> ops; if (!param.server_context->sent_initial_metadata_) { @@ -189,6 +191,12 @@ class BidiStreamingHandler : public MethodHandler { if (param.server_context->compression_level_set()) { ops.set_compression_level(param.server_context->compression_level()); } + if (write_needed_ && status.ok()) { + // If we needed a write but never did one, we need to mark the + // status as a fail + status = Status(StatusCode::INTERNAL, + "Service did not provide response message"); + } } ops.ServerSendStatus(param.server_context->trailing_metadata_, status); param.call->PerformOps(&ops); @@ -196,10 +204,36 @@ class BidiStreamingHandler : public MethodHandler { } private: - std::function<Status(ServiceType*, ServerContext*, - ServerReaderWriter<ResponseType, RequestType>*)> - func_; - ServiceType* service_; + std::function<Status(ServerContext*, Streamer*)> func_; + const bool write_needed_; +}; + +template <class ServiceType, class RequestType, class ResponseType> +class BidiStreamingHandler + : public TemplatedBidiStreamingHandler< + ServerReaderWriter<ResponseType, RequestType>, false> { + public: + BidiStreamingHandler( + std::function<Status(ServiceType*, ServerContext*, + ServerReaderWriter<ResponseType, RequestType>*)> + func, + ServiceType* service) + : TemplatedBidiStreamingHandler< + ServerReaderWriter<ResponseType, RequestType>, false>(std::bind( + func, service, std::placeholders::_1, std::placeholders::_2)) {} +}; + +template <class RequestType, class ResponseType> +class StreamedUnaryHandler + : public TemplatedBidiStreamingHandler< + ServerUnaryStreamer<RequestType, ResponseType>, true> { + public: + explicit StreamedUnaryHandler( + std::function<Status(ServerContext*, + ServerUnaryStreamer<RequestType, ResponseType>*)> + func) + : TemplatedBidiStreamingHandler< + ServerUnaryStreamer<RequestType, ResponseType>, true>(func) {} }; // Handle unknown method by returning UNIMPLEMENTED error. diff --git a/include/grpc++/impl/codegen/rpc_method.h b/include/grpc++/impl/codegen/rpc_method.h index 39cb4f75df..4897428074 100644 --- a/include/grpc++/impl/codegen/rpc_method.h +++ b/include/grpc++/impl/codegen/rpc_method.h @@ -60,11 +60,12 @@ class RpcMethod { const char* name() const { return name_; } RpcType method_type() const { return method_type_; } + void SetMethodType(RpcType type) { method_type_ = type; } void* channel_tag() const { return channel_tag_; } private: const char* const name_; - const RpcType method_type_; + RpcType method_type_; void* const channel_tag_; }; diff --git a/include/grpc++/impl/codegen/rpc_service_method.h b/include/grpc++/impl/codegen/rpc_service_method.h index 8b1f026c91..52124fba0b 100644 --- a/include/grpc++/impl/codegen/rpc_service_method.h +++ b/include/grpc++/impl/codegen/rpc_service_method.h @@ -82,6 +82,7 @@ class RpcServiceMethod : public RpcMethod { // if MethodHandler is nullptr, then this is an async method MethodHandler* handler() const { return handler_.get(); } void ResetHandler() { handler_.reset(); } + void SetHandler(MethodHandler* handler) { handler_.reset(handler); } private: void* server_tag_; diff --git a/include/grpc++/impl/codegen/server_context.h b/include/grpc++/impl/codegen/server_context.h index 08212af861..379c9f7cf8 100644 --- a/include/grpc++/impl/codegen/server_context.h +++ b/include/grpc++/impl/codegen/server_context.h @@ -65,8 +65,10 @@ template <class R> class ServerReader; template <class W> class ServerWriter; +namespace internal { template <class W, class R> -class ServerReaderWriter; +class ServerReaderWriterBody; +} template <class ServiceType, class RequestType, class ResponseType> class RpcMethodHandler; template <class ServiceType, class RequestType, class ResponseType> @@ -166,6 +168,10 @@ class ServerContext { async_notify_when_done_tag_ = tag; } + // Should be used for framework-level extensions only. + // Applications never need to call this method. + grpc_call* c_call() { return call_; } + private: friend class ::grpc::testing::InteropServerContextInspector; friend class ::grpc::ServerInterface; @@ -183,15 +189,15 @@ class ServerContext { template <class W> friend class ::grpc::ServerWriter; template <class W, class R> - friend class ::grpc::ServerReaderWriter; + friend class ::grpc::internal::ServerReaderWriterBody; template <class ServiceType, class RequestType, class ResponseType> friend class RpcMethodHandler; template <class ServiceType, class RequestType, class ResponseType> friend class ClientStreamingHandler; template <class ServiceType, class RequestType, class ResponseType> friend class ServerStreamingHandler; - template <class ServiceType, class RequestType, class ResponseType> - friend class BidiStreamingHandler; + template <class Streamer, bool WriteNeeded> + friend class TemplatedBidiStreamingHandler; friend class UnknownMethodHandler; friend class ::grpc::ClientContext; diff --git a/include/grpc++/impl/codegen/service_type.h b/include/grpc++/impl/codegen/service_type.h index c19dfc7d45..72b2225312 100644 --- a/include/grpc++/impl/codegen/service_type.h +++ b/include/grpc++/impl/codegen/service_type.h @@ -147,6 +147,17 @@ class Service { methods_[index].reset(); } + void MarkMethodStreamedUnary(int index, + MethodHandler* streamed_unary_method) { + GPR_CODEGEN_ASSERT(methods_[index] && methods_[index]->handler() && + "Cannot mark an async or generic method Streamed Unary"); + methods_[index]->SetHandler(streamed_unary_method); + + // From the server's point of view, streamed unary is a special + // case of BIDI_STREAMING that has 1 read and 1 write, in that order. + methods_[index]->SetMethodType(::grpc::RpcMethod::BIDI_STREAMING); + } + private: friend class Server; friend class ServerInterface; diff --git a/include/grpc++/impl/codegen/sync_stream.h b/include/grpc++/impl/codegen/sync_stream.h index b2b972760d..e1d4660ae7 100644 --- a/include/grpc++/impl/codegen/sync_stream.h +++ b/include/grpc++/impl/codegen/sync_stream.h @@ -64,12 +64,24 @@ class ClientStreamingInterface { virtual Status Finish() = 0; }; +/// Common interface for all synchronous server side streaming. +class ServerStreamingInterface { + public: + virtual ~ServerStreamingInterface() {} + + /// Blocking send initial metadata to client. + virtual void SendInitialMetadata() = 0; +}; + /// An interface that yields a sequence of messages of type \a R. template <class R> class ReaderInterface { public: virtual ~ReaderInterface() {} + /// Upper bound on the next message size available for reading on this stream + virtual bool NextMessageSize(uint32_t* sz) = 0; + /// Blocking read a message and parse to \a msg. Returns \a true on success. /// This is thread-safe with respect to \a Write or \WritesDone methods on /// the same stream. It should not be called concurrently with another \a @@ -148,6 +160,11 @@ class ClientReader GRPC_FINAL : public ClientReaderInterface<R> { cq_.Pluck(&ops); /// status ignored } + bool NextMessageSize(uint32_t* sz) GRPC_OVERRIDE { + *sz = call_.max_message_size(); + return true; + } + bool Read(R* msg) GRPC_OVERRIDE { CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>> ops; if (!context_->initial_metadata_received_) { @@ -293,6 +310,11 @@ class ClientReaderWriter GRPC_FINAL : public ClientReaderWriterInterface<W, R> { cq_.Pluck(&ops); // status ignored } + bool NextMessageSize(uint32_t* sz) GRPC_OVERRIDE { + *sz = call_.max_message_size(); + return true; + } + bool Read(R* msg) GRPC_OVERRIDE { CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>> ops; if (!context_->initial_metadata_received_) { @@ -336,12 +358,17 @@ class ClientReaderWriter GRPC_FINAL : public ClientReaderWriterInterface<W, R> { Call call_; }; +/// Server-side interface for streaming reads of message of type \a R. +template <class R> +class ServerReaderInterface : public ServerStreamingInterface, + public ReaderInterface<R> {}; + template <class R> -class ServerReader GRPC_FINAL : public ReaderInterface<R> { +class ServerReader GRPC_FINAL : public ServerReaderInterface<R> { public: ServerReader(Call* call, ServerContext* ctx) : call_(call), ctx_(ctx) {} - void SendInitialMetadata() { + void SendInitialMetadata() GRPC_OVERRIDE { GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_); CallOpSet<CallOpSendInitialMetadata> ops; @@ -355,6 +382,11 @@ class ServerReader GRPC_FINAL : public ReaderInterface<R> { call_->cq()->Pluck(&ops); } + bool NextMessageSize(uint32_t* sz) GRPC_OVERRIDE { + *sz = call_->max_message_size(); + return true; + } + bool Read(R* msg) GRPC_OVERRIDE { CallOpSet<CallOpRecvMessage<R>> ops; ops.RecvMessage(msg); @@ -367,12 +399,17 @@ class ServerReader GRPC_FINAL : public ReaderInterface<R> { ServerContext* const ctx_; }; +/// Server-side interface for streaming writes of message of type \a W. template <class W> -class ServerWriter GRPC_FINAL : public WriterInterface<W> { +class ServerWriterInterface : public ServerStreamingInterface, + public WriterInterface<W> {}; + +template <class W> +class ServerWriter GRPC_FINAL : public ServerWriterInterface<W> { public: ServerWriter(Call* call, ServerContext* ctx) : call_(call), ctx_(ctx) {} - void SendInitialMetadata() { + void SendInitialMetadata() GRPC_OVERRIDE { GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_); CallOpSet<CallOpSendInitialMetadata> ops; @@ -411,10 +448,17 @@ class ServerWriter GRPC_FINAL : public WriterInterface<W> { /// Server-side interface for bi-directional streaming. template <class W, class R> -class ServerReaderWriter GRPC_FINAL : public WriterInterface<W>, - public ReaderInterface<R> { +class ServerReaderWriterInterface : public ServerStreamingInterface, + public WriterInterface<W>, + public ReaderInterface<R> {}; + +// Actual implementation of bi-directional streaming +namespace internal { +template <class W, class R> +class ServerReaderWriterBody GRPC_FINAL { public: - ServerReaderWriter(Call* call, ServerContext* ctx) : call_(call), ctx_(ctx) {} + ServerReaderWriterBody(Call* call, ServerContext* ctx) + : call_(call), ctx_(ctx) {} void SendInitialMetadata() { GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_); @@ -430,15 +474,19 @@ class ServerReaderWriter GRPC_FINAL : public WriterInterface<W>, call_->cq()->Pluck(&ops); } - bool Read(R* msg) GRPC_OVERRIDE { + bool NextMessageSize(uint32_t* sz) { + *sz = call_->max_message_size(); + return true; + } + + bool Read(R* msg) { CallOpSet<CallOpRecvMessage<R>> ops; ops.RecvMessage(msg); call_->PerformOps(&ops); return call_->cq()->Pluck(&ops) && ops.got_message; } - using WriterInterface<W>::Write; - bool Write(const W& msg, const WriteOptions& options) GRPC_OVERRIDE { + bool Write(const W& msg, const WriteOptions& options) { CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage> ops; if (!ops.SendMessage(msg, options).ok()) { return false; @@ -459,6 +507,76 @@ class ServerReaderWriter GRPC_FINAL : public WriterInterface<W>, Call* const call_; ServerContext* const ctx_; }; +} + +// class to represent the user API for a bidirectional streaming call +template <class W, class R> +class ServerReaderWriter GRPC_FINAL : public ServerReaderWriterInterface<W, R> { + public: + ServerReaderWriter(Call* call, ServerContext* ctx) : body_(call, ctx) {} + + void SendInitialMetadata() GRPC_OVERRIDE { body_.SendInitialMetadata(); } + + bool NextMessageSize(uint32_t* sz) GRPC_OVERRIDE { + return body_.NextMessageSize(sz); + } + + bool Read(R* msg) GRPC_OVERRIDE { return body_.Read(msg); } + + using WriterInterface<W>::Write; + bool Write(const W& msg, const WriteOptions& options) GRPC_OVERRIDE { + return body_.Write(msg, options); + } + + private: + internal::ServerReaderWriterBody<W, R> body_; +}; + +/// A class to represent a flow-controlled unary call. This is something +/// of a hybrid between conventional unary and streaming. This is invoked +/// through a unary call on the client side, but the server responds to it +/// as though it were a single-ping-pong streaming call. The server can use +/// the \a NextMessageSize method to determine an upper-bound on the size of +/// the message. +/// A key difference relative to streaming: ServerUnaryStreamer +/// must have exactly 1 Read and exactly 1 Write, in that order, to function +/// correctly. Otherwise, the RPC is in error. +template <class RequestType, class ResponseType> +class ServerUnaryStreamer GRPC_FINAL + : public ServerReaderWriterInterface<ResponseType, RequestType> { + public: + ServerUnaryStreamer(Call* call, ServerContext* ctx) + : body_(call, ctx), read_done_(false), write_done_(false) {} + + void SendInitialMetadata() GRPC_OVERRIDE { body_.SendInitialMetadata(); } + + bool NextMessageSize(uint32_t* sz) GRPC_OVERRIDE { + return body_.NextMessageSize(sz); + } + + bool Read(RequestType* request) GRPC_OVERRIDE { + if (read_done_) { + return false; + } + read_done_ = true; + return body_.Read(request); + } + + using WriterInterface<ResponseType>::Write; + bool Write(const ResponseType& response, + const WriteOptions& options) GRPC_OVERRIDE { + if (write_done_ || !read_done_) { + return false; + } + write_done_ = true; + return body_.Write(response, options); + } + + private: + internal::ServerReaderWriterBody<ResponseType, RequestType> body_; + bool read_done_; + bool write_done_; +}; } // namespace grpc diff --git a/include/grpc++/impl/server_builder_plugin.h b/include/grpc++/impl/server_builder_plugin.h index 1e157efa11..61632e32fa 100644 --- a/include/grpc++/impl/server_builder_plugin.h +++ b/include/grpc++/impl/server_builder_plugin.h @@ -41,6 +41,7 @@ namespace grpc { class ServerInitializer; +class ChannelArguments; class ServerBuilderPlugin { public: @@ -58,6 +59,10 @@ class ServerBuilderPlugin { // ServerBuilderOption::UpdatePlugins virtual void ChangeArguments(const grpc::string& name, void* value) = 0; + // UpdateChannelArguments will be called in ServerBuilder::BuildAndStart(), + // before the Server instance is created. + virtual void UpdateChannelArguments(ChannelArguments* args) {} + virtual bool has_sync_methods() const { return false; } virtual bool has_async_methods() const { return false; } }; |