Adding usbredir to the mix

qemu-display/Cargo.toml

@@ -20,3 +20,4 @@ serde_bytes = "0.11.5"
 futures-util = { version = "0.3.8", features = ["async-await-macro"] }
 once_cell = "1.5"
 futures = "0.3.13"
+usbredirhost = "0.0.1"

qemu-display/src/chardev.rs

@@ -32,7 +32,6 @@ pub trait Chardev {
 #[derive(derivative::Derivative)]
 #[derivative(Debug)]
 pub struct Chardev {
-    #[derivative(Debug = "ignore")]
     pub proxy: AsyncChardevProxy<'static>,
 }
 

qemu-display/src/display.rs

@@ -4,7 +4,7 @@ use zbus::azync::Connection;
 use zbus::fdo::ManagedObjects;
 use zvariant::OwnedObjectPath;
 
-use crate::{Audio, Chardev, Result, UsbRedir};
+use crate::{Audio, Chardev, Clipboard, Result, UsbRedir};
 
 pub struct Display {
     conn: Connection,
@@ -38,6 +38,17 @@ impl Display {
         Ok(Some(Audio::new(&self.conn).await?))
     }
 
+    pub async fn clipboard(&self) -> Result<Option<Clipboard>> {
+        if !self
+            .objects
+            .contains_key(&OwnedObjectPath::try_from("/org/qemu/Display1/Clipboard").unwrap())
+        {
+            return Ok(None);
+        }
+
+        Ok(Some(Clipboard::new(&self.conn).await?))
+    }
+
     pub async fn chardevs(&self) -> Vec<Chardev> {
         stream::iter(&self.objects)
             .filter_map(|(p, _ifaces)| async move {

qemu-display/src/error.rs

@@ -1,3 +1,5 @@
+use usbredirhost::rusb;
+
 use std::error;
 use std::fmt;
 use std::io;
@@ -6,14 +8,18 @@ use std::io;
 pub enum Error {
     Io(io::Error),
     Zbus(zbus::Error),
+    Rusb(rusb::Error),
+    Usbredir(usbredirhost::Error),
     Failed(String),
 }
 
 impl fmt::Display for Error {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
-            Error::Io(e) => write!(f, "{}", e),
+            Error::Io(e) => write!(f, "IO error: {}", e),
-            Error::Zbus(e) => write!(f, "{}", e),
+            Error::Zbus(e) => write!(f, "zbus error: {}", e),
+            Error::Rusb(e) => write!(f, "rusb error: {}", e),
+            Error::Usbredir(e) => write!(f, "usbredir error: {}", e),
             Error::Failed(e) => write!(f, "{}", e),
         }
     }
@@ -24,6 +30,8 @@ impl error::Error for Error {
         match self {
             Error::Io(e) => Some(e),
             Error::Zbus(e) => Some(e),
+            Error::Rusb(e) => Some(e),
+            Error::Usbredir(e) => Some(e),
             Error::Failed(_) => None,
         }
     }
@@ -53,4 +61,16 @@ impl From<zvariant::Error> for Error {
     }
 }
 
+impl From<rusb::Error> for Error {
+    fn from(e: rusb::Error) -> Self {
+        Error::Rusb(e)
+    }
+}
+
+impl From<usbredirhost::Error> for Error {
+    fn from(e: usbredirhost::Error) -> Self {
+        Error::Usbredir(e)
+    }
+}
+
 pub type Result<T> = std::result::Result<T, Error>;

qemu-display/src/usbredir.rs

@@ -1,10 +1,260 @@
-use crate::Chardev;
+use std::{
+    cell::RefCell,
+    collections::HashMap,
+    default::Default,
+    io::{Read, Write},
+    os::unix::{
+        io::{AsRawFd, RawFd},
+        net::UnixStream,
+    },
+    sync::{Arc, Mutex},
+    thread::JoinHandle,
+};
 
-pub struct UsbRedir;
+use usbredirhost::{
+    rusb::{self, UsbContext},
+    Device, DeviceHandler, LogLevel,
+};
+
+use crate::{Chardev, Error, Result};
+
+#[derive(Debug)]
+struct InnerHandler {
+    device_fd: Option<zvariant::OwnedFd>,
+    stream: UnixStream,
+    stream_thread: JoinHandle<()>,
+    ctxt: rusb::Context,
+    ctxt_thread: Option<JoinHandle<()>>,
+    quit: bool,
+}
+
+#[derive(Clone, Debug)]
+struct Handler {
+    inner: Arc<Mutex<InnerHandler>>,
+}
+
+impl DeviceHandler for Handler {
+    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        let mut inner = self.inner.lock().unwrap();
+        let read = match fd_poll_readable(inner.stream.as_raw_fd(), false) {
+            Ok(true) => {
+                let read = inner.stream.read(buf);
+                if let Ok(0) = read {
+                    Err(std::io::Error::new(
+                        std::io::ErrorKind::BrokenPipe,
+                        "disconnected",
+                    ))
+                } else {
+                    read
+                }
+            }
+            Ok(false) => Ok(0),
+            Err(e) => Err(e),
+        };
+
+        inner.quit = read.is_err();
+        read
+    }
+
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        let mut inner = self.inner.lock().unwrap();
+        let write = inner.stream.write_all(buf);
+        inner.quit = write.is_err();
+        write?;
+        Ok(buf.len())
+    }
+
+    fn log(&mut self, _level: LogLevel, _msg: &str) {}
+
+    fn flush_writes(&mut self) {}
+}
+
+#[zbus::dbus_proxy(
+    interface = "org.freedesktop.usbredir1",
+    default_service = "org.freedesktop.usbredir1",
+    default_path = "/org/freedesktop/usbredir1"
+)]
+trait SystemHelper {
+    fn open_bus_dev(&self, bus: u8, dev: u8) -> zbus::fdo::Result<zbus::zvariant::OwnedFd>;
+}
+
+impl Handler {
+    async fn new(device: &rusb::Device<rusb::Context>, chardev: &Chardev) -> Result<Self> {
+        let ctxt = device.context().clone();
+
+        let (dev, device_fd) = match device.open() {
+            Ok(it) => (it, None),
+            Err(rusb::Error::Access) => {
+                let (bus, dev) = (device.bus_number(), device.address());
+                let sysbus = zbus::azync::Connection::system().await?;
+                let fd = AsyncSystemHelperProxy::new(&sysbus)
+                    .await?
+                    .open_bus_dev(bus, dev)
+                    .await?;
+                unsafe { (ctxt.open_device_with_fd(fd.as_raw_fd())?, Some(fd)) }
+            }
+            Err(e) => {
+                return Err(e.into());
+            }
+        };
+
+        let (stream, peer) = UnixStream::pair()?;
+        chardev.proxy.register(dbg!(peer.as_raw_fd()).into()).await?;
+
+        let c = ctxt.clone();
+        let stream_fd = stream.as_raw_fd();
+        dbg!(stream_fd);
+        // really annoying libusb/usbredir APIs...
+        let stream_thread = std::thread::spawn(move || loop {
+            let ret = fd_poll_readable(stream_fd, true);
+            c.interrupt_handle_events();
+            if ret.is_err() {
+                dbg!();
+                break;
+            }
+        });
+
+        let handler = Self {
+            inner: Arc::new(Mutex::new(InnerHandler {
+                device_fd,
+                stream,
+                stream_thread,
+                quit: false,
+                ctxt: ctxt.clone(),
+                ctxt_thread: Default::default(),
+            })),
+        };
+
+        let redirdev = Device::new(&ctxt, Some(dev), handler.clone(), LogLevel::None as _)?;
+        let c = ctxt.clone();
+        let inner = handler.inner.clone();
+        let ctxt_thread = std::thread::spawn(move || loop {
+            if inner.lock().unwrap().quit {
+                dbg!();
+                break;
+            }
+            if let Ok(true) = fd_poll_readable(stream_fd, false) {
+                redirdev.read_peer().unwrap();
+            }
+            if redirdev.has_data_to_write() > 0 {
+                redirdev.write_peer().unwrap();
+            }
+            c.handle_events(None).unwrap();
+        });
+        handler
+            .inner
+            .lock()
+            .unwrap()
+            .ctxt_thread
+            .replace(ctxt_thread);
+
+        Ok(handler)
+    }
+}
+
+impl Drop for InnerHandler {
+    fn drop(&mut self) {
+        //FIXME: for some reason close stream doesn't HUP qemu ??
+        dbg!()
+    }
+}
+
+impl Drop for Handler {
+    fn drop(&mut self) {
+        let mut inner = self.inner.lock().unwrap();
+        inner.quit = true;
+        inner.ctxt.interrupt_handle_events();
+    }
+}
+
+#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
+struct Key(u8, u8);
+
+impl Key {
+    fn from_device(device: &rusb::Device<rusb::Context>) -> Self {
+        Self(device.bus_number(), device.address())
+    }
+}
+
+#[derive(Debug)]
+struct Inner {
+    chardevs: Vec<Chardev>,
+    handlers: HashMap<Key, Handler>,
+}
+
+impl Inner {
+    async fn available_chardev(&self) -> Option<&Chardev> {
+        for c in &self.chardevs {
+            if c.proxy.owner().await.unwrap_or_default().is_empty() {
+                return Some(c);
+            }
+        }
+        None
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct UsbRedir {
+    inner: Arc<RefCell<Inner>>,
+}
 
 impl UsbRedir {
     pub fn new(chardevs: Vec<Chardev>) -> Self {
-        dbg!(chardevs);
+        Self {
-        Self
+            inner: Arc::new(RefCell::new(Inner {
+                chardevs,
+                handlers: Default::default(),
+            })),
+        }
+    }
+
+    pub async fn set_device_state(
+        &self,
+        device: &rusb::Device<rusb::Context>,
+        state: bool,
+    ) -> Result<bool> {
+        let mut inner = self.inner.borrow_mut();
+        let key = Key::from_device(device);
+
+        if state {
+            if !inner.handlers.contains_key(&key) {
+                let chardev = inner
+                    .available_chardev()
+                    .await
+                    .ok_or_else(|| Error::Failed("There are no free USB channels".into()))?;
+                let handler = Handler::new(device, chardev).await?;
+                inner.handlers.insert(key, handler);
+            }
+        } else {
+            inner.handlers.remove(&key);
+        }
+
+        Ok(state)
+    }
+
+    pub fn is_device_connected(&self, device: &rusb::Device<rusb::Context>) -> bool {
+        let inner = self.inner.borrow();
+
+        inner.handlers.contains_key(&Key::from_device(device))
+    }
+}
+
+fn fd_poll_readable(fd: RawFd, wait: bool) -> std::io::Result<bool> {
+    let mut fds = [libc::pollfd {
+        fd,
+        events: libc::POLLIN|libc::POLLHUP,
+        revents: 0,
+    }];
+    let ret = unsafe { libc::poll(fds.as_mut_ptr(), 1, if wait { -1 } else { 0 }) };
+    if ret > 0 {
+        if fds[0].revents & libc::POLLHUP != 0 {
+            Err(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "hup"))
+        } else {
+            Ok(fds[0].revents & libc::POLLIN != 0)
+        }
+    } else if ret == 0 {
+        Ok(false)
+    } else {
+        Err(std::io::Error::last_os_error())
     }
 }

qemu-rdw/src/clipboard.rs

@@ -17,9 +17,7 @@ pub struct Handler {
 }
 
 impl Handler {
-    pub async fn new(conn: &zbus::azync::Connection) -> Result<Self, Box<dyn Error>> {
+    pub async fn new(ctxt: Clipboard) -> Result<Self, Box<dyn Error>> {
-        let ctxt = Clipboard::new(conn).await?;
-
         let rx = ctxt
             .glib_listen()
             .await

qemu-rdw/src/display.rs

@@ -17,7 +17,7 @@ mod imp {
     }
 
     unsafe impl ClassStruct for RdwDisplayQemuClass {
-        type Type = DisplayQemu;
+        type Type = Display;
     }
 
     #[repr(C)]
@@ -34,24 +34,24 @@ mod imp {
     }
 
     unsafe impl InstanceStruct for RdwDisplayQemu {
-        type Type = DisplayQemu;
+        type Type = Display;
     }
 
     #[derive(Debug, Default)]
-    pub struct DisplayQemu {
+    pub struct Display {
         pub(crate) console: OnceCell<Console>,
     }
 
     #[glib::object_subclass]
-    impl ObjectSubclass for DisplayQemu {
+    impl ObjectSubclass for Display {
         const NAME: &'static str = "RdwDisplayQemu";
-        type Type = super::DisplayQemu;
+        type Type = super::Display;
         type ParentType = rdw::Display;
         type Class = RdwDisplayQemuClass;
         type Instance = RdwDisplayQemu;
     }
 
-    impl ObjectImpl for DisplayQemu {
+    impl ObjectImpl for Display {
         fn constructed(&self, obj: &Self::Type) {
             self.parent_constructed(obj);
 
@@ -129,7 +129,7 @@ mod imp {
         }
     }
 
-    impl WidgetImpl for DisplayQemu {
+    impl WidgetImpl for Display {
         fn realize(&self, widget: &Self::Type) {
             self.parent_realize(widget);
 
@@ -228,9 +228,9 @@ mod imp {
         }
     }
 
-    impl rdw::DisplayImpl for DisplayQemu {}
+    impl rdw::DisplayImpl for Display {}
 
-    impl DisplayQemu {
+    impl Display {
         pub(crate) fn set_console(&self, console: Console) {
             self.console.set(console).unwrap();
         }
@@ -238,19 +238,19 @@ mod imp {
 }
 
 glib::wrapper! {
-    pub struct DisplayQemu(ObjectSubclass<imp::DisplayQemu>) @extends rdw::Display, gtk::Widget, @implements gtk::Accessible, gtk::Buildable, gtk::ConstraintTarget;
+    pub struct Display(ObjectSubclass<imp::Display>) @extends rdw::Display, gtk::Widget, @implements gtk::Accessible, gtk::Buildable, gtk::ConstraintTarget;
 }
 
-impl DisplayQemu {
+impl Display {
     pub fn new(console: Console) -> Self {
         let obj = glib::Object::new::<Self>(&[]).unwrap();
-        let self_ = imp::DisplayQemu::from_instance(&obj);
+        let self_ = imp::Display::from_instance(&obj);
         self_.set_console(console);
         obj
     }
 
     pub(crate) fn console(&self) -> &Console {
-        let self_ = imp::DisplayQemu::from_instance(self);
+        let self_ = imp::Display::from_instance(self);
         self_.console.get().unwrap()
     }
 }

qemu-rdw/src/main.rs

@@ -1,61 +1,93 @@
 use gio::ApplicationFlags;
-use glib::{clone, MainContext};
+use glib::MainContext;
 use gtk::{gio, glib, prelude::*};
-use once_cell::sync::OnceCell;
 use qemu_display::{Chardev, Console, Display};
-use std::os::unix::io::AsRawFd;
+use std::cell::RefCell;
-use std::os::unix::net::UnixStream;
+use std::sync::Arc;
 use zbus::Connection;
 
 mod audio;
 mod clipboard;
-mod display_qemu;
+mod display;
+mod usbredir;
 
-fn main() {
+struct Inner {
-    pretty_env_logger::init();
+    app: gtk::Application,
+    conn: zbus::azync::Connection,
+    usbredir: RefCell<Option<usbredir::Handler>>,
+    audio: RefCell<Option<audio::Handler>>,
+    clipboard: RefCell<Option<clipboard::Handler>>,
+}
 
+#[derive(Clone)]
+struct App {
+    inner: Arc<Inner>,
+}
+
+impl App {
+    fn new() -> Self {
         let app = gtk::Application::new(Some("org.qemu.rdw.demo"), ApplicationFlags::NON_UNIQUE);
-
+        let conn = Connection::session()
-    let conn: zbus::azync::Connection = Connection::session()
             .expect("Failed to connect to DBus")
             .into();
 
-    let audio = std::sync::Arc::new(OnceCell::new());
+        let app = App {
-    let clipboard = std::sync::Arc::new(OnceCell::new());
+            inner: Arc::new(Inner {
+                app,
+                conn,
+                usbredir: Default::default(),
+                audio: Default::default(),
+                clipboard: Default::default(),
+            }),
+        };
 
-    app.connect_activate(move |app| {
+        let app_clone = app.clone();
-        let window = gtk::ApplicationWindow::new(app);
+        app.inner.app.connect_activate(move |app| {
+            let ui_src = include_str!("main.ui");
+            let builder = gtk::Builder::new();
+            builder
+                .add_from_string(ui_src)
+                .expect("Couldn't add from string");
+            let window: gtk::ApplicationWindow =
+                builder.object("window").expect("Couldn't get window");
+            window.set_application(Some(app));
 
-        window.set_title(Some("rdw demo"));
+            let app_clone = app_clone.clone();
-        window.set_default_size(1024, 768);
+            MainContext::default().spawn_local(async move {
+                let display = Display::new(app_clone.connection()).await.unwrap();
 
-        let conn = conn.clone();
+                let console = Console::new(app_clone.connection(), 0)
-        let audio_clone = audio.clone();
+                    .await
-        let clipboard_clone = clipboard.clone();
+                    .expect("Failed to get the QEMU console");
-        MainContext::default().spawn_local(clone!(@strong window => async move {
+                let rdw = display::Display::new(console);
-            let display = Display::new(&conn).await.unwrap();
+                app_clone
+                    .inner
+                    .app
+                    .active_window()
+                    .unwrap()
+                    .set_child(Some(&rdw));
 
-            let console = Console::new(&conn, 0).await.expect("Failed to get the QEMU console");
+                app_clone.set_usbredir(usbredir::Handler::new(display.usbredir().await));
-            let rdw = display_qemu::DisplayQemu::new(console);
-            window.set_child(Some(&rdw));
-
-            let usbredir = display.usbredir().await;
 
                 if let Ok(Some(audio)) = display.audio().await {
                     match audio::Handler::new(audio).await {
-                    Ok(handler) => audio_clone.set(handler).unwrap(),
+                        Ok(handler) => app_clone.set_audio(handler),
                         Err(e) => log::warn!("Failed to setup audio: {}", e),
                     }
                 }
 
-            match clipboard::Handler::new(&conn).await {
+                if let Ok(Some(clipboard)) = display.clipboard().await {
-                Ok(handler) => clipboard_clone.set(handler).unwrap(),
+                    match clipboard::Handler::new(clipboard).await {
+                        Ok(handler) => app_clone.set_clipboard(handler),
                         Err(e) => log::warn!("Failed to setup clipboard: {}", e),
                     }
+                }
 
-            if let Ok(c) = Chardev::new(&conn, "qmp").await {
+                if let Ok(c) = Chardev::new(app_clone.connection(), "qmp").await {
-                use std::io::BufReader;
                     use std::io::prelude::*;
+                    use std::io::BufReader;
+                    use std::os::unix::io::AsRawFd;
+                    use std::os::unix::net::UnixStream;
 
                     let (p0, p1) = UnixStream::pair().unwrap();
                     if c.proxy.register(p1.as_raw_fd().into()).await.is_ok() {
@@ -70,8 +102,49 @@ fn main() {
                 }
 
                 window.show();
-        }));
+            });
         });
 
+        let action_usb = gio::SimpleAction::new("usb", None);
+        let app_clone = app.clone();
+        action_usb.connect_activate(move |_, _| {
+            let usbredir = app_clone.inner.usbredir.borrow();
+            if let Some(usbredir) = usbredir.as_ref() {
+                let dialog = gtk::Dialog::new();
+                dialog.set_transient_for(app_clone.inner.app.active_window().as_ref());
+                dialog.set_child(Some(&usbredir.widget()));
+                dialog.show();
+            }
+        });
+        app.inner.app.add_action(&action_usb);
+
+        app
+    }
+
+    fn connection(&self) -> &zbus::azync::Connection {
+        &self.inner.conn
+    }
+
+    fn set_usbredir(&self, usbredir: usbredir::Handler) {
+        self.inner.usbredir.replace(Some(usbredir));
+    }
+
+    fn set_audio(&self, audio: audio::Handler) {
+        self.inner.audio.replace(Some(audio));
+    }
+
+    fn set_clipboard(&self, clipboard: clipboard::Handler) {
+        self.inner.clipboard.replace(Some(clipboard));
+    }
+
+    fn run(&self) -> i32 {
+        self.inner.app.run()
+    }
+}
+
+fn main() {
+    pretty_env_logger::init();
+
+    let app = App::new();
     app.run();
 }

qemu-rdw/src/main.ui

@@ -0,0 +1,19 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<interface>
+  <object class="GtkApplicationWindow" id="window">
+    <property name="title" translatable="yes">qemu-rdw demo</property>
+    <property name="default-width">1024</property>
+    <property name="default-height">768</property>
+    <child type="titlebar">
+      <object class="GtkHeaderBar" id="headerbar">
+        <property name="show_title_buttons">True</property>
+        <child type="end">
+          <object class="GtkButton">
+            <property name="action-name">app.usb</property>
+            <property name="label">USB devices</property>
+          </object>
+        </child>
+      </object>
+    </child>
+  </object>
+</interface>

qemu-rdw/src/usbredir.rs

@@ -0,0 +1,53 @@
+use glib::{clone, MainContext};
+use gtk::{glib, prelude::*};
+use qemu_display::UsbRedir;
+
+#[derive(Clone, Debug)]
+pub struct Handler {
+    usbredir: UsbRedir,
+}
+
+impl Handler {
+    pub fn new(usbredir: UsbRedir) -> Self {
+        Self { usbredir }
+    }
+
+    pub fn widget(&self) -> rdw::UsbRedir {
+        let widget = rdw::UsbRedir::new();
+
+        let usbredir = self.usbredir.clone();
+        widget
+            .model()
+            .connect_items_changed(clone!(@weak widget => move |model, pos, _rm, add| {
+                for pos in pos..pos + add {
+                    let item = model.item(pos).unwrap();
+                    if let Some(dev) = item.downcast_ref::<rdw::UsbDevice>().unwrap().device() {
+                        item.set_property("active", usbredir.is_device_connected(&dev)).unwrap();
+                    }
+                }
+            }));
+
+        let usbredir = self.usbredir.clone();
+        widget.connect_device_state_set(move |widget, item, state| {
+            let device = match item.device() {
+                Some(it) => it.clone(),
+                _ => return,
+            };
+
+            let usbredir = usbredir.clone();
+            MainContext::default().spawn_local(clone!(@weak item, @weak widget => async move {
+                match usbredir.set_device_state(&device, state).await {
+                    Ok(active) => item.set_property("active", active).unwrap(),
+                    Err(e) => {
+                        if state {
+                            item.set_property("active", false).unwrap();
+                        }
+                        widget.emit_by_name("show-error",&[&e.to_string()]).unwrap();
+                    },
+                }
+            }));
+        });
+
+        widget
+    }
+}