Merge pull request #9759 from yang-g/max_recv_msg_size

Restore the NextMessageSize for streamed calls

include/grpc++/impl/codegen/call.h

@@ -618,7 +618,17 @@ class Call final {
  public:
   /* call is owned by the caller */
   Call(grpc_call* call, CallHook* call_hook, CompletionQueue* cq)
-      : call_hook_(call_hook), cq_(cq), call_(call) {}
+      : call_hook_(call_hook),
+        cq_(cq),
+        call_(call),
+        max_receive_message_size_(-1) {}
+
+  Call(grpc_call* call, CallHook* call_hook, CompletionQueue* cq,
+       int max_receive_message_size)
+      : call_hook_(call_hook),
+        cq_(cq),
+        call_(call),
+        max_receive_message_size_(max_receive_message_size) {}
 
   void PerformOps(CallOpSetInterface* ops) {
     call_hook_->PerformOpsOnCall(ops, this);
@@ -627,10 +637,13 @@ class Call final {
   grpc_call* call() const { return call_; }
   CompletionQueue* cq() const { return cq_; }
 
+  int max_receive_message_size() const { return max_receive_message_size_; }
+
  private:
   CallHook* call_hook_;
   CompletionQueue* cq_;
   grpc_call* call_;
+  int max_receive_message_size_;
 };
 
 }  // namespace grpc

include/grpc++/impl/codegen/sync_stream.h

@@ -160,7 +160,7 @@ class ClientReader final : public ClientReaderInterface<R> {
   }
 
   bool NextMessageSize(uint32_t* sz) override {
-    *sz = INT_MAX;
+    *sz = call_.max_receive_message_size();
     return true;
   }
 
@@ -310,7 +310,7 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
   }
 
   bool NextMessageSize(uint32_t* sz) override {
-    *sz = INT_MAX;
+    *sz = call_.max_receive_message_size();
     return true;
   }
 
@@ -382,7 +382,7 @@ class ServerReader final : public ServerReaderInterface<R> {
   }
 
   bool NextMessageSize(uint32_t* sz) override {
-    *sz = INT_MAX;
+    *sz = call_->max_receive_message_size();
     return true;
   }
 
@@ -474,7 +474,7 @@ class ServerReaderWriterBody final {
   }
 
   bool NextMessageSize(uint32_t* sz) {
-    *sz = INT_MAX;
+    *sz = call_->max_receive_message_size();
     return true;
   }
 

src/cpp/server/server_cc.cc

@@ -186,7 +186,7 @@ class Server::SyncRequest final : public CompletionQueueTag {
    public:
     explicit CallData(Server* server, SyncRequest* mrd)
         : cq_(mrd->cq_),
-          call_(mrd->call_, server, &cq_),
+          call_(mrd->call_, server, &cq_, server->max_receive_message_size()),
           ctx_(mrd->deadline_, mrd->request_metadata_.metadata,
                mrd->request_metadata_.count),
           has_request_payload_(mrd->has_request_payload_),
@@ -590,7 +590,7 @@ bool ServerInterface::BaseAsyncRequest::FinalizeResult(void** tag,
   }
   context_->set_call(call_);
   context_->cq_ = call_cq_;
-  Call call(call_, server_, call_cq_);
+  Call call(call_, server_, call_cq_, server_->max_receive_message_size());
   if (*status && call_) {
     context_->BeginCompletionOp(&call);
   }