wlroots: Fix crash when socket is overwhelmed

Previously when the output buffer was overwhelmed, additional
events were submitted until the outgoing buffer filled up, which
causes the wayland-connection to 'break' and not accept further attempts
to flush() the socket.

.gitignore

@@ -1 +1,2 @@
 /target
+.gdbinit

src/backend/consumer/libei.rs

@@ -7,7 +7,7 @@ impl LibeiConsumer {
 }
 
 impl EventConsumer for LibeiConsumer {
-    fn consume(&self, _: crate::event::Event, _: crate::client::ClientHandle) {
+    fn consume(&mut self, _: crate::event::Event, _: crate::client::ClientHandle) {
         log::error!("libei backend not yet implemented!");
         todo!()
     }

src/backend/consumer/wlroots.rs

@@ -1,7 +1,9 @@
 use wayland_client::WEnum;
+use wayland_client::backend::WaylandError;
 use crate::client::{ClientHandle, ClientEvent};
 use crate::consumer::EventConsumer;
 use std::collections::HashMap;
+use std::io;
 use std::os::fd::OwnedFd;
 use std::os::unix::prelude::AsRawFd;
 
@@ -46,6 +48,7 @@ struct State {
 
 // App State, implements Dispatch event handlers
 pub(crate) struct WlrootsConsumer {
+    last_flush_failed: bool,
     state: State,
     queue: EventQueue<State>,
 }
@@ -89,6 +92,7 @@ impl WlrootsConsumer {
         let input_for_client: HashMap<ClientHandle, VirtualInput> = HashMap::new();
 
         let mut consumer = WlrootsConsumer {
+            last_flush_failed: false,
             state: State {
                 keymap: None,
                 input_for_client,
@@ -137,11 +141,36 @@ impl State {
 }
 
 impl EventConsumer for WlrootsConsumer {
-    fn consume(&self, event: Event, client_handle: ClientHandle) {
+    fn consume(&mut self, event: Event, client_handle: ClientHandle) {
         if let Some(virtual_input) = self.state.input_for_client.get(&client_handle) {
+            if self.last_flush_failed {
+                if let Err(WaylandError::Io(e)) = self.queue.flush() {
+                    if e.kind() == io::ErrorKind::WouldBlock {
+                        /*
+                         * outgoing buffer is full - sending more events
+                         * will overwhelm the output buffer and leave the
+                         * wayland connection in a broken state
+                         */
+                        log::warn!("can't keep up, discarding event: ({client_handle}) - {event:?}");
+                        return
+                    }
+                }
+            }
             virtual_input.consume_event(event).unwrap();
-            if let Err(e) = self.queue.flush() {
-                log::error!("{}", e);
+            match self.queue.flush() {
+                Err(WaylandError::Io(e)) if e.kind() == io::ErrorKind::WouldBlock => {
+                    self.last_flush_failed = true;
+                    log::warn!("can't keep up, retrying ...");
+                }
+                Err(WaylandError::Io(e)) => {
+                    log::error!("{e}")
+                },
+                Err(WaylandError::Protocol(e)) => {
+                    panic!("wayland protocol violation: {e}")
+                }
+                Ok(()) => {
+                    self.last_flush_failed = false;
+                },
             }
         }
     }

src/backend/consumer/x11.rs

@@ -31,7 +31,7 @@ impl X11Consumer {
 }
 
 impl EventConsumer for X11Consumer {
-    fn consume(&self, event: Event, _: ClientHandle) {
+    fn consume(&mut self, event: Event, _: ClientHandle) {
         match event {
             Event::Pointer(pointer_event) => match pointer_event {
                 crate::event::PointerEvent::Motion {

src/backend/consumer/xdg_desktop_portal.rs

@@ -7,7 +7,7 @@ impl DesktopPortalConsumer {
 }
 
 impl EventConsumer for DesktopPortalConsumer {
-    fn consume(&self, _: crate::event::Event, _: crate::client::ClientHandle) {
+    fn consume(&mut self, _: crate::event::Event, _: crate::client::ClientHandle) {
         log::error!("xdg_desktop_portal backend not yet implemented!");
     }
 

src/consumer.rs

@@ -15,7 +15,7 @@ enum Backend {
 
 pub trait EventConsumer {
     /// Event corresponding to an abstract `client_handle`
-    fn consume(&self, event: Event, client_handle: ClientHandle);
+    fn consume(&mut self, event: Event, client_handle: ClientHandle);
 
     /// Event corresponding to a configuration change
     fn notify(&mut self, client_event: ClientEvent);