feat: add decode method in DataProcess trait

examples/consumer.rs

@@ -1,4 +1,4 @@
-use std::str::from_utf8;
+use std::fmt::{Display, Formatter};
 use std::time::Duration;
 
 use shm_ringbuf::consumer::process::{DataProcess, ResultSender};
@@ -23,8 +23,15 @@ async fn main() {
 pub struct StringPrint;
 
 impl DataProcess for StringPrint {
-    async fn process(&self, data: &[u8], result_sender: ResultSender) {
-        if let Err(e) = self.do_process(data).await {
+    type Message = String;
+    type Error = Error;
+
+    fn decode(&self, data: &[u8]) -> Result<Self::Message, Self::Error> {
+        String::from_utf8(data.to_vec()).map_err(|_| Error::DecodeError)
+    }
+
+    async fn process(&self, msg: Self::Message, result_sender: ResultSender) {
+        if let Err(e) = self.do_process(&msg).await {
             result_sender.push_result(e).await;
         } else {
             result_sender.push_ok().await;
@@ -33,8 +40,7 @@ impl DataProcess for StringPrint {
 }
 
 impl StringPrint {
-    async fn do_process(&self, data: &[u8]) -> Result<(), Error> {
-        let msg = from_utf8(data).map_err(|_| Error::DecodeError)?;
+    async fn do_process(&self, msg: &str) -> Result<(), Error> {
         info!("receive: {}", msg);
         Ok(())
     }
@@ -70,3 +76,11 @@ impl From<Error> for DataProcessResult {
         }
     }
 }
+
+impl Display for Error {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.message())
+    }
+}
+
+impl std::error::Error for Error {}

src/consumer.rs

@@ -19,8 +19,8 @@ use tokio::time::sleep;
 use tokio_util::sync::CancellationToken;
 use tracing::warn;
 
-use crate::error::DataProcessResult;
 use crate::error::CHECKSUM_MISMATCH;
+use crate::error::{DataProcessResult, DECODE_ERROR};
 use crate::fd_pass::FdRecvServer;
 use crate::grpc::proto::shm_control_server::ShmControlServer;
 use crate::grpc::server::ShmCtlHandler;
@@ -231,8 +231,22 @@ where
             session: session.clone(),
         };
 
-        processor.process(data_slice, result_sender).await;
+        let result = processor.decode(data_slice);
 
         unsafe { ringbuf.advance_consume_offset(data_block.total_len()) }
+        drop(data_block);
+
+        match result {
+            Ok(message) => {
+                processor.process(message, result_sender).await;
+            }
+            Err(e) => {
+                let result = DataProcessResult {
+                    status_code: DECODE_ERROR,
+                    message: e.to_string(),
+                };
+                result_sender.push_result(result).await;
+            }
+        };
     }
 }

src/consumer/process.rs

@@ -1,3 +1,4 @@
+use std::error::Error;
 use std::fmt::Debug;
 use std::future::Future;
 
@@ -6,9 +7,15 @@ use crate::error::DataProcessResult;
 use super::session_manager::SessionRef;
 
 pub trait DataProcess: Send + Sync {
+    type Message: Debug;
+
+    type Error: Error;
+
+    fn decode(&self, data: &[u8]) -> Result<Self::Message, Self::Error>;
+
     fn process(
         &self,
-        data: &[u8],
+        message: Self::Message,
         result_sender: ResultSender,
     ) -> impl Future<Output = ()>;
 }

src/error.rs

@@ -194,3 +194,4 @@ impl DataProcessResult {
 /// 100000 - 200000 are error codes used internally by shm-ringbuf and should
 /// not be used as business codes.
 pub const CHECKSUM_MISMATCH: u32 = 100000;
+pub const DECODE_ERROR: u32 = 100001;

src/producer/fetch.rs

@@ -78,7 +78,6 @@ impl ResultFetcher {
                 };
 
                 fetcher.inner.normal.store(false, Ordering::Relaxed);
-                fetcher.inner.subscriptions.clear();
                 sleep(fetcher.inner.retry_interval).await;
             }
         });

tests/common.rs

@@ -1,10 +1,10 @@
-use std::{str::from_utf8, sync::Arc, time::Duration};
-
 use shm_ringbuf::{
     consumer::process::{DataProcess, ResultSender},
     error::DataProcessResult,
     producer::{prealloc::PreAlloc, RingbufProducer},
 };
+use std::fmt::{Display, Formatter};
+use std::{sync::Arc, time::Duration};
 use tokio::{sync::mpsc::Sender, time::sleep};
 use tracing::{error, warn};
 
@@ -13,8 +13,15 @@ pub struct MsgForward {
 }
 
 impl DataProcess for MsgForward {
-    async fn process(&self, data: &[u8], result_sender: ResultSender) {
-        if let Err(e) = self.do_process(data).await {
+    type Message = String;
+    type Error = Error;
+
+    fn decode(&self, data: &[u8]) -> Result<Self::Message, Self::Error> {
+        String::from_utf8(data.to_vec()).map_err(|_| Error::DecodeError)
+    }
+
+    async fn process(&self, msg: Self::Message, result_sender: ResultSender) {
+        if let Err(e) = self.do_process(&msg).await {
             result_sender.push_result(e).await;
         } else {
             result_sender.push_ok().await;
@@ -23,9 +30,7 @@ impl DataProcess for MsgForward {
 }
 
 impl MsgForward {
-    async fn do_process(&self, data: &[u8]) -> Result<(), Error> {
-        let msg = from_utf8(data).map_err(|_| Error::DecodeError)?;
-
+    async fn do_process(&self, msg: &str) -> Result<(), Error> {
         let _ = self.sender.send(msg.to_string()).await;
 
         Ok(())
@@ -63,6 +68,14 @@ impl From<Error> for DataProcessResult {
     }
 }
 
+impl Display for Error {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.message())
+    }
+}
+
+impl std::error::Error for Error {}
+
 pub fn msg_num() -> usize {
     std::env::var("MSG_NUM")
         .unwrap_or_else(|_| "10000".to_string())