aboutsummaryrefslogtreecommitdiffhomepage
path: root/include/grpc++/impl
diff options
context:
space:
mode:
Diffstat (limited to 'include/grpc++/impl')
-rw-r--r--include/grpc++/impl/codegen/call.h6
-rw-r--r--include/grpc++/impl/codegen/completion_queue.h8
-rw-r--r--include/grpc++/impl/codegen/fc_unary.h90
-rw-r--r--include/grpc++/impl/codegen/method_handler_impl.h36
-rw-r--r--include/grpc++/impl/codegen/rpc_method.h3
-rw-r--r--include/grpc++/impl/codegen/rpc_service_method.h1
-rw-r--r--include/grpc++/impl/codegen/server_context.h8
-rw-r--r--include/grpc++/impl/codegen/service_type.h10
-rw-r--r--include/grpc++/impl/codegen/sync_stream.h44
9 files changed, 191 insertions, 15 deletions
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/completion_queue.h b/include/grpc++/impl/codegen/completion_queue.h
index 03009e0561..95ece77ce5 100644
--- a/include/grpc++/impl/codegen/completion_queue.h
+++ b/include/grpc++/impl/codegen/completion_queue.h
@@ -69,7 +69,7 @@ class ServerReader;
template <class W>
class ServerWriter;
template <class W, class R>
-class ServerReaderWriter;
+class ServerReaderWriterInterface;
template <class ServiceType, class RequestType, class ResponseType>
class RpcMethodHandler;
template <class ServiceType, class RequestType, class ResponseType>
@@ -78,6 +78,8 @@ template <class ServiceType, class RequestType, class ResponseType>
class ServerStreamingHandler;
template <class ServiceType, class RequestType, class ResponseType>
class BidiStreamingHandler;
+template <class ServiceType, class RequestType, class ResponseType>
+class FCUnaryMethodHandler;
class UnknownMethodHandler;
class Channel;
@@ -178,7 +180,7 @@ class CompletionQueue : private GrpcLibraryCodegen {
template <class W>
friend class ::grpc::ServerWriter;
template <class W, class R>
- friend class ::grpc::ServerReaderWriter;
+ friend class ::grpc::ServerReaderWriterInterface;
template <class ServiceType, class RequestType, class ResponseType>
friend class RpcMethodHandler;
template <class ServiceType, class RequestType, class ResponseType>
@@ -187,6 +189,8 @@ class CompletionQueue : private GrpcLibraryCodegen {
friend class ServerStreamingHandler;
template <class ServiceType, class RequestType, class ResponseType>
friend class BidiStreamingHandler;
+ template <class ServiceType, class RequestType, class ResponseType>
+ friend class FCUnaryMethodHandler;
friend class UnknownMethodHandler;
friend class ::grpc::Server;
friend class ::grpc::ServerContext;
diff --git a/include/grpc++/impl/codegen/fc_unary.h b/include/grpc++/impl/codegen/fc_unary.h
new file mode 100644
index 0000000000..a423f1f0b3
--- /dev/null
+++ b/include/grpc++/impl/codegen/fc_unary.h
@@ -0,0 +1,90 @@
+/*
+ *
+ * Copyright 2016, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ * * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#ifndef GRPCXX_IMPL_CODEGEN_FC_UNARY_H
+#define GRPCXX_IMPL_CODEGEN_FC_UNARY_H
+
+#include <grpc++/impl/codegen/call.h>
+#include <grpc++/impl/codegen/completion_queue.h>
+#include <grpc++/impl/codegen/core_codegen_interface.h>
+#include <grpc++/impl/codegen/server_context.h>
+#include <grpc++/impl/codegen/sync_stream.h>
+
+namespace grpc {
+/// 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: an FCUnary 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 FCUnary GRPC_FINAL
+ : public ServerReaderWriterInterface<ResponseType, RequestType> {
+ public:
+ FCUnary(Call* call, ServerContext* ctx)
+ : ServerReaderWriterInterface<ResponseType, RequestType>(call, ctx),
+ read_done_(false),
+ write_done_(false) {}
+
+ ~FCUnary() {}
+
+ bool Read(RequestType* request) GRPC_OVERRIDE {
+ if (read_done_) {
+ return false;
+ }
+ read_done_ = true;
+ return ServerReaderWriterInterface<ResponseType, RequestType>::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 ServerReaderWriterInterface<ResponseType, RequestType>::Write(
+ response, options);
+ }
+
+ private:
+ bool read_done_;
+ bool write_done_;
+};
+} // namespace grpc
+
+#endif // GRPCXX_IMPL_CODEGEN_FC_UNARY_H
diff --git a/include/grpc++/impl/codegen/method_handler_impl.h b/include/grpc++/impl/codegen/method_handler_impl.h
index 2f4be644ba..9c3af53b3a 100644
--- a/include/grpc++/impl/codegen/method_handler_impl.h
+++ b/include/grpc++/impl/codegen/method_handler_impl.h
@@ -35,6 +35,7 @@
#define GRPCXX_IMPL_CODEGEN_METHOD_HANDLER_IMPL_H
#include <grpc++/impl/codegen/core_codegen_interface.h>
+#include <grpc++/impl/codegen/fc_unary.h>
#include <grpc++/impl/codegen/rpc_service_method.h>
#include <grpc++/impl/codegen/sync_stream.h>
@@ -202,6 +203,41 @@ class BidiStreamingHandler : public MethodHandler {
ServiceType* service_;
};
+// A wrapper class of an application provided rpc method handler
+// specifically to apply to the flow-controlled implementation of a unary
+// method.
+/// The argument to the constructor should be a member function already
+/// bound to the appropriate service instance. The declaration gets too
+/// complicated
+/// otherwise.
+template <class ServiceType, class RequestType, class ResponseType>
+class FCUnaryMethodHandler : public MethodHandler {
+ public:
+ FCUnaryMethodHandler(
+ std::function<Status(ServerContext*, FCUnary<RequestType, ResponseType>*)>
+ func)
+ : func_(func) {}
+
+ void RunHandler(const HandlerParameter& param) GRPC_FINAL {
+ FCUnary<RequestType, ResponseType> fc_unary(param.call,
+ param.server_context);
+ Status status = func_(param.server_context, &fc_unary);
+ if (!param.server_context->sent_initial_metadata_) {
+ // means that the write never happened, which is bad
+ } else {
+ CallOpSet<CallOpServerSendStatus> ops;
+ ops.ServerSendStatus(param.server_context->trailing_metadata_, status);
+ param.call->PerformOps(&ops);
+ param.call->cq()->Pluck(&ops);
+ }
+ }
+
+ private:
+ // Application provided rpc handler function, already bound to its service.
+ std::function<Status(ServerContext*, FCUnary<RequestType, ResponseType>*)>
+ func_;
+};
+
// Handle unknown method by returning UNIMPLEMENTED error.
class UnknownMethodHandler : public MethodHandler {
public:
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..aadae893ad 100644
--- a/include/grpc++/impl/codegen/server_context.h
+++ b/include/grpc++/impl/codegen/server_context.h
@@ -66,7 +66,7 @@ class ServerReader;
template <class W>
class ServerWriter;
template <class W, class R>
-class ServerReaderWriter;
+class ServerReaderWriterInterface;
template <class ServiceType, class RequestType, class ResponseType>
class RpcMethodHandler;
template <class ServiceType, class RequestType, class ResponseType>
@@ -75,6 +75,8 @@ template <class ServiceType, class RequestType, class ResponseType>
class ServerStreamingHandler;
template <class ServiceType, class RequestType, class ResponseType>
class BidiStreamingHandler;
+template <class ServiceType, class RequestType, class ResponseType>
+class FCUnaryMethodHandler;
class UnknownMethodHandler;
class Call;
@@ -183,7 +185,7 @@ class ServerContext {
template <class W>
friend class ::grpc::ServerWriter;
template <class W, class R>
- friend class ::grpc::ServerReaderWriter;
+ friend class ::grpc::ServerReaderWriterInterface;
template <class ServiceType, class RequestType, class ResponseType>
friend class RpcMethodHandler;
template <class ServiceType, class RequestType, class ResponseType>
@@ -192,6 +194,8 @@ class ServerContext {
friend class ServerStreamingHandler;
template <class ServiceType, class RequestType, class ResponseType>
friend class BidiStreamingHandler;
+ template <class ServiceType, class RequestType, class ResponseType>
+ friend class FCUnaryMethodHandler;
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..4af40422a1 100644
--- a/include/grpc++/impl/codegen/service_type.h
+++ b/include/grpc++/impl/codegen/service_type.h
@@ -147,6 +147,16 @@ class Service {
methods_[index].reset();
}
+ void MarkMethodFCUnary(int index, MethodHandler* fc_unary_method) {
+ GPR_CODEGEN_ASSERT(methods_[index] && methods_[index]->handler() &&
+ "Cannot mark an async or generic method as FCUnary");
+ methods_[index]->SetHandler(fc_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..d9b7e6fec5 100644
--- a/include/grpc++/impl/codegen/sync_stream.h
+++ b/include/grpc++/impl/codegen/sync_stream.h
@@ -70,6 +70,9 @@ 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 +151,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 +301,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_) {
@@ -355,6 +368,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);
@@ -411,12 +429,12 @@ 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 WriterInterface<W>,
+ public ReaderInterface<R> {
public:
- ServerReaderWriter(Call* call, ServerContext* ctx) : call_(call), ctx_(ctx) {}
-
- void SendInitialMetadata() {
+ ServerReaderWriterInterface(Call* call, ServerContext* ctx)
+ : call_(call), ctx_(ctx) {}
+ virtual void SendInitialMetadata() {
GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
CallOpSet<CallOpSendInitialMetadata> ops;
@@ -430,7 +448,12 @@ class ServerReaderWriter GRPC_FINAL : public WriterInterface<W>,
call_->cq()->Pluck(&ops);
}
- bool Read(R* msg) GRPC_OVERRIDE {
+ virtual bool NextMessageSize(uint32_t* sz) GRPC_OVERRIDE {
+ *sz = call_->max_message_size();
+ return true;
+ }
+
+ virtual bool Read(R* msg) GRPC_OVERRIDE {
CallOpSet<CallOpRecvMessage<R>> ops;
ops.RecvMessage(msg);
call_->PerformOps(&ops);
@@ -438,7 +461,7 @@ class ServerReaderWriter GRPC_FINAL : public WriterInterface<W>,
}
using WriterInterface<W>::Write;
- bool Write(const W& msg, const WriteOptions& options) GRPC_OVERRIDE {
+ virtual bool Write(const W& msg, const WriteOptions& options) GRPC_OVERRIDE {
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage> ops;
if (!ops.SendMessage(msg, options).ok()) {
return false;
@@ -460,6 +483,13 @@ class ServerReaderWriter GRPC_FINAL : public WriterInterface<W>,
ServerContext* const ctx_;
};
+template <class W, class R>
+class ServerReaderWriter GRPC_FINAL : public ServerReaderWriterInterface<W, R> {
+ public:
+ ServerReaderWriter(Call* call, ServerContext* ctx)
+ : ServerReaderWriterInterface<W, R>(call, ctx) {}
+};
+
} // namespace grpc
#endif // GRPCXX_IMPL_CODEGEN_SYNC_STREAM_H
generated by cgit v1.2.3 (git 2.39.1) at 2025-01-02 13:25:40 +0000