pixelflux/wayland/
dcclient.rs

1//! Data-control CLIENT: the native clipboard bridge to a NESTED app compositor
2//! (labwc/kwin running under pixelflux).
3//!
4//! Apps in a nested session use the inner compositor's selection, which
5//! pixelflux's own clipboard machinery never sees. selkies bridges it over the
6//! ScreenCapture ABI backed by this module instead of forking wl-copy/wl-paste
7//! per operation: one-shot [`list_types`]/[`read`], a [`write`] that returns with
8//! the selection taken (a detached thread then serves paste requests until
9//! another client takes the selection), and [`watch`] (a thread reporting
10//! selection changes to a Python callback). The compositor is spoken to through
11//! `ext_data_control_manager_v1` (KWin since Plasma 6.3, wlroots since 0.19) or
12//! `zwlr_data_control_manager_v1` (earlier wlroots and KWin), preferring the
13//! standardized ext form when both are advertised. Every compositor round-trip
14//! is deadline-bounded via [`wlclient::bounded_roundtrip`].
15
16use std::collections::HashMap;
17use std::os::fd::AsFd;
18use std::os::unix::net::UnixStream;
19use std::sync::atomic::{AtomicBool, Ordering};
20use std::sync::{Arc, Mutex};
21use std::time::Duration;
22
23use pyo3::{Py, PyAny, Python};
24use wayland_client::backend::ObjectId;
25use wayland_client::protocol::{wl_registry, wl_seat};
26use wayland_client::{delegate_noop, Connection, Dispatch, Proxy, QueueHandle};
27use wayland_protocols::ext::data_control::v1::client::{
28    ext_data_control_device_v1::{self, ExtDataControlDeviceV1},
29    ext_data_control_manager_v1::ExtDataControlManagerV1,
30    ext_data_control_offer_v1::{self, ExtDataControlOfferV1},
31    ext_data_control_source_v1::{self, ExtDataControlSourceV1},
32};
33use wayland_protocols_wlr::data_control::v1::client::{
34    zwlr_data_control_device_v1::{self, ZwlrDataControlDeviceV1},
35    zwlr_data_control_manager_v1::ZwlrDataControlManagerV1,
36    zwlr_data_control_offer_v1::{self, ZwlrDataControlOfferV1},
37    zwlr_data_control_source_v1::{self, ZwlrDataControlSourceV1},
38};
39
40use crate::wayland::wlclient::{
41    bounded_roundtrip, impl_sync_callback, pipe_cloexec, read_fd_to_end, wait_readable,
42    write_fd_all, SyncState, IO_TIMEOUT,
43};
44
45/// How often a background thread wakes from its socket poll to check its stop
46/// flag, bounding unwatch/shutdown latency.
47const STOP_POLL: Duration = Duration::from_millis(500);
48
49/// The two data-control protocol families behind one shape: both define the
50/// same manager/device/source/offer object graph with identical semantics, so
51/// each wrapper dispatches the one request set to whichever family the
52/// compositor advertised. Objects never mix families: everything descends from
53/// the single manager chosen at bind time.
54enum DcManager {
55    Ext(ExtDataControlManagerV1),
56    Wlr(ZwlrDataControlManagerV1),
57}
58
59enum DcDevice {
60    Ext(ExtDataControlDeviceV1),
61    Wlr(ZwlrDataControlDeviceV1),
62}
63
64enum DcSource {
65    Ext(ExtDataControlSourceV1),
66    Wlr(ZwlrDataControlSourceV1),
67}
68
69#[derive(Clone)]
70enum DcOffer {
71    Ext(ExtDataControlOfferV1),
72    Wlr(ZwlrDataControlOfferV1),
73}
74
75impl DcManager {
76    fn get_data_device(&self, seat: &wl_seat::WlSeat, qh: &QueueHandle<DcState>) -> DcDevice {
77        match self {
78            DcManager::Ext(m) => DcDevice::Ext(m.get_data_device(seat, qh, ())),
79            DcManager::Wlr(m) => DcDevice::Wlr(m.get_data_device(seat, qh, ())),
80        }
81    }
82
83    fn create_data_source(&self, qh: &QueueHandle<DcState>) -> DcSource {
84        match self {
85            DcManager::Ext(m) => DcSource::Ext(m.create_data_source(qh, ())),
86            DcManager::Wlr(m) => DcSource::Wlr(m.create_data_source(qh, ())),
87        }
88    }
89}
90
91impl DcDevice {
92    fn set_selection(&self, source: Option<&DcSource>) {
93        match self {
94            DcDevice::Ext(d) => d.set_selection(source.map(|s| match s {
95                DcSource::Ext(s) => s,
96                DcSource::Wlr(_) => unreachable!("mixed data-control families"),
97            })),
98            DcDevice::Wlr(d) => d.set_selection(source.map(|s| match s {
99                DcSource::Wlr(s) => s,
100                DcSource::Ext(_) => unreachable!("mixed data-control families"),
101            })),
102        }
103    }
104
105    fn destroy(&self) {
106        match self {
107            DcDevice::Ext(d) => d.destroy(),
108            DcDevice::Wlr(d) => d.destroy(),
109        }
110    }
111}
112
113impl DcSource {
114    fn offer(&self, mime: String) {
115        match self {
116            DcSource::Ext(s) => s.offer(mime),
117            DcSource::Wlr(s) => s.offer(mime),
118        }
119    }
120
121    fn destroy(&self) {
122        match self {
123            DcSource::Ext(s) => s.destroy(),
124            DcSource::Wlr(s) => s.destroy(),
125        }
126    }
127}
128
129impl DcOffer {
130    fn id(&self) -> ObjectId {
131        match self {
132            DcOffer::Ext(o) => o.id(),
133            DcOffer::Wlr(o) => o.id(),
134        }
135    }
136
137    fn receive(&self, mime: String, fd: std::os::fd::BorrowedFd<'_>) {
138        match self {
139            DcOffer::Ext(o) => o.receive(mime, fd),
140            DcOffer::Wlr(o) => o.receive(mime, fd),
141        }
142    }
143
144    fn destroy(&self) {
145        match self {
146            DcOffer::Ext(o) => o.destroy(),
147            DcOffer::Wlr(o) => o.destroy(),
148        }
149    }
150}
151
152#[derive(Default)]
153struct DcState {
154    seat: Option<wl_seat::WlSeat>,
155    manager_ext: Option<ExtDataControlManagerV1>,
156    manager_wlr: Option<ZwlrDataControlManagerV1>,
157    /// Advertised mimes per live offer.
158    offer_mimes: HashMap<ObjectId, Vec<String>>,
159    selection: Option<DcOffer>,
160    /// Set on every `selection` event (the watch loop's change edge).
161    selection_changed: bool,
162    /// Compositor told this device it is done (seat gone).
163    finished: bool,
164    /// The write path's source lost the selection to another client.
165    cancelled: bool,
166    /// Mime -> bytes served by the write path's source.
167    serve: Vec<(String, Vec<u8>)>,
168    sync_done: bool,
169}
170
171impl SyncState for DcState {
172    fn sync_done_mut(&mut self) -> &mut bool {
173        &mut self.sync_done
174    }
175}
176impl_sync_callback!(DcState);
177
178impl DcState {
179    /// The bound manager, preferring ext over wlr; the loser is released.
180    fn take_manager(&mut self) -> Result<DcManager, String> {
181        if let Some(m) = self.manager_ext.take() {
182            if let Some(w) = self.manager_wlr.take() {
183                w.destroy();
184            }
185            return Ok(DcManager::Ext(m));
186        }
187        self.manager_wlr.take().map(DcManager::Wlr).ok_or_else(|| {
188            "app compositor advertises neither ext_data_control_manager_v1 \
189             nor zwlr_data_control_manager_v1"
190                .to_string()
191        })
192    }
193
194    fn on_data_offer(&mut self, id: ObjectId) {
195        self.offer_mimes.entry(id).or_default();
196    }
197
198    fn on_offer_mime(&mut self, id: ObjectId, mime_type: String) {
199        self.offer_mimes.entry(id).or_default().push(mime_type);
200    }
201
202    fn on_selection(&mut self, offer: Option<DcOffer>) {
203        // Replaced offers are dead objects; drop their proxy and mimes so a
204        // long-lived watch connection doesn't accumulate them.
205        if let Some(old) = self.selection.take()
206            && offer.as_ref().map(|o| o.id()) != Some(old.id()) {
207                self.offer_mimes.remove(&old.id());
208                old.destroy();
209            }
210        self.selection = offer;
211        self.selection_changed = true;
212    }
213
214    /// This bridge carries the regular selection only, but ext v1 still
215    /// introduces an offer per primary-selection change; release it (unless the
216    /// compositor reused the regular selection's object) so it can't pile up.
217    fn on_primary_selection(&mut self, offer: Option<DcOffer>) {
218        if let Some(o) = offer
219            && self.selection.as_ref().map(|s| s.id()) != Some(o.id()) {
220                self.offer_mimes.remove(&o.id());
221                o.destroy();
222            }
223    }
224
225    fn on_send(&mut self, mime_type: &str, fd: std::os::fd::OwnedFd) {
226        if let Some((_, data)) = self.serve.iter().find(|(m, _)| m == mime_type) {
227            let _ = write_fd_all(&fd, data, IO_TIMEOUT);
228        }
229        // fd drops here, closing the pipe so the paster sees EOF.
230    }
231}
232
233impl Dispatch<wl_registry::WlRegistry, ()> for DcState {
234    fn event(
235        state: &mut Self,
236        registry: &wl_registry::WlRegistry,
237        event: wl_registry::Event,
238        _: &(),
239        _: &Connection,
240        qh: &QueueHandle<Self>,
241    ) {
242        if let wl_registry::Event::Global { name, interface, .. } = event {
243            // Version 1 of each suffices: the seat is only an argument, and v1
244            // data-control carries the regular selection this bridge needs.
245            match interface.as_str() {
246                "wl_seat" if state.seat.is_none() => {
247                    state.seat = Some(registry.bind(name, 1, qh, ()));
248                }
249                "ext_data_control_manager_v1" if state.manager_ext.is_none() => {
250                    state.manager_ext = Some(registry.bind(name, 1, qh, ()));
251                }
252                "zwlr_data_control_manager_v1" if state.manager_wlr.is_none() => {
253                    state.manager_wlr = Some(registry.bind(name, 1, qh, ()));
254                }
255                _ => {}
256            }
257        }
258    }
259}
260
261delegate_noop!(DcState: ignore wl_seat::WlSeat);
262delegate_noop!(DcState: ExtDataControlManagerV1);
263delegate_noop!(DcState: ZwlrDataControlManagerV1);
264
265impl Dispatch<ExtDataControlOfferV1, ()> for DcState {
266    fn event(
267        state: &mut Self,
268        offer: &ExtDataControlOfferV1,
269        event: ext_data_control_offer_v1::Event,
270        _: &(),
271        _: &Connection,
272        _: &QueueHandle<Self>,
273    ) {
274        if let ext_data_control_offer_v1::Event::Offer { mime_type } = event {
275            state.on_offer_mime(offer.id(), mime_type);
276        }
277    }
278}
279
280impl Dispatch<ZwlrDataControlOfferV1, ()> for DcState {
281    fn event(
282        state: &mut Self,
283        offer: &ZwlrDataControlOfferV1,
284        event: zwlr_data_control_offer_v1::Event,
285        _: &(),
286        _: &Connection,
287        _: &QueueHandle<Self>,
288    ) {
289        if let zwlr_data_control_offer_v1::Event::Offer { mime_type } = event {
290            state.on_offer_mime(offer.id(), mime_type);
291        }
292    }
293}
294
295impl Dispatch<ExtDataControlDeviceV1, ()> for DcState {
296    fn event(
297        state: &mut Self,
298        _: &ExtDataControlDeviceV1,
299        event: ext_data_control_device_v1::Event,
300        _: &(),
301        _: &Connection,
302        _: &QueueHandle<Self>,
303    ) {
304        match event {
305            ext_data_control_device_v1::Event::DataOffer { id } => {
306                state.on_data_offer(id.id());
307            }
308            ext_data_control_device_v1::Event::Selection { id } => {
309                state.on_selection(id.map(DcOffer::Ext));
310            }
311            ext_data_control_device_v1::Event::PrimarySelection { id } => {
312                state.on_primary_selection(id.map(DcOffer::Ext));
313            }
314            ext_data_control_device_v1::Event::Finished => {
315                state.finished = true;
316            }
317            _ => {}
318        }
319    }
320
321    wayland_client::event_created_child!(DcState, ExtDataControlDeviceV1, [
322        ext_data_control_device_v1::EVT_DATA_OFFER_OPCODE => (ExtDataControlOfferV1, ()),
323    ]);
324}
325
326impl Dispatch<ZwlrDataControlDeviceV1, ()> for DcState {
327    fn event(
328        state: &mut Self,
329        _: &ZwlrDataControlDeviceV1,
330        event: zwlr_data_control_device_v1::Event,
331        _: &(),
332        _: &Connection,
333        _: &QueueHandle<Self>,
334    ) {
335        match event {
336            zwlr_data_control_device_v1::Event::DataOffer { id } => {
337                state.on_data_offer(id.id());
338            }
339            zwlr_data_control_device_v1::Event::Selection { id } => {
340                state.on_selection(id.map(DcOffer::Wlr));
341            }
342            zwlr_data_control_device_v1::Event::Finished => {
343                state.finished = true;
344            }
345            _ => {}
346        }
347    }
348
349    wayland_client::event_created_child!(DcState, ZwlrDataControlDeviceV1, [
350        zwlr_data_control_device_v1::EVT_DATA_OFFER_OPCODE => (ZwlrDataControlOfferV1, ()),
351    ]);
352}
353
354impl Dispatch<ExtDataControlSourceV1, ()> for DcState {
355    fn event(
356        state: &mut Self,
357        _: &ExtDataControlSourceV1,
358        event: ext_data_control_source_v1::Event,
359        _: &(),
360        _: &Connection,
361        _: &QueueHandle<Self>,
362    ) {
363        match event {
364            ext_data_control_source_v1::Event::Send { mime_type, fd } => {
365                state.on_send(&mime_type, fd);
366            }
367            ext_data_control_source_v1::Event::Cancelled => {
368                state.cancelled = true;
369            }
370            _ => {}
371        }
372    }
373}
374
375impl Dispatch<ZwlrDataControlSourceV1, ()> for DcState {
376    fn event(
377        state: &mut Self,
378        _: &ZwlrDataControlSourceV1,
379        event: zwlr_data_control_source_v1::Event,
380        _: &(),
381        _: &Connection,
382        _: &QueueHandle<Self>,
383    ) {
384        match event {
385            zwlr_data_control_source_v1::Event::Send { mime_type, fd } => {
386                state.on_send(&mime_type, fd);
387            }
388            zwlr_data_control_source_v1::Event::Cancelled => {
389                state.cancelled = true;
390            }
391            _ => {}
392        }
393    }
394}
395
396/// Connect to `socket_path` and return (connection, queue, state, device) with
397/// the current selection already delivered.
398fn open_device(
399    socket_path: &str,
400) -> Result<(Connection, wayland_client::EventQueue<DcState>, DcState, DcDevice), String> {
401    let stream =
402        UnixStream::connect(socket_path).map_err(|e| format!("connect {socket_path}: {e}"))?;
403    let conn = Connection::from_socket(stream).map_err(|e| format!("wayland setup: {e}"))?;
404    let mut queue = conn.new_event_queue();
405    let qh = queue.handle();
406    let _registry = conn.display().get_registry(&qh, ());
407    let mut state = DcState::default();
408    bounded_roundtrip(&conn, &mut queue, &mut state)?;
409    let seat = state.seat.clone().ok_or("app compositor advertises no wl_seat")?;
410    let manager = state.take_manager()?;
411    let device = manager.get_data_device(&seat, &qh);
412    bounded_roundtrip(&conn, &mut queue, &mut state)?;
413    Ok((conn, queue, state, device))
414}
415
416/// Mimes offered by the current selection (empty when nothing is copied).
417pub(crate) fn list_types(socket_path: &str) -> Result<Vec<String>, String> {
418    let (_conn, _queue, state, device) = open_device(socket_path)?;
419    let out = state
420        .selection
421        .as_ref()
422        .and_then(|o| state.offer_mimes.get(&o.id()).cloned())
423        .unwrap_or_default();
424    device.destroy();
425    Ok(out)
426}
427
428/// The current selection's payload for `mime`, or None when there is no
429/// selection or it does not offer that mime.
430pub(crate) fn read(socket_path: &str, mime: &str) -> Result<Option<Vec<u8>>, String> {
431    let (conn, mut queue, mut state, device) = open_device(socket_path)?;
432    let Some(offer) = state.selection.clone() else {
433        device.destroy();
434        return Ok(None);
435    };
436    let offered = state.offer_mimes.get(&offer.id()).is_some_and(|m| m.iter().any(|x| x == mime));
437    if !offered {
438        device.destroy();
439        return Ok(None);
440    }
441    let (rd, wr) = pipe_cloexec()?;
442    offer.receive(mime.to_string(), wr.as_fd());
443    queue.flush().map_err(|e| format!("flush: {e}"))?;
444    drop(wr);
445    // The source app writes into the pipe as it pleases; dispatch is not needed
446    // for the bytes, only the fd read.
447    let data = read_fd_to_end(&rd, IO_TIMEOUT)?;
448    let _ = bounded_roundtrip(&conn, &mut queue, &mut state);
449    device.destroy();
450    Ok(Some(data))
451}
452
453/// Take the selection, serving `entries` (mime, bytes) to every paster from a
454/// detached thread until another client takes the selection (or the compositor
455/// goes away). The selection is compositor-acknowledged when this returns, so a
456/// caller may immediately trigger a paste against it. Replacing a previous
457/// write is implicit: the compositor cancels the old source when the new one
458/// takes the selection.
459pub(crate) fn write(socket_path: &str, entries: Vec<(String, Vec<u8>)>) -> Result<(), String> {
460    let stream =
461        UnixStream::connect(socket_path).map_err(|e| format!("connect {socket_path}: {e}"))?;
462    let conn = Connection::from_socket(stream).map_err(|e| format!("wayland setup: {e}"))?;
463    let mut queue = conn.new_event_queue();
464    let qh = queue.handle();
465    let _registry = conn.display().get_registry(&qh, ());
466    let mut state = DcState::default();
467    bounded_roundtrip(&conn, &mut queue, &mut state)?;
468    let seat = state.seat.clone().ok_or("app compositor advertises no wl_seat")?;
469    let manager = state.take_manager()?;
470    let device = manager.get_data_device(&seat, &qh);
471    let source = manager.create_data_source(&qh);
472    for (mime, _) in &entries {
473        source.offer(mime.clone());
474    }
475    state.serve = entries;
476    device.set_selection(Some(&source));
477    bounded_roundtrip(&conn, &mut queue, &mut state)?;
478    std::thread::Builder::new()
479        .name("pf-dc-selection".into())
480        .spawn(move || {
481            if let Err(e) = serve_selection(conn, queue, state, device, source) {
482                eprintln!("[Clipboard] app-compositor selection serve ended: {e}");
483            }
484        })
485        .map_err(|e| format!("spawn: {e}"))?;
486    Ok(())
487}
488
489fn serve_selection(
490    conn: Connection,
491    mut queue: wayland_client::EventQueue<DcState>,
492    mut state: DcState,
493    device: DcDevice,
494    source: DcSource,
495) -> Result<(), String> {
496    while !state.cancelled && !state.finished {
497        // Sends are served inside dispatch; block until the compositor has
498        // something (with a poll so a dead compositor can't pin the thread).
499        queue.flush().map_err(|e| format!("flush: {e}"))?;
500        if let Some(guard) = conn.prepare_read() {
501            use std::os::fd::AsRawFd;
502            if wait_readable(guard.connection_fd().as_raw_fd(), STOP_POLL)? {
503                guard.read().map_err(|e| format!("read: {e}"))?;
504            }
505        }
506        queue.dispatch_pending(&mut state).map_err(|e| format!("dispatch: {e}"))?;
507    }
508    source.destroy();
509    device.destroy();
510    let _ = queue.flush();
511    Ok(())
512}
513
514/// Drop the selection (the compositor also cancels whatever source held it).
515pub(crate) fn clear(socket_path: &str) -> Result<(), String> {
516    let (conn, mut queue, mut state, device) = open_device(socket_path)?;
517    device.set_selection(None);
518    bounded_roundtrip(&conn, &mut queue, &mut state)?;
519    device.destroy();
520    Ok(())
521}
522
523struct WatchHandle {
524    stop: Arc<AtomicBool>,
525}
526
527static WATCHERS: Mutex<Option<HashMap<String, WatchHandle>>> = Mutex::new(None);
528
529/// Report every selection change on `socket_path` (including the one current at
530/// start) to `callback(mimes: list[str])` from a background thread. A second
531/// watch on the same socket replaces the first.
532pub(crate) fn watch(socket_path: &str, callback: Py<PyAny>) -> Result<(), String> {
533    let stop = Arc::new(AtomicBool::new(false));
534    {
535        let mut reg = WATCHERS.lock().unwrap();
536        let map = reg.get_or_insert_with(HashMap::new);
537        if let Some(old) = map.insert(socket_path.to_string(), WatchHandle { stop: stop.clone() })
538        {
539            old.stop.store(true, Ordering::Relaxed);
540        }
541    }
542    let path = socket_path.to_string();
543    std::thread::Builder::new()
544        .name("pf-dc-watch".into())
545        .spawn(move || {
546            if let Err(e) = watch_loop(&path, callback, &stop) {
547                eprintln!("[Clipboard] app-compositor watch ended: {e}");
548            }
549        })
550        .map_err(|e| format!("spawn: {e}"))?;
551    Ok(())
552}
553
554/// Stop every clipboard watch (process teardown sweep; watches are not tied to
555/// captures, so the global stop helper must reach them explicitly).
556pub(crate) fn unwatch_all() {
557    let mut reg = WATCHERS.lock().unwrap();
558    if let Some(map) = reg.as_mut() {
559        for (_, handle) in map.drain() {
560            handle.stop.store(true, Ordering::Relaxed);
561        }
562    }
563}
564
565/// Stop the watch on `socket_path` (no-op when none is running).
566pub(crate) fn unwatch(socket_path: &str) {
567    let mut reg = WATCHERS.lock().unwrap();
568    if let Some(map) = reg.as_mut()
569        && let Some(handle) = map.remove(socket_path) {
570            handle.stop.store(true, Ordering::Relaxed);
571        }
572}
573
574fn watch_loop(socket_path: &str, callback: Py<PyAny>, stop: &AtomicBool) -> Result<(), String> {
575    let (conn, mut queue, mut state, device) = open_device(socket_path)?;
576    while !stop.load(Ordering::Relaxed) && !state.finished {
577        if state.selection_changed {
578            state.selection_changed = false;
579            let mimes = state
580                .selection
581                .as_ref()
582                .and_then(|o| state.offer_mimes.get(&o.id()).cloned())
583                .unwrap_or_default();
584            if !mimes.is_empty() {
585                if crate::PY_SHUTDOWN.load(Ordering::Relaxed) {
586                    break;
587                }
588                Python::attach(|py| {
589                    if let Err(e) = callback.call1(py, (mimes,)) {
590                        e.print(py);
591                    }
592                });
593            }
594        }
595        queue.flush().map_err(|e| format!("flush: {e}"))?;
596        if let Some(guard) = conn.prepare_read() {
597            use std::os::fd::AsRawFd;
598            if wait_readable(guard.connection_fd().as_raw_fd(), STOP_POLL)? {
599                guard.read().map_err(|e| format!("read: {e}"))?;
600            }
601        }
602        queue.dispatch_pending(&mut state).map_err(|e| format!("dispatch: {e}"))?;
603    }
604    device.destroy();
605    let _ = queue.flush();
606    Ok(())
607}
pixelflux/wayland/dcclient.rs

pixelflux/wayland/
dcclient.rs
