feat: support detach grpc server (#12)

* feat: support detach grpc server

* expose some structs

src/consumer.rs

@@ -4,6 +4,8 @@ pub mod settings;
 pub(crate) mod session_manager;
 
 use std::fmt::Debug;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -19,13 +21,16 @@ use tracing::warn;
 
 use crate::error::DataProcessResult;
 use crate::fd_pass::FdRecvServer;
+use crate::grpc::proto::shm_control_server::ShmControlServer;
+use crate::grpc::server::ShmCtlHandler;
 use crate::grpc::server::ShmCtlServer;
 
 pub struct RingbufConsumer {
     session_manager: SessionManagerRef,
     notify: Arc<Notify>,
     settings: ConsumerSettings,
     cancel: CancellationToken,
+    detach_grpc: AtomicBool,
 }
 
 impl RingbufConsumer {
@@ -34,16 +39,16 @@ impl RingbufConsumer {
             settings.max_session_capacity,
             settings.session_tti,
         ));
-
         let notify = Arc::new(Notify::new());
-
         let cancel = CancellationToken::new();
+        let detach_grpc = AtomicBool::new(false);
 
         RingbufConsumer {
             session_manager,
             notify,
             cancel,
             settings,
+            detach_grpc,
         }
     }
 
@@ -52,7 +57,9 @@ impl RingbufConsumer {
         P: DataProcess<Error = E>,
         E: Into<DataProcessResult> + Debug + Send,
     {
-        self.start_grpc_server().await;
+        if !self.detach_grpc.load(Ordering::Relaxed) {
+            self.start_grpc_server().await;
+        }
 
         self.start_fdrecv_server().await;
 
@@ -61,6 +68,15 @@ impl RingbufConsumer {
         self.process_loop(&processor, interval, Some(cancel)).await;
     }
 
+    pub fn detach_grpc_server(&self) -> ShmControlServer<ShmCtlHandler> {
+        let handler = ShmCtlHandler {
+            notify: self.notify.clone(),
+            session_manager: self.session_manager.clone(),
+        };
+        self.detach_grpc.store(true, Ordering::Relaxed);
+        ShmControlServer::new(handler)
+    }
+
     pub fn cancel(&self) {
         self.cancel.cancel();
     }

src/grpc/server.rs

@@ -93,9 +93,9 @@ where
     }
 }
 
-struct ShmCtlHandler {
-    notify: Arc<Notify>,
-    session_manager: SessionManagerRef,
+pub struct ShmCtlHandler {
+    pub(crate) notify: Arc<Notify>,
+    pub(crate) session_manager: SessionManagerRef,
 }
 
 #[async_trait::async_trait]

src/lib.rs

@@ -8,5 +8,7 @@ mod macros;
 mod memfd;
 mod ringbuf;
 
+pub use grpc::proto::shm_control_server::ShmControlServer;
+pub use grpc::server::ShmCtlHandler;
 #[cfg(feature = "benchmark")]
 pub use ringbuf::Ringbuf;