stream: Fix default data won't be received issue

Protobuf encodes default values, e.g., 0 for number, "" for string,
as empty payload and we wouldn't send empty payload to a receiver.

This commit change the method of empty message detecting.

Fixes #169
Fixes #207

Signed-off-by: Tim Zhang <[email protected]>

Author: Tim-Zhang

src/asynchronous/client.rs

@@ -15,10 +15,10 @@ use nix::unistd::close;
 use tokio::{self, sync::mpsc, task};
 
 use crate::common::client_connect;
-use crate::error::{Error, Result};
+use crate::error::{get_rpc_status, Error, Result};
 use crate::proto::{
-    Code, Codec, GenMessage, Message, MessageHeader, Request, Response, FLAG_REMOTE_CLOSED,
-    FLAG_REMOTE_OPEN, MESSAGE_TYPE_DATA, MESSAGE_TYPE_RESPONSE,
+    Code, Codec, GenMessage, Message, MessageHeader, Request, Response, FLAG_NO_DATA,
+    FLAG_REMOTE_CLOSED, FLAG_REMOTE_OPEN, MESSAGE_TYPE_DATA, MESSAGE_TYPE_RESPONSE,
 };
 use crate::r#async::connection::*;
 use crate::r#async::shutdown;
@@ -117,13 +117,20 @@ impl Client {
         streaming_server: bool,
     ) -> Result<StreamInner> {
         let stream_id = self.next_stream_id.fetch_add(2, Ordering::Relaxed);
+        let is_req_payload_empty = req.payload.is_empty();
 
         let mut msg: GenMessage = Message::new_request(stream_id, req)?
             .try_into()
             .map_err(|e: protobuf::Error| Error::Others(e.to_string()))?;
 
         if streaming_client {
-            msg.header.add_flags(FLAG_REMOTE_OPEN);
+            if !is_req_payload_empty {
+                return Err(get_rpc_status(
+                    Code::INVALID_ARGUMENT,
+                    "Creating a ClientStream and sending payload at the same time is not allowed",
+                ));
+            }
+            msg.header.add_flags(FLAG_REMOTE_OPEN | FLAG_NO_DATA);
         } else {
             msg.header.add_flags(FLAG_REMOTE_CLOSED);
         }

src/asynchronous/server.rs

@@ -37,7 +37,7 @@ use crate::context;
 use crate::error::{get_status, Error, Result};
 use crate::proto::{
     check_oversize, Code, Codec, GenMessage, Message, MessageHeader, Request, Response, Status,
-    FLAG_NO_DATA, FLAG_REMOTE_CLOSED, FLAG_REMOTE_OPEN, MESSAGE_TYPE_DATA, MESSAGE_TYPE_REQUEST,
+    FLAG_NO_DATA, FLAG_REMOTE_CLOSED, MESSAGE_TYPE_DATA, MESSAGE_TYPE_REQUEST,
 };
 use crate::r#async::connection::*;
 use crate::r#async::shutdown;
@@ -606,8 +606,8 @@ impl HandlerContext {
         let stream_tx = tx.clone();
         self.streams.lock().unwrap().insert(stream_id, tx);
 
-        let _remote_close = (req_msg.header.flags & FLAG_REMOTE_CLOSED) == FLAG_REMOTE_CLOSED;
-        let _remote_open = (req_msg.header.flags & FLAG_REMOTE_OPEN) == FLAG_REMOTE_OPEN;
+        let no_data = (req_msg.header.flags & FLAG_NO_DATA) == FLAG_NO_DATA;
+
         let si = StreamInner::new(
             stream_id,
             self.tx.clone(),
@@ -627,7 +627,7 @@ impl HandlerContext {
 
         let task = spawn(async move { stream.handler(ctx, si).await });
 
-        if !req.payload.is_empty() {
+        if !no_data {
             // Fake the first data message.
             let msg = GenMessage {
                 header: MessageHeader::new_data(stream_id, req.payload.len() as u32),