fix history size limit

fixes #5

Cargo.lock

@@ -522,18 +522,44 @@ dependencies = [
  "thiserror 1.0.69",
 ]
 
+[[package]]
+name = "calloop"
+version = "0.14.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cb9f6e1368bd4621d2c86baa7e37de77a938adf5221e5dd3d6133340101b309e"
+dependencies = [
+ "bitflags 2.9.4",
+ "nix",
+ "polling",
+ "rustix 1.1.2",
+ "slab",
+ "tracing",
+]
+
 [[package]]
 name = "calloop-wayland-source"
 version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "95a66a987056935f7efce4ab5668920b5d0dac4a7c99991a67395f13702ddd20"
 dependencies = [
- "calloop",
+ "calloop 0.13.0",
  "rustix 0.38.44",
  "wayland-backend",
  "wayland-client",
 ]
 
+[[package]]
+name = "calloop-wayland-source"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "138efcf0940a02ebf0cc8d1eff41a1682a46b431630f4c52450d6265876021fa"
+dependencies = [
+ "calloop 0.14.3",
+ "rustix 1.1.2",
+ "wayland-backend",
+ "wayland-client",
+]
+
 [[package]]
 name = "cc"
 version = "1.2.36"
@@ -621,12 +647,12 @@ dependencies = [
 name = "clippyboard-daemon"
 version = "0.1.0"
 dependencies = [
+ "calloop 0.14.3",
+ "calloop-wayland-source 0.4.1",
  "ciborium",
  "clippyboard-shared",
- "ctrlc",
  "dirs",
  "eyre",
- "rustix 1.1.2",
  "serde",
  "tracing",
  "tracing-subscriber",
@@ -757,17 +783,6 @@ version = "0.2.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5"
 
-[[package]]
-name = "ctrlc"
-version = "3.5.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "881c5d0a13b2f1498e2306e82cbada78390e152d4b1378fb28a84f4dcd0dc4f3"
-dependencies = [
- "dispatch",
- "nix",
- "windows-sys 0.61.0",
-]
-
 [[package]]
 name = "cursor-icon"
 version = "1.2.0"
@@ -792,7 +807,7 @@ dependencies = [
  "libc",
  "option-ext",
  "redox_users",
- "windows-sys 0.60.2",
+ "windows-sys 0.61.0",
 ]
 
 [[package]]
@@ -2680,7 +2695,7 @@ dependencies = [
  "errno",
  "libc",
  "linux-raw-sys 0.11.0",
- "windows-sys 0.60.2",
+ "windows-sys 0.61.0",
 ]
 
 [[package]]
@@ -2818,8 +2833,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3457dea1f0eb631b4034d61d4d8c32074caa6cd1ab2d59f2327bd8461e2c0016"
 dependencies = [
  "bitflags 2.9.4",
- "calloop",
- "calloop-wayland-source",
+ "calloop 0.13.0",
+ "calloop-wayland-source 0.3.0",
  "cursor-icon",
  "libc",
  "log",
@@ -3070,6 +3085,7 @@ version = "0.1.41"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "784e0ac535deb450455cbfa28a6f0df145ea1bb7ae51b821cf5e7927fdcfbdd0"
 dependencies = [
+ "log",
  "pin-project-lite",
  "tracing-attributes",
  "tracing-core",
@@ -4055,7 +4071,7 @@ dependencies = [
  "bitflags 2.9.4",
  "block2",
  "bytemuck",
- "calloop",
+ "calloop 0.13.0",
  "cfg_aliases",
  "concurrent-queue",
  "core-foundation 0.9.4",

clippyboard-daemon/Cargo.toml

@@ -6,13 +6,13 @@ edition = "2024"
 [dependencies]
 clippyboard-shared = { path = "../clippyboard-shared" }
 ciborium = "0.2.2"
-ctrlc = "3.5.0"
 dirs = "6.0.0"
 eyre = "0.6.12"
-rustix = "1.1.2"
 serde = "1.0.219"
 tracing = { version = "0.1.41", features = ["attributes"] }
 tracing-subscriber = { version = "0.3.20", features = ["env-filter"] }
 wayland-backend = { version = "0.3.11", features = ["client_system"] }
 wayland-client = "0.31.11"
 wayland-protocols = { version = "0.32.9", features = ["staging", "client"] }
+calloop = { version = "0.14.3", features = ["signals"] }
+calloop-wayland-source = "0.4.1"

clippyboard-daemon/src/main.rs

@@ -1,16 +1,19 @@
+use calloop::EventLoop;
+use calloop::Interest;
+use calloop::Mode;
+use calloop::generic::Generic;
+use calloop::signals::Signal;
+use calloop::signals::Signals;
+use calloop_wayland_source::WaylandSource;
 use clippyboard_shared::HistoryItem;
 use eyre::Context;
-use eyre::ContextCompat;
 use eyre::bail;
-use rustix::event::PollFd;
-use rustix::event::PollFlags;
-use rustix::fs::OFlags;
 use std::collections::HashMap;
 use std::collections::HashSet;
 use std::convert::Infallible;
+use std::fmt::Display;
 use std::io;
 use std::io::ErrorKind;
-use std::io::PipeReader;
 use std::io::{BufReader, BufWriter, PipeWriter, Read, Write};
 use std::os::fd::AsFd;
 use std::os::unix::net::{UnixListener, UnixStream};
@@ -24,7 +27,6 @@ use tracing::error;
 use tracing::info;
 use tracing::warn;
 use tracing_subscriber::EnvFilter;
-use wayland_client::EventQueue;
 use wayland_client::protocol::wl_registry::WlRegistry;
 use wayland_client::protocol::wl_seat::WlSeat;
 use wayland_client::{Dispatch, Proxy, QueueHandle, event_created_child};
@@ -38,19 +40,25 @@ use wayland_protocols::ext::data_control::v1::client::ext_data_control_offer_v1:
 use wayland_protocols::ext::data_control::v1::client::ext_data_control_source_v1;
 use wayland_protocols::ext::data_control::v1::client::ext_data_control_source_v1::ExtDataControlSourceV1;
 
-const MAX_ENTRY_SIZE: u64 = 50_000_000;
-const MAX_HISTORY_BYTE_SIZE: usize = 100_000_000;
+const MEGABYTE: usize = 1024 * 1024;
+const MAX_ENTRY_SIZE: usize = 50 * MEGABYTE;
+const MAX_HISTORY_BYTE_SIZE: usize = 100 * MEGABYTE;
 
 const MIME_TYPES: &[&str] = &["text/plain", "image/png", "image/jpg"];
 
-struct SharedState {
+struct SharedStateInner {
     next_item_id: AtomicU64,
     items: Mutex<Vec<HistoryItem>>,
-    notify_write_send: PipeWriter,
 
     data_control_manager: OnceLock<ExtDataControlManagerV1>,
     data_control_devices: Mutex<HashMap</*seat global name */ u32, ExtDataControlDeviceV1>>,
-    qh: QueueHandle<WlState>,
+    qh: QueueHandle<SharedState>,
+}
+
+struct SharedState {
+    inner: Arc<SharedStateInner>,
+    /// wl_seat that arrived before the data control manager so we weren't able to grab their device immediatly.
+    deferred_seats: Vec<WlSeat>,
 }
 
 struct InProgressOffer {
@@ -58,14 +66,7 @@ struct InProgressOffer {
     time: Duration,
 }
 
-struct WlState {
-    shared_state: Arc<SharedState>,
-
-    /// wl_seat that arrived before the data control manager so we weren't able to grab their device immediatly.
-    deferred_seats: Vec<WlSeat>,
-}
-
-impl Dispatch<WlRegistry, ()> for WlState {
+impl Dispatch<WlRegistry, ()> for SharedState {
     fn event(
         state: &mut Self,
         proxy: &WlRegistry,
@@ -84,14 +85,14 @@ impl Dispatch<WlRegistry, ()> for WlState {
                     info!("A new seat was connected");
                     let seat: WlSeat = proxy.bind(name, 1, qhandle, ());
 
-                    match state.shared_state.data_control_manager.get() {
+                    match state.inner.data_control_manager.get() {
                         None => {
                             state.deferred_seats.push(seat);
                         }
                         Some(manager) => {
                             let device = manager.get_data_device(&seat, qhandle, ());
                             state
-                                .shared_state
+                                .inner
                                 .data_control_devices
                                 .lock()
                                 .unwrap()
@@ -104,7 +105,7 @@ impl Dispatch<WlRegistry, ()> for WlState {
                     for seat in state.deferred_seats.drain(..) {
                         let device = manager.get_data_device(&seat, qhandle, ());
                         state
-                            .shared_state
+                            .inner
                             .data_control_devices
                             .lock()
                             .unwrap()
@@ -112,7 +113,7 @@ impl Dispatch<WlRegistry, ()> for WlState {
                     }
 
                     state
-                        .shared_state
+                        .inner
                         .data_control_manager
                         .set(manager)
                         .expect("ext_data_control_manager_v1 already set, global appeared twice?");
@@ -121,7 +122,7 @@ impl Dispatch<WlRegistry, ()> for WlState {
             wayland_client::protocol::wl_registry::Event::GlobalRemove { name } => {
                 // try to remove, if it's not a wl_seat it may not exist
                 state
-                    .shared_state
+                    .inner
                     .data_control_devices
                     .lock()
                     .unwrap()
@@ -131,7 +132,7 @@ impl Dispatch<WlRegistry, ()> for WlState {
         }
     }
 }
-impl Dispatch<ExtDataControlManagerV1, ()> for WlState {
+impl Dispatch<ExtDataControlManagerV1, ()> for SharedState {
     fn event(
         _state: &mut Self,
         _proxy: &ExtDataControlManagerV1,
@@ -143,7 +144,7 @@ impl Dispatch<ExtDataControlManagerV1, ()> for WlState {
         // no events at the time of writing
     }
 }
-impl Dispatch<WlSeat, ()> for WlState {
+impl Dispatch<WlSeat, ()> for SharedState {
     fn event(
         _state: &mut Self,
         _proxy: &WlSeat,
@@ -155,7 +156,7 @@ impl Dispatch<WlSeat, ()> for WlState {
         // we don't care about anything about the seat
     }
 }
-impl Dispatch<ExtDataControlDeviceV1, ()> for WlState {
+impl Dispatch<ExtDataControlDeviceV1, ()> for SharedState {
     fn event(
         state: &mut Self,
         _proxy: &ExtDataControlDeviceV1,
@@ -191,7 +192,7 @@ impl Dispatch<ExtDataControlDeviceV1, ()> for WlState {
                     };
                     drop(mime_types);
 
-                    let history_state = state.shared_state.clone();
+                    let history_state = state.inner.clone();
                     let time = offer_data.time;
 
                     let (reader, writer) = std::io::pipe().unwrap();
@@ -240,7 +241,7 @@ impl Dispatch<ExtDataControlDeviceV1, ()> for WlState {
         }
     }
 
-    event_created_child!(WlState, ExtDataControlDeviceV1, [
+    event_created_child!(SharedState, ExtDataControlDeviceV1, [
         EVT_DATA_OFFER_OPCODE => (ExtDataControlOfferV1, InProgressOffer {
             mime_types: Default::default(),
             time: SystemTime::now()
@@ -250,7 +251,7 @@ impl Dispatch<ExtDataControlDeviceV1, ()> for WlState {
     ]);
 }
 
-impl Dispatch<ExtDataControlOfferV1, InProgressOffer> for WlState {
+impl Dispatch<ExtDataControlOfferV1, InProgressOffer> for SharedState {
     fn event(
         _state: &mut Self,
         _proxy: &ExtDataControlOfferV1,
@@ -268,7 +269,7 @@ impl Dispatch<ExtDataControlOfferV1, InProgressOffer> for WlState {
     }
 }
 
-impl Dispatch<ExtDataControlSourceV1, OfferData> for WlState {
+impl Dispatch<ExtDataControlSourceV1, OfferData> for SharedState {
     fn event(
         _state: &mut Self,
         proxy: &ExtDataControlSourceV1,
@@ -302,15 +303,9 @@ impl Dispatch<ExtDataControlSourceV1, OfferData> for WlState {
     }
 }
 
-impl SharedState {
-    fn notify_wayland_request(&self) {
-        let _ = (&self.notify_write_send).write_all(&[0]);
-    }
-}
-
 fn do_copy_into_clipboard(
     entry: HistoryItem,
-    shared_state: &SharedState,
+    shared_state: &SharedStateInner,
 ) -> Result<(), eyre::Error> {
     for device in &*shared_state.data_control_devices.lock().unwrap() {
         let data_source = shared_state
@@ -341,34 +336,8 @@ fn do_copy_into_clipboard(
     Ok(())
 }
 
-fn dispatch_wayland(
-    mut queue: EventQueue<WlState>,
-    mut wl_state: WlState,
-    notify_write_recv: PipeReader,
-) -> eyre::Result<()> {
-    loop {
-        queue
-            .dispatch_pending(&mut wl_state)
-            .wrap_err("dispatching Wayland events")?;
-
-        let read_guard = queue
-            .prepare_read()
-            .wrap_err("preparing read from Wayland socket")?;
-        let _ = queue.flush();
-
-        let pollfd1_read = PollFd::from_borrowed_fd(read_guard.connection_fd(), PollFlags::IN);
-        let pollfd_signal = PollFd::from_borrowed_fd(notify_write_recv.as_fd(), PollFlags::IN);
-
-        let _ = rustix::event::poll(&mut [pollfd1_read, pollfd_signal], None);
-
-        read_guard
-            .read_without_dispatch()
-            .wrap_err("reading from wayland socket")?;
-    }
-}
-
 #[tracing::instrument(skip(peer, shared_state))]
-fn handle_peer(mut peer: UnixStream, shared_state: &SharedState) -> eyre::Result<()> {
+fn handle_peer(mut peer: UnixStream, shared_state: &SharedStateInner) -> eyre::Result<()> {
     let mut request = [0; 1];
     let Ok(()) = peer.read_exact(&mut request) else {
         return Ok(());
@@ -396,7 +365,7 @@ struct OfferData(Arc<[u8]>);
 
 fn handle_copy_message(
     mut peer: UnixStream,
-    shared_state: &SharedState,
+    shared_state: &SharedStateInner,
 ) -> Result<(), eyre::Error> {
     let mut id = [0; 8];
     peer.read_exact(&mut id).wrap_err("failed to read id")?;
@@ -412,30 +381,26 @@ fn handle_copy_message(
 
     do_copy_into_clipboard(item, &shared_state).wrap_err("doing copy")?;
 
-    shared_state.notify_wayland_request();
-
     Ok(())
 }
 
-fn handle_clear_message(shared_state: &SharedState) -> eyre::Result<()> {
+fn handle_clear_message(shared_state: &SharedStateInner) -> eyre::Result<()> {
     shared_state.items.lock().unwrap().clear();
 
     for device in &*shared_state.data_control_devices.lock().unwrap() {
         device.1.set_selection(None);
     }
 
-    shared_state.notify_wayland_request();
-
     Ok(())
 }
 
 fn read_fd_into_history(
-    history_state: &SharedState,
+    history_state: &SharedStateInner,
     time: std::time::Duration,
     mime: String,
     data_reader: impl Read,
 ) -> Result<(), eyre::Error> {
-    let mut data_reader = BufReader::new(data_reader).take(MAX_ENTRY_SIZE);
+    let mut data_reader = BufReader::new(data_reader).take(MAX_ENTRY_SIZE as u64);
     let mut data = Vec::new();
     data_reader
         .read_to_end(&mut data)
@@ -465,31 +430,34 @@ fn read_fd_into_history(
         running_total += item.data.len() + std::mem::size_of::<HistoryItem>();
         if running_total > crate::MAX_HISTORY_BYTE_SIZE {
             cutoff = Some(idx);
+            break;
         }
     }
     if let Some(cutoff) = cutoff {
         info!(
-            "Dropping old {} items because limit of {} bytes was reached for the history",
+            "Dropping old {} items beca{} bytes was reached for the history",
             cutoff + 1,
-            crate::MAX_HISTORY_BYTE_SIZE
+            format_history_size(crate::MAX_HISTORY_BYTE_SIZE)
         );
         items.splice(0..=cutoff, []);
     }
     info!(
-        "Successfully stored clipboard value of mime type {mime} (new history size {running_total})"
+        "Successfully stored clipboard value of mime type {mime} (new history size {})",
+        format_history_size(running_total)
     );
     Ok(())
 }
 
+fn format_history_size(size: usize) -> impl Display {
+    if size < MEGABYTE {
+        return format!("{size}B");
+    }
+    format!("{}MB", size / MEGABYTE)
+}
+
 fn main() -> eyre::Result<()> {
     let socket_path = clippyboard_shared::socket_path()?;
 
-    let socket_path2 = socket_path.clone();
-    let _ = ctrlc::set_handler(move || {
-        cleanup(&socket_path2);
-        std::process::exit(130); // sigint
-    });
-
     let Err(err) = main_inner(&socket_path);
 
     if let Some(ioerr) = err.downcast_ref::<io::Error>()
@@ -514,35 +482,27 @@ pub fn main_inner(socket_path: &PathBuf) -> eyre::Result<Infallible> {
     let conn =
         wayland_client::Connection::connect_to_env().wrap_err("connecting to the compositor")?;
 
-    let mut queue = conn.new_event_queue::<WlState>();
+    let mut queue = conn.new_event_queue::<SharedState>();
 
-    let (notify_write_recv, notify_write_send) = std::io::pipe().expect("todo");
+    let mut shared_state = SharedState {
+        inner: Arc::new(SharedStateInner {
+            next_item_id: AtomicU64::new(0),
+            items: Mutex::new(Vec::<HistoryItem>::new()),
 
-    let shared_state = Arc::new(SharedState {
-        next_item_id: AtomicU64::new(0),
-        items: Mutex::new(Vec::<HistoryItem>::new()),
-        notify_write_send,
-
-        data_control_manager: OnceLock::new(),
-        data_control_devices: Mutex::new(HashMap::new()),
-        qh: queue.handle(),
-    });
-
-    let history_state2 = shared_state.clone();
-
-    let mut wl_state = WlState {
+            data_control_manager: OnceLock::new(),
+            data_control_devices: Mutex::new(HashMap::new()),
+            qh: queue.handle(),
+        }),
         deferred_seats: Vec::new(),
-
-        shared_state: history_state2,
     };
 
     conn.display().get_registry(&queue.handle(), ());
 
     queue
-        .roundtrip(&mut wl_state)
+        .roundtrip(&mut shared_state)
         .wrap_err("failed to set up wayland state")?;
 
-    if wl_state.shared_state.data_control_manager.get().is_none() {
+    if shared_state.inner.data_control_manager.get().is_none() {
         bail!(
             "{} not found, the ext-data-control-v1 Wayland extension is likely unsupported by your compositor.\n\
             check https://wayland.app/protocols/ext-data-control-v1#compositor-support\
@@ -551,35 +511,65 @@ pub fn main_inner(socket_path: &PathBuf) -> eyre::Result<Infallible> {
         );
     }
 
-    rustix::fs::fcntl_setfl(notify_write_recv.as_fd(), OFlags::NONBLOCK).expect("todo");
-    rustix::fs::fcntl_setfl(conn.as_fd(), OFlags::NONBLOCK).expect("TODO");
+    let mut event_loop =
+        EventLoop::<SharedState>::try_new().wrap_err("failed to initialize event_loop")?;
 
-    let socket_path_clone = socket_path.to_owned();
-    std::thread::spawn(move || {
-        if let Err(err) = dispatch_wayland(queue, wl_state, notify_write_recv) {
-            error!("error on Wayland thread: {err:?}");
-            cleanup(&socket_path_clone);
-            std::process::exit(1);
-        }
-    });
+    WaylandSource::new(conn.clone(), queue)
+        .insert(event_loop.handle())
+        .unwrap();
+
+    event_loop
+        .handle()
+        .insert_source(
+            Signals::new(&[Signal::SIGINT, Signal::SIGTERM])
+                .wrap_err("failed to create signal listener")?,
+            |_, _, _| {
+                cleanup(socket_path);
+                std::process::exit(0);
+            },
+        )
+        .wrap_err("failed to register signalfd handlers")?;
+
+    event_loop
+        .handle()
+        .insert_source(
+            Generic::new(socket, Interest::READ, Mode::Level),
+            |_, socket, shared_state| {
+                let peer = socket.accept();
+                match peer {
+                    Ok((peer, _)) => {
+                        let history_state = shared_state.inner.clone();
+                        std::thread::spawn(move || {
+                            let result = handle_peer(peer, &history_state);
+                            if let Err(err) = result {
+                                warn!("Error handling peer: {err:?}");
+                            }
+                        });
+                    }
+                    Err(err) => {
+                        warn!("Error accepting peer: {err}");
+                    }
+                }
+
+                Ok(calloop::PostAction::Continue)
+            },
+        )
+        .wrap_err("failed to register socket event source")?;
 
     info!("Listening on {}", socket_path.display());
 
-    for peer in socket.incoming() {
-        match peer {
-            Ok(peer) => {
-                let history_state = shared_state.clone();
-                std::thread::spawn(move || {
-                    let result = handle_peer(peer, &history_state);
-                    if let Err(err) = result {
-                        warn!("Error handling peer: {err:?}");
-                    }
-                });
-            }
-            Err(err) => {
-                warn!("Error accepting peer: {err}");
-            }
-        }
+    let socket_path_clone = socket_path.clone();
+
+    let result = event_loop.run(
+        std::time::Duration::from_millis(100),
+        &mut shared_state,
+        |_| {},
+    );
+
+    if let Err(err) = result {
+        error!("error on Wayland thread: {err:?}");
+        cleanup(&socket_path_clone);
+        std::process::exit(1);
     }
 
     unreachable!("socket.incoming will never return None")