Merge pull request #7018 from vjpai/fc_unary

Support server-side streaming of a unary RPC

9 files changed, 181 insertions, 33 deletions

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++/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..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 c9d1f4d69e..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>
@@ -187,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 7601ceae92..e1d4660ae7 100644
--- a/include/grpc++/impl/codegen/sync_stream.h
+++ b/include/grpc++/impl/codegen/sync_stream.h
@@ -79,6 +79,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
@@ -157,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_) {
@@ -302,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_) {
@@ -369,6 +382,11 @@ class ServerReader GRPC_FINAL : public ServerReaderInterface<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);
@@ -434,12 +452,15 @@ class ServerReaderWriterInterface : public ServerStreamingInterface,
                                     public WriterInterface<W>,
                                     public ReaderInterface<R> {};
 
+// Actual implementation of bi-directional streaming
+namespace internal {
 template <class W, class R>
-class ServerReaderWriter GRPC_FINAL : public ServerReaderWriterInterface<W, 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() GRPC_OVERRIDE {
+  void SendInitialMetadata() {
     GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
 
     CallOpSet<CallOpSendInitialMetadata> ops;
@@ -453,15 +474,19 @@ class ServerReaderWriter GRPC_FINAL : public ServerReaderWriterInterface<W, R> {
     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;
@@ -482,6 +507,76 @@ class ServerReaderWriter GRPC_FINAL : public ServerReaderWriterInterface<W, R> {
   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