refactor: extract with_writer on FuseSession for notify api

Author: xujihui1985

src/api/server/mod.rs

@@ -51,7 +51,6 @@ pub const MAX_REQ_PAGES: u16 = 256; // 1MB
 pub struct Server<F: FileSystem + Sync> {
     fs: F,
     vers: ArcSwap<ServerVersion>,
-    enabled: Option<FsOptions>,
 }
 
 impl<F: FileSystem + Sync> Server<F> {
@@ -63,7 +62,6 @@ impl<F: FileSystem + Sync> Server<F> {
                 major: KERNEL_VERSION,
                 minor: KERNEL_MINOR_VERSION,
             })),
-            enabled: None,
         }
     }
 }

src/api/server/sync_io.rs

@@ -63,19 +63,15 @@ impl<F: FileSystem + Sync> Server<F> {
         &self, 
         mut w: FuseDevWriter<'_, S>,
     ) -> Result<()> {
-        if !self.enabled.map_or(false,|enabled| {
-            enabled.contains(FsOptions::HAS_RESEND)
-        }) {
-            return Ok(0)
-        }
         let mut buffer_writer = w.split_at(0).map_err(Error::FailedToSplitWriter)?;
         let mut header = OutHeader::default();
         header.unique = 0;
         header.error = NotifyOpcode::Resend as i32;
         buffer_writer
             .write_obj(header)
             .map_err(Error::FailedToWrite)?;
-        buffer_writer.commit(None).map_err(Error::InvalidMessage)
+        buffer_writer.commit(None).map_err(Error::InvalidMessage)?;
+        Ok(())
     }
 
     /// Main entrance to handle requests from the transport layer.
@@ -765,7 +761,6 @@ impl<F: FileSystem + Sync> Server<F> {
                 }
                 let vers = ServerVersion { major, minor };
                 self.vers.store(Arc::new(vers));
-                self.enabled = Some(enabled);
                 if minor < KERNEL_MINOR_VERSION_INIT_OUT_SIZE {
                     ctx.reply_ok(
                         Some(

src/transport/fusedev/linux_session.rs

@@ -238,6 +238,19 @@ impl FuseSession {
         }
     }
 
+    /// Create a new fuse message channel with a specific buffer size.
+    pub fn with_writer<F>(&mut self, f: F)
+    where
+        F: FnOnce(FuseDevWriter),
+    {
+        if let Some(file) = &self.file {
+            let fd = file.as_raw_fd();
+            let mut buf = vec![0x0u8; self.bufsize];
+            let writer = FuseDevWriter::new(fd, &mut buf).unwrap();
+            f(writer);
+        }
+    }
+
     /// Wake channel loop and exit
     pub fn wake(&self) -> Result<()> {
         let wakers = self

tests/passthrough/src/main.rs

@@ -1,3 +1,5 @@
+use fuse_backend_rs::transport::FuseDevWriter;
+use fuse_backend_rs::transport::Writer;
 use log::{error, info, warn, LevelFilter};
 use std::env;
 use std::fs;
@@ -11,6 +13,7 @@ use signal_hook::{consts::TERM_SIGNALS, iterator::Signals};
 use fuse_backend_rs::api::{server::Server, Vfs, VfsOptions};
 use fuse_backend_rs::passthrough::{Config, PassthroughFs};
 use fuse_backend_rs::transport::{FuseChannel, FuseSession};
+use std::os::unix::io::{AsRawFd, FromRawFd, RawFd};
 
 use simple_logger::SimpleLogger;
 
@@ -58,6 +61,11 @@ impl Daemon {
         let mut se =
             FuseSession::new(Path::new(&self.mountpoint), "testpassthrough", "", false).unwrap();
         se.mount().unwrap();
+        
+        se.with_writer(|writer| {
+            self.server.notify_resend(writer).unwrap();
+        });
+
         for _ in 0..self.thread_cnt {
             let mut server = FuseServer {
                 server: self.server.clone(),