feat(fusedev): add try_with_writer on FuseSession; update

src/transport/fusedev/linux_session.rs

@@ -16,6 +16,7 @@ use std::os::unix::net::UnixStream;
 use std::path::{Path, PathBuf};
 use std::sync::{Arc, Mutex};
 
+use crate::transport::fusedev::FuseSessionExt;
 use mio::{Events, Poll, Token, Waker};
 use nix::errno::Errno;
 use nix::fcntl::{fcntl, FcntlArg, FdFlag, OFlag};
@@ -238,19 +239,6 @@ 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
@@ -281,6 +269,16 @@ impl Drop for FuseSession {
     }
 }
 
+impl FuseSessionExt for FuseSession {
+    fn file(&self) -> Option<&File> {
+        self.file.as_ref()
+    }
+
+    fn bufsize(&self) -> usize {
+        self.bufsize
+    }
+}
+
 /// A fuse channel abstraction.
 ///
 /// Each session can hold multiple channels.

src/transport/fusedev/macos_session.rs

@@ -35,6 +35,7 @@ use super::{
     Error::IoError, Error::SessionFailure, FuseBuf, FuseDevWriter, Reader, Result,
     FUSE_HEADER_SIZE, FUSE_KERN_BUF_PAGES,
 };
+use crate::transport::fusedev::FuseSessionExt;
 use crate::transport::pagesize;
 
 const OSXFUSE_MOUNT_PROG: &str = "/Library/Filesystems/macfuse.fs/Contents/Resources/mount_macfuse";
@@ -203,6 +204,16 @@ impl Drop for FuseSession {
     }
 }
 
+impl FuseSessionExt for FuseSession {
+    fn file(&self) -> Option<&File> {
+        self.file.as_ref()
+    }
+
+    fn bufsize(&self) -> usize {
+        self.bufsize
+    }
+}
+
 /// A fuse channel abstruction. Each session can hold multiple channels.
 pub struct FuseChannel {
     file: File,

src/transport/fusedev/mod.rs

@@ -11,6 +11,7 @@ use std::collections::VecDeque;
 use std::io::{self, IoSlice, Write};
 use std::marker::PhantomData;
 use std::mem::ManuallyDrop;
+use std::os::fd::AsRawFd;
 use std::os::unix::io::RawFd;
 
 use nix::sys::uio::writev;
@@ -334,6 +335,44 @@ impl<S: BitmapSlice> Write for FuseDevWriter<'_, S> {
     }
 }
 
+/// Extension trait for FuseSession to provide helper methods.
+pub trait FuseSessionExt {
+    /// Get the underlying file of the fuse session.
+    fn file(&self) -> Option<&std::fs::File>;
+
+    /// Get the buffer size of the fuse session.
+    fn bufsize(&self) -> usize;
+
+    /// Create a new fuse message writer and pass it to the given closure.
+    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);
+        }
+    }
+
+    /// Create a new fuse message writer and pass it to the given closure. and return the result from the closure.
+    fn try_with_writer<F, R, E>(&mut self, f: F) -> std::result::Result<R, E>
+    where
+        F: FnOnce(FuseDevWriter) -> std::result::Result<R, E>,
+        E: From<Error>,
+    {
+        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)?;
+            f(writer)
+        } else {
+            Err(Error::SessionFailure("invalid fuse session".into()).into())
+        }
+    }
+}
+
 #[cfg(feature = "async-io")]
 mod async_io {
     use super::*;

src/transport/mod.rs

@@ -42,7 +42,7 @@ mod virtiofs;
 
 pub use self::fs_cache_req_handler::FsCacheReqHandler;
 #[cfg(feature = "fusedev")]
-pub use self::fusedev::{FuseBuf, FuseChannel, FuseDevWriter, FuseSession};
+pub use self::fusedev::{FuseBuf, FuseChannel, FuseDevWriter, FuseSession, FuseSessionExt};
 #[cfg(feature = "virtiofs")]
 pub use self::virtiofs::VirtioFsWriter;
 

tests/passthrough/src/main.rs

@@ -1,3 +1,4 @@
+use fuse_backend_rs::transport::FuseSessionExt as _;
 use log::{error, info, warn, LevelFilter};
 use std::env;
 use std::fs;
@@ -23,6 +24,17 @@ pub struct Daemon {
     session: Option<FuseSession>,
 }
 
+pub enum PassthroughFsError {
+    FuseError(fuse_backend_rs::Error),
+    TransportError(fuse_backend_rs::transport::Error),
+}
+
+impl From<fuse_backend_rs::transport::Error> for PassthroughFsError {
+    fn from(e: fuse_backend_rs::transport::Error) -> Self {
+        PassthroughFsError::TransportError(e)
+    }
+}
+
 #[allow(dead_code)]
 impl Daemon {
     /// Creates a fusedev daemon instance
@@ -59,11 +71,10 @@ impl Daemon {
             FuseSession::new(Path::new(&self.mountpoint), "testpassthrough", "", false).unwrap();
         se.mount().unwrap();
 
-        se.with_writer(|writer| {
+        se.try_with_writer(|writer| {
             self.server
-                .notify_resend(writer)
-                .unwrap_or_else(|e| println!("failed to send resend notification {}", e));
-        });
+                .notify_resend(writer).map_err(PassthroughFsError::FuseError)
+        }).map_err(|_| Error::from_raw_os_error(libc::EINVAL))?;
 
         for _ in 0..self.thread_cnt {
             let mut server = FuseServer {