improve capture error handling

input-capture/src/dummy.rs

@@ -6,6 +6,8 @@ use futures_core::Stream;
 
 use input_event::Event;
 
+use crate::CaptureError;
+
 use super::{CaptureHandle, InputCapture, Position};
 
 pub struct DummyInputCapture {}
@@ -37,7 +39,7 @@ impl InputCapture for DummyInputCapture {
 }
 
 impl Stream for DummyInputCapture {
-    type Item = io::Result<(CaptureHandle, Event)>;
+    type Item = Result<(CaptureHandle, Event), CaptureError>;
 
     fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         Poll::Pending

input-capture/src/lib.rs

@@ -92,7 +92,9 @@ impl Display for Backend {
     }
 }
 
-pub trait InputCapture: Stream<Item = io::Result<(CaptureHandle, Event)>> + Unpin {
+pub trait InputCapture:
+    Stream<Item = Result<(CaptureHandle, Event), CaptureError>> + Unpin
+{
     /// create a new client with the given id
     fn create(&mut self, id: CaptureHandle, pos: Position) -> io::Result<()>;
 
@@ -105,8 +107,10 @@ pub trait InputCapture: Stream<Item = io::Result<(CaptureHandle, Event)>> + Unpi
 
 pub async fn create_backend(
     backend: Backend,
-) -> Result<Box<dyn InputCapture<Item = io::Result<(CaptureHandle, Event)>>>, CaptureCreationError>
+) -> Result<
-{
+    Box<dyn InputCapture<Item = Result<(CaptureHandle, Event), CaptureError>>>,
+    CaptureCreationError,
+> {
     match backend {
         #[cfg(all(unix, feature = "libei", not(target_os = "macos")))]
         Backend::InputCapturePortal => Ok(Box::new(libei::LibeiInputCapture::new().await?)),
@@ -124,8 +128,10 @@ pub async fn create_backend(
 
 pub async fn create(
     backend: Option<Backend>,
-) -> Result<Box<dyn InputCapture<Item = io::Result<(CaptureHandle, Event)>>>, CaptureCreationError>
+) -> Result<
-{
+    Box<dyn InputCapture<Item = Result<(CaptureHandle, Event), CaptureError>>>,
+    CaptureCreationError,
+> {
     if let Some(backend) = backend {
         let b = create_backend(backend).await;
         if b.is_ok() {

input-capture/src/libei.rs

@@ -5,7 +5,7 @@ use ashpd::{
     },
     enumflags2::BitFlags,
 };
-use futures::StreamExt;
+use futures::{FutureExt, StreamExt};
 use reis::{
     ei::{self, keyboard::KeyState},
     eis::button::ButtonState,
@@ -17,9 +17,9 @@ use std::{
     collections::HashMap,
     io,
     os::unix::net::UnixStream,
-    pin::Pin,
+    pin::{pin, Pin},
     rc::Rc,
-    task::{ready, Context, Poll},
+    task::{Context, Poll},
 };
 use tokio::{
     sync::mpsc::{Receiver, Sender},
@@ -192,7 +192,7 @@ async fn libei_event_handler(
             .map_err(ReisConvertEventStreamError::from)?;
         log::trace!("from ei: {ei_event:?}");
         let client = current_client.get();
-        handle_ei_event(ei_event, client, &context, &event_tx).await;
+        handle_ei_event(ei_event, client, &context, &event_tx).await?;
     }
 }
 
@@ -252,10 +252,9 @@ async fn do_capture<'a>(
         if active_clients.is_empty() {
             wait_for_active_client(&mut notify_rx, &mut active_clients).await;
             if notify_rx.is_closed() {
-                break Ok(());
+                break;
-            } else {
-                continue;
             }
+            continue;
         }
 
         let current_client = Rc::new(Cell::new(None));
@@ -270,8 +269,7 @@ async fn do_capture<'a>(
         let (context, ei_event_stream) = connect_to_eis(input_capture, &session).await?;
 
         // async event task
-        let mut ei_task: JoinHandle<Result<(), CaptureError>> =
+        let mut ei_task = pin!(libei_event_handler(
-            tokio::task::spawn_local(libei_event_handler(
             ei_event_stream,
             context,
             event_tx.clone(),
@@ -320,10 +318,8 @@ async fn do_capture<'a>(
                             break;
                         }
                         res = &mut ei_task => {
-                            if let Err(e) = res.expect("ei task paniced") {
+                            /* propagate errors to toplevel task */
-                                log::warn!("libei task exited: {e}");
+                            res?;
-                            }
-                            break;
                         }
                     }
                     release_capture(
@@ -342,19 +338,16 @@ async fn do_capture<'a>(
                     }
                 },
                 res = &mut ei_task => {
-                    if let Err(e) = res.expect("ei task paniced") {
+                    res?;
-                        log::warn!("libei task exited: {e}");
                 }
+            }
+        }
+        input_capture.disable(&session).await?;
+        if event_tx.is_closed() {
             break;
         }
     }
-        }
+    Ok(())
-        ei_task.abort();
-        input_capture.disable(&session).await?;
-        if event_tx.is_closed() {
-            break Ok(());
-        }
-    }
 }
 
 async fn release_capture(
@@ -410,7 +403,7 @@ async fn handle_ei_event(
     current_client: Option<CaptureHandle>,
     context: &ei::Context,
     event_tx: &Sender<(CaptureHandle, Event)>,
-) {
+) -> Result<(), CaptureError> {
     match ei_event {
         EiEvent::SeatAdded(s) => {
             s.seat.bind_capabilities(&[
@@ -421,7 +414,7 @@ async fn handle_ei_event(
                 DeviceCapability::Scroll,
                 DeviceCapability::Button,
             ]);
-            context.flush().unwrap();
+            context.flush().map_err(|e| io::Error::new(e.kind(), e))?;
         }
         EiEvent::SeatRemoved(_) => {}
         EiEvent::DeviceAdded(_) => {}
@@ -439,7 +432,7 @@ async fn handle_ei_event(
                 event_tx
                     .send((current_client, Event::Keyboard(modifier_event)))
                     .await
-                    .unwrap();
+                    .map_err(|_| CaptureError::EndOfStream)?;
             }
         }
         EiEvent::Frame(_) => {}
@@ -459,7 +452,7 @@ async fn handle_ei_event(
                 event_tx
                     .send((current_client, Event::Pointer(motion_event)))
                     .await
-                    .unwrap();
+                    .map_err(|_| CaptureError::EndOfStream)?;
             }
         }
         EiEvent::PointerMotionAbsolute(_) => {}
@@ -476,7 +469,7 @@ async fn handle_ei_event(
                 event_tx
                     .send((current_client, Event::Pointer(button_event)))
                     .await
-                    .unwrap();
+                    .map_err(|_| CaptureError::EndOfStream)?;
             }
         }
         EiEvent::ScrollDelta(delta) => {
@@ -500,7 +493,7 @@ async fn handle_ei_event(
                     event_tx
                         .send((handle, Event::Pointer(event)))
                         .await
-                        .unwrap();
+                        .map_err(|_| CaptureError::EndOfStream)?;
                 }
             }
         }
@@ -516,7 +509,7 @@ async fn handle_ei_event(
                     event_tx
                         .send((current_client, Event::Pointer(event)))
                         .await
-                        .unwrap();
+                        .map_err(|_| CaptureError::EndOfStream)?;
                 }
             }
             if scroll.discrete_dx != 0 {
@@ -528,7 +521,7 @@ async fn handle_ei_event(
                     event_tx
                         .send((current_client, Event::Pointer(event)))
                         .await
-                        .unwrap();
+                        .map_err(|_| CaptureError::EndOfStream)?;
                 }
             };
         }
@@ -545,7 +538,7 @@ async fn handle_ei_event(
                 event_tx
                     .send((current_client, Event::Keyboard(key_event)))
                     .await
-                    .unwrap();
+                    .map_err(|_| CaptureError::EndOfStream)?;
             }
         }
         EiEvent::TouchDown(_) => {}
@@ -555,6 +548,7 @@ async fn handle_ei_event(
             log::error!("disconnect: {d:?}");
         }
     }
+    Ok(())
 }
 
 impl<'a> LanMouseInputCapture for LibeiInputCapture<'a> {
@@ -584,12 +578,15 @@ impl<'a> LanMouseInputCapture for LibeiInputCapture<'a> {
 }
 
 impl<'a> Stream for LibeiInputCapture<'a> {
-    type Item = io::Result<(CaptureHandle, Event)>;
+    type Item = Result<(CaptureHandle, Event), CaptureError>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        match ready!(self.event_rx.poll_recv(cx)) {
+        match self.libei_task.poll_unpin(cx) {
-            None => Poll::Ready(None),
+            Poll::Ready(r) => match r.expect("failed to join") {
-            Some(e) => Poll::Ready(Some(Ok(e))),
+                Ok(()) => Poll::Ready(None),
+                Err(e) => Poll::Ready(Some(Err(e))),
+            },
+            Poll::Pending => self.event_rx.poll_recv(cx).map(|e| e.map(Result::Ok)),
         }
     }
 }

input-capture/src/macos.rs

@@ -1,4 +1,6 @@
-use crate::{error::MacOSInputCaptureCreationError, CaptureHandle, InputCapture, Position};
+use crate::{
+    error::MacOSInputCaptureCreationError, CaptureError, CaptureHandle, InputCapture, Position,
+};
 use futures_core::Stream;
 use input_event::Event;
 use std::task::{Context, Poll};
@@ -13,7 +15,7 @@ impl MacOSInputCapture {
 }
 
 impl Stream for MacOSInputCapture {
-    type Item = io::Result<(CaptureHandle, Event)>;
+    type Item = Result<(CaptureHandle, Event), CaptureError>;
 
     fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         Poll::Pending

input-capture/src/wayland.rs

@@ -62,6 +62,8 @@ use tempfile;
 
 use input_event::{Event, KeyboardEvent, PointerEvent};
 
+use crate::CaptureError;
+
 use super::{
     error::{LayerShellCaptureCreationError, WaylandBindError},
     CaptureHandle, InputCapture, Position,
@@ -582,7 +584,7 @@ impl InputCapture for WaylandInputCapture {
 }
 
 impl Stream for WaylandInputCapture {
-    type Item = io::Result<(CaptureHandle, Event)>;
+    type Item = Result<(CaptureHandle, Event), CaptureError>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         if let Some(event) = self.0.get_mut().state.pending_events.pop_front() {
@@ -600,7 +602,7 @@ impl Stream for WaylandInputCapture {
                     // prepare next read
                     match inner.prepare_read() {
                         Ok(_) => {}
-                        Err(e) => return Poll::Ready(Some(Err(e))),
+                        Err(e) => return Poll::Ready(Some(Err(e.into()))),
                     }
                 }
 
@@ -610,14 +612,14 @@ impl Stream for WaylandInputCapture {
                 // flush outgoing events
                 if let Err(e) = inner.flush_events() {
                     if e.kind() != ErrorKind::WouldBlock {
-                        return Poll::Ready(Some(Err(e)));
+                        return Poll::Ready(Some(Err(e.into())));
                     }
                 }
 
                 // prepare for the next read
                 match inner.prepare_read() {
                     Ok(_) => {}
-                    Err(e) => return Poll::Ready(Some(Err(e))),
+                    Err(e) => return Poll::Ready(Some(Err(e.into()))),
                 }
             }
 

input-capture/src/windows.rs

@@ -36,7 +36,7 @@ use input_event::{
     Event, KeyboardEvent, PointerEvent, BTN_BACK, BTN_FORWARD, BTN_LEFT, BTN_MIDDLE, BTN_RIGHT,
 };
 
-use super::{CaptureHandle, InputCapture, Position};
+use super::{CaptureError, CaptureHandle, InputCapture, Position};
 
 enum Request {
     Create(CaptureHandle, Position),
@@ -609,7 +609,7 @@ impl WindowsInputCapture {
 }
 
 impl Stream for WindowsInputCapture {
-    type Item = io::Result<(CaptureHandle, Event)>;
+    type Item = Result<(CaptureHandle, Event), CaptureError>;
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         match ready!(self.event_rx.poll_recv(cx)) {
             None => Poll::Ready(None),

input-capture/src/x11.rs

@@ -3,6 +3,8 @@ use std::task::Poll;
 
 use futures_core::Stream;
 
+use crate::CaptureError;
+
 use super::InputCapture;
 use input_event::Event;
 
@@ -32,7 +34,7 @@ impl InputCapture for X11InputCapture {
 }
 
 impl Stream for X11InputCapture {
-    type Item = io::Result<(CaptureHandle, Event)>;
+    type Item = Result<(CaptureHandle, Event), CaptureError>;
 
     fn poll_next(
         self: std::pin::Pin<&mut Self>,

resources/window.ui

@@ -118,6 +118,25 @@
                             </child>
                           </object>
                         </child>
+                        <child>
+                          <object class="AdwActionRow">
+                            <property name="title">capture /  emulation status</property>
+                            <child>
+                              <object class="GtkButton" id="input_emulation_button">
+                                <property name="icon-name">input-mouse-symbolic</property>
+                                <property name="valign">center</property>
+                                <signal name="clicked" handler="handle_emulation" swapped="true"/>
+                              </object>
+                            </child>
+                            <child>
+                              <object class="GtkButton" id="input_capture_button">
+                                <property name="icon-name">input-mouse-symbolic</property>
+                                <property name="valign">center</property>
+                                <signal name="clicked" handler="handle_capture" swapped="true"/>
+                              </object>
+                            </child>
+                          </object>
+                        </child>
                       </object>
                     </child>
                     <child>

src/frontend.rs

@@ -109,6 +109,10 @@ pub enum FrontendRequest {
     UpdateFixIps(ClientHandle, Vec<IpAddr>),
     /// request the state of the given client
     GetState(ClientHandle),
+    /// request reenabling input capture
+    EnableCapture,
+    /// request reenabling input emulation
+    EnableEmulation,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

src/frontend/gtk/window.rs

@@ -215,6 +215,13 @@ impl Window {
         }
     }
 
+    pub fn request_capture(&self) {
+        self.request(FrontendRequest::EnableCapture);
+    }
+
+    pub fn request_emulation(&self) {
+        self.request(FrontendRequest::EnableEmulation);
+    }
     pub fn request_client_state(&self, client: &ClientObject) {
         let handle = client.handle();
         let event = FrontendRequest::GetState(handle);

src/frontend/gtk/window/imp.rs

@@ -30,6 +30,10 @@ pub struct Window {
     pub hostname_label: TemplateChild<Label>,
     #[template_child]
     pub toast_overlay: TemplateChild<ToastOverlay>,
+    #[template_child]
+    pub input_emulation_button: TemplateChild<Button>,
+    #[template_child]
+    pub input_capture_button: TemplateChild<Button>,
     pub clients: RefCell<Option<gio::ListStore>>,
     #[cfg(unix)]
     pub stream: RefCell<Option<UnixStream>>,
@@ -100,6 +104,17 @@ impl Window {
         self.port_edit_cancel.set_visible(false);
     }
 
+    #[template_callback]
+    fn handle_emulation(&self) {
+        self.obj().request_emulation();
+    }
+
+    #[template_callback]
+    fn handle_capture(&self) {
+        log::info!("requesting capture");
+        self.obj().request_capture();
+    }
+
     pub fn set_port(&self, port: u16) {
         self.port.set(port);
         if port == DEFAULT_PORT {

src/server/capture_task.rs

@@ -22,6 +22,8 @@ pub enum CaptureEvent {
     Destroy(CaptureHandle),
     /// termination signal
     Terminate,
+    /// restart input capture
+    Restart,
 }
 
 pub fn new(
@@ -55,6 +57,13 @@ pub fn new(
                             }
                             CaptureEvent::Create(h, p) => capture.create(h, p)?,
                             CaptureEvent::Destroy(h) => capture.destroy(h)?,
+                            CaptureEvent::Restart => {
+                                let clients = server.client_manager.borrow().get_client_states().map(|(h, (c,_))| (h, c.pos)).collect::<Vec<_>>();
+                                capture = input_capture::create(backend).await?;
+                                for (handle, pos) in clients {
+                                    capture.create(handle, pos.into())?;
+                                }
+                            }
                             CaptureEvent::Terminate => break,
                         },
                         None => break,

src/server/emulation_task.rs

@@ -30,6 +30,8 @@ pub enum EmulationEvent {
     ReleaseKeys(ClientHandle),
     /// termination signal
     Terminate,
+    /// restart input emulation
+    Restart,
 }
 
 pub fn new(
@@ -90,6 +92,13 @@ async fn emulation_task(
                         EmulationEvent::Create(h) => emulation.create(h).await,
                         EmulationEvent::Destroy(h) => emulation.destroy(h).await,
                         EmulationEvent::ReleaseKeys(c) => release_keys(&server, &mut emulation, c).await?,
+                        EmulationEvent::Restart => {
+                            let clients = server.client_manager.borrow().get_client_states().map(|(h, _)| h).collect::<Vec<_>>();
+                            emulation = input_emulation::create(backend).await?;
+                            for handle in clients {
+                                emulation.create(handle).await;
+                            }
+                        },
                         EmulationEvent::Terminate => break,
                     },
                     None => break,

src/server/frontend_task.rs

@@ -106,6 +106,12 @@ async fn handle_frontend_event(
 ) -> bool {
     log::debug!("frontend: {event:?}");
     match event {
+        FrontendRequest::EnableCapture => {
+            let _ = capture.send(CaptureEvent::Restart).await;
+        }
+        FrontendRequest::EnableEmulation => {
+            let _ = emulate.send(EmulationEvent::Restart).await;
+        }
         FrontendRequest::Create => {
             let handle = add_client(server, frontend).await;
             resolve_dns(server, resolve_tx, handle).await;