21596462741

Committed 02 Feb 2026 03:34PM UTC coverage: 59.173% (-0.2%) from 59.326%

Build # 21596462741

Build Type

push

github

Committed by

lpenz

Commit Message

user_wrapper: ignore errors while sending key thorugh the channel

The error could come in a race when the thread was already being dropped.

Also: no longer drop user in engine, it's not necessary anymore and it
was actually enlarging the window of the race mentioned above.

Run Details

0 of 4 new or added lines in 1 file covered. (0.0%)

11 existing lines in 2 files now uncovered.

458 of 774 relevant lines covered (59.17%)

1.53 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

22.73

/src/user_wrapper.rs

// Copyright (C) 2025 Leandro Lisboa Penz <lpenz@lpenz.org>
// This file is subject to the terms and conditions defined in
// file 'LICENSE', which is part of this source code package.

//! Wrapper for user interaction.
//!
//! For now we just check if the user has typed ENTER, which makes
//! ogle exit after the current run is over.

use crossterm::event::{self, Event};
use crossterm::tty::IsTty;
use pin_project::pin_project;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tokio::time::{Duration, sleep};
use tokio_stream::Stream;
use tracing::instrument;

/// A wrapper for `crossterm` that polls the keyboard and provides the
/// keypress in a tokio stream.
///
/// Also provides a virtual implementation for use in tests.
#[pin_project(project = UserStreamProj)]
#[derive(Debug, Default)]
pub enum UserStream {
    /// A real implementation that reads a line from stdin.
    Real(mpsc::UnboundedReceiver<String>, JoinHandle<()>),
    /// A virtual implementation that doesn't do anything.
    #[default]
    Virtual,
}

impl UserStream {
    pub fn new_real() -> Option<UserStream> {
        let stdin = io::stdin();
        if stdin.is_tty() {
            let (tx, rx) = mpsc::unbounded_channel::<String>();
            let handle = tokio::spawn(async move {
                loop {
                    if matches!(event::poll(Duration::from_secs(0)), Ok(true))
                        && let Ok(Event::Key(key_event)) = event::read()
                    {
                        // If sending the key fails, it's probably because we
                        // are in the process of being dropped, so we can
                        // ignore it:
                        let _ = tx.send(format!("{}", key_event.code));
                    } else {
                        // We tokio-sleep here to provide a cancellation point:
                        sleep(Duration::from_millis(127)).await;
                    }
                }
            });
            Some(UserStream::Real(rx, handle))
        } else {
            None
        }
    }

    pub fn new_virtual() -> UserStream {
        UserStream::Virtual
    }
}

impl Stream for UserStream {
    type Item = String;

    #[instrument(level = "debug", ret, skip(cx))]
    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = self.project();
        match this {
            UserStreamProj::Real(rx, _) => {
                let next = Pin::new(rx).poll_recv(cx);
                match next {
                    Poll::Ready(Some(s)) => Poll::Ready(Some(s)),
                    Poll::Ready(None) => Poll::Ready(None),
                    Poll::Pending => Poll::Pending,
                }
            }
            UserStreamProj::Virtual {} => Poll::Ready(None),
        }
    }
}

1	// Copyright (C) 2025 Leandro Lisboa Penz <lpenz@lpenz.org>
2	// This file is subject to the terms and conditions defined in
3	// file 'LICENSE', which is part of this source code package.
4
5	//! Wrapper for user interaction.
6	//!
7	//! For now we just check if the user has typed ENTER, which makes
8	//! ogle exit after the current run is over.
9
10	use crossterm::event::{self, Event};
11	use crossterm::tty::IsTty;
12	use pin_project::pin_project;
13	use std::pin::Pin;
14	use std::task::{Context, Poll};
15	use tokio::io;
16	use tokio::sync::mpsc;
17	use tokio::task::JoinHandle;
18	use tokio::time::{Duration, sleep};
19	use tokio_stream::Stream;
20	use tracing::instrument;
21
22	/// A wrapper for `crossterm` that polls the keyboard and provides the
23	/// keypress in a tokio stream.
24	///
25	/// Also provides a virtual implementation for use in tests.
26	#[pin_project(project = UserStreamProj)]
27	#[derive(Debug, Default)]
28	pub enum UserStream {
29	/// A real implementation that reads a line from stdin.
30	Real(mpsc::UnboundedReceiver<String>, JoinHandle<()>),
31	/// A virtual implementation that doesn't do anything.
32	#[default]
33	Virtual,
34	}
35
36	impl UserStream {
37	pub fn new_real() -> Option<UserStream> {	×
38	let stdin = io::stdin();	×
39	if stdin.is_tty() {	×
40	let (tx, rx) = mpsc::unbounded_channel::<String>();	×
41	let handle = tokio::spawn(async move {	×
42	loop {
43	if matches!(event::poll(Duration::from_secs(0)), Ok(true))	×
44	&& let Ok(Event::Key(key_event)) = event::read()	×
UNCOV 45	{	×
NEW 46	// If sending the key fails, it's probably because we	×
NEW 47	// are in the process of being dropped, so we can	×
NEW 48	// ignore it:	×
NEW 49	let _ = tx.send(format!("{}", key_event.code));	×
UNCOV 50	} else {	×
51	// We tokio-sleep here to provide a cancellation point:
52	sleep(Duration::from_millis(127)).await;	×
53	}
54	}
55	});
56	Some(UserStream::Real(rx, handle))	×
57	} else {
58	None	×
59	}
60	}	×
61
62	pub fn new_virtual() -> UserStream {	2✔
63	UserStream::Virtual	2✔
64	}	2✔
65	}
66
67	impl Stream for UserStream {
68	type Item = String;
69
70	#[instrument(level = "debug", ret, skip(cx))]
71	fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {	2✔
72	let this = self.project();
73	match this {
74	UserStreamProj::Real(rx, _) => {
75	let next = Pin::new(rx).poll_recv(cx);
76	match next {
77	Poll::Ready(Some(s)) => Poll::Ready(Some(s)),
78	Poll::Ready(None) => Poll::Ready(None),
79	Poll::Pending => Poll::Pending,
80	}
81	}
82	UserStreamProj::Virtual {} => Poll::Ready(None),
83	}
84	}	2✔
85	}

lpenz / ogle / 21596462741

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous