fix: fix the rebase caused the error

fix rebase error.

Signed-off-by: jokemanfire <hu.dingyang@zte.com.cn>

Author: jokemanfire

.github/workflows/bvt.yml

@@ -15,11 +15,7 @@ jobs:
         shell: bash
       - name: Check
         run: |
-          make deps
-          make check
-          make -C compiler check
-          make -C ttrpc-codegen check
-          make -C codegen check
+          make check-all
 
   make:
     name: Build

Cargo.toml

@@ -39,7 +39,7 @@ prost-build = { version = "0.11", optional = true }
 
 [features]
 default = ["sync","rustprotobuf"]
-async = ["dep:async-trait", "dep:tokio", "dep:futures", "dep:tokio-vsock"]
+async = ["async-trait", "async-stream", "tokio", "futures", "tokio-vsock"]
 sync = []
 prost = ["dep:prost", "dep:prost-build"]
 rustprotobuf = ["dep:protobuf"]

src/proto.rs

@@ -241,23 +241,9 @@ impl GenMessage {
             .await
             .map_err(|e| Error::Socket(e.to_string()))?;
 
-        if header.length > MESSAGE_LENGTH_MAX as u32 {
-            #[cfg(not(feature = "prost"))]
-            return Err(get_rpc_status(
-                Code::INVALID_ARGUMENT,
-                format!(
-                    "message length {} exceed maximum message size of {}",
-                    header.length, MESSAGE_LENGTH_MAX
-                ),
-            ));
-            #[cfg(feature = "prost")]
-            return Err(get_rpc_status(
-                Code::InvalidArgument,
-                format!(
-                    "message length {} exceed maximum message size of {}",
-                    header.length, MESSAGE_LENGTH_MAX
-                ),
-            ));
+        if let Err(e) = check_oversize(header.length as usize, true) {
+            discard_message_body(reader, &header).await?;
+            return Err(GenMessageError::ReturnError(header, e));
         }
 
         let mut content = vec![0; header.length as usize];
@@ -407,23 +393,12 @@ where
             .await
             .map_err(|e| Error::Socket(e.to_string()))?;
 
-        if header.length > MESSAGE_LENGTH_MAX as u32 {
-            #[cfg(not(feature = "prost"))]
-            return Err(get_rpc_status(
-                Code::INVALID_ARGUMENT,
-                format!(
-                    "message length {} exceed maximum message size of {}",
-                    header.length, MESSAGE_LENGTH_MAX
-                ),
-            ));
-            #[cfg(feature = "prost")]
-            return Err(get_rpc_status(
-                Code::InvalidArgument,
-                format!(
-                    "message length {} exceed maximum message size of {}",
-                    header.length, MESSAGE_LENGTH_MAX
-                ),
-            ));
+        if check_oversize(header.length as usize, true).is_err() {
+            discard_message_body(reader, &header).await?;
+            return Ok(Self {
+                header,
+                payload: C::decode("").map_err(err_to_others_err!(e, "Decode payload failed."))?,
+            });
         }
 
         let mut content = vec![0; header.length as usize];

src/sync/channel.rs

@@ -97,17 +97,10 @@ pub fn read_message(conn: &PipeConnection) -> Result<(MessageHeader, Result<Vec<
     let mh = read_message_header(conn)?;
     trace!("Got Message header {:?}", mh);
 
-    if mh.length > MESSAGE_LENGTH_MAX as u32 {
-        return Err(get_rpc_status(
-            #[cfg(not(feature = "prost"))]
-            Code::INVALID_ARGUMENT,
-            #[cfg(feature = "prost")]
-            Code::InvalidArgument,
-            format!(
-                "message length {} exceed maximum message size of {}",
-                mh.length, MESSAGE_LENGTH_MAX
-            ),
-        ));
+    let mh_len = mh.length as usize;
+    if let Err(e) = check_oversize(mh_len, true) {
+        discard_count(conn, mh_len)?;
+        return Ok((mh, Err(e)));
     }
 
     let buf = read_count(conn, mh.length as usize)?;

src/sync/server.rs

@@ -173,20 +173,13 @@ fn start_method_handler_thread(
             {
                 let mut s = CodedInputStream::from_bytes(&buf);
                 req = Request::new();
-                req = Request::new();
                 if let Err(x) = req.merge_from(&mut s) {
                     let status = get_status(Code::INVALID_ARGUMENT, x.to_string());
                     let mut res = Response::new();
                     res.set_status(status);
                     if let Err(x) = response_to_channel(mh.stream_id, res, res_tx.clone()) {
                         debug!("response_to_channel get error {:?}", x);
-                        quit.store(true, Ordering::SeqCst);
-                        // the client connection would be closed and
-                        // the connection dealing main thread would have
-                        // exited.
-                        control_tx
-                            .send(())
-                            .unwrap_or_else(|err| trace!("Failed to send {:?}", err));
+                        quit_connection(quit, control_tx);
                         break;
                     }
                     continue;

src/sync/utils.rs

@@ -61,20 +61,10 @@ pub fn response_to_channel(
 
 pub fn response_error_to_channel(
     stream_id: u32,
-    res: Response,
+    e: Error,
     tx: std::sync::mpsc::Sender<(MessageHeader, Vec<u8>)>,
 ) -> Result<()> {
-    let buf = res.encode_to_vec();
-
-    let mh = MessageHeader {
-        length: buf.len() as u32,
-        stream_id,
-        type_: MESSAGE_TYPE_RESPONSE,
-        flags: 0,
-    };
-    tx.send((mh, buf)).map_err(err_to_others_err!(e, ""))?;
-
-    Ok(())
+    response_to_channel(stream_id, e.into(), tx)
 }
 
 /// Handle request in sync mode.

tests/run-examples.rs

@@ -90,9 +90,7 @@ fn run_examples() -> Result<(), Box<dyn std::error::Error>> {
     {
         println!("Running examples with rustprotobuf feature");
         run_example("server", "client", "example")?;
-        #[cfg(unix)]
         run_example("async-server", "async-client", "example")?;
-        #[cfg(unix)]
         run_example("async-stream-server", "async-stream-client", "example")?;
     }