tokio-wayland-loop.rs

tokio-wayland-loop.rs

use smithay_client_toolkit::reexports::client::{
    EventQueue, Interface, Main, MessageGroup, Proxy, ProxyMap,
};

/// Tokio task for polling the Wayland socket, dispatching the callbacks.
/// This makes wayland_event_chan work.
pub async fn wayland_loop(event_queue: &mut EventQueue) -> Result<(), Box<dyn std::error::Error>> {
    use std::task::Poll;

    let fd = event_queue.display().get_connection_fd();
    let wl_reg = tokio::io::PollEvented::new(mio::unix::EventedFd(&fd))?;

    loop {
        tokio::future::poll_fn(|cx| {
                eprintln!("pollfn!");
            if let Some(guard) = event_queue.prepare_read() {
                if let Err(e) = event_queue.display().flush() {
                    eprintln!("Error flushing the wayland socket: {:?}", e);
                }

                eprintln!("poll!");
                futures::ready!(wl_reg.poll_read_ready(cx, mio::Ready::readable()))?;

                if let Err(e) = guard.read_events() {
                    if e.kind() == std::io::ErrorKind::WouldBlock {
                        wl_reg.clear_read_ready(cx, mio::Ready::readable())?;
                        eprintln!("pollfn-clr!");
                        return Poll::Pending;
                    } else {
                        eprintln!("Error reading from the wayland socket: {:?}", e);
                    }
                }

                event_queue
                    .dispatch(&mut (), |_, _, _| {})
                    // .dispatch_pending(&mut (), |_, _, _| {}) // did we not read?!
                    ?;

                let res: Result<(), Box<dyn std::error::Error>> = Ok(());
                eprintln!("pollfn-ready!");
                Poll::Ready(res)
            } else {
                event_queue.dispatch_pending(&mut (), |_, _, _| {})?;
                eprintln!("pollfn-pend!");
                Poll::Pending
            }
        })
        .await?;
        tokio::task::yield_now().await; // required for reacting to events on the same thread ??
    }
}

/// Creates a broadcast channel for a Wayland object's events.
/// Returns Sender to allow generating new subscribers.
///
/// Requires a wayland_loop to be running.
pub fn wayland_event_chan<I>(
    obj: &Main<I>,
) -> tokio::sync::broadcast::Sender<std::sync::Arc<I::Event>>
where
    I: Interface + AsRef<Proxy<I>> + From<Proxy<I>> + Sync,
    I::Event: MessageGroup<Map = ProxyMap>,
{
    let (tx, _) = tokio::sync::broadcast::channel(8);
    let tx2 = tx.clone();
    obj.quick_assign(move |_, event, _| {
        if let Ok(_) = tx.send(std::sync::Arc::new(event)) {
        } else {
            eprintln!("Event-to-channel send with no receivers?");
        }
        ()
    });
    tx2
}

/// Creates a oneshot channel for a Wayland object's events, intended for WlCallback.
///
/// Requires a wayland_loop to be running.
pub fn wayland_event_chan_oneshot<I>(
    obj: &Main<I>,
) -> tokio::sync::oneshot::Receiver<I::Event>
where
    I: Interface + AsRef<Proxy<I>> + From<Proxy<I>> + Sync,
    I::Event: MessageGroup<Map = ProxyMap>,
{
    let (tx, rx) = tokio::sync::oneshot::channel();
    // would be great to have a quick_assign with FnOnce
    let txc = std::cell::Cell::new(Some(tx));
    obj.quick_assign(move |_, event, _| {
        if let Ok(_) = txc.take().unwrap().send(event) {
        } else {
            eprintln!("Event-to-oneshot-channel send with no receiver?");
        }
        ()
    });
    rx
}

macro_rules! wayselect {
    ( $( $query:expr => | $binder:ident | $handler:expr )* $( ; raw $rquery:expr => | $rbinder:ident | $rhandler:expr )* ) => {
        tokio::select! {
            $(ev = $query => { match ev {
                Ok($binder) => $handler
                Err(er) => eprintln!("wayselect: {:?}", er)
            } })*
            $($rbinder = $rquery => { $rhandler })*
        }
    }
}

macro_rules! wayselectloop {
    ( $($tokens:tt)* ) => { loop { wayselect! { $($tokens)* } } }
}