20974679380

Committed 13 Jan 2026 10:20PM UTC coverage: 57.901%. Remained the same

Build # 20974679380

Build Type

push

github

Committed by

lpenz

Commit Message

WIP stdin cancellation

Run Details

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

2 existing lines in 1 file now uncovered.

458 of 791 relevant lines covered (57.9%)

1.52 hits per line

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

15.15

/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 is_terminal::IsTerminal;
use pin_project::pin_project;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::{self, AsyncBufReadExt};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tokio_stream::Stream;
use tracing::instrument;

#[derive(Debug)]
pub struct StdinWrapper {
    rx: mpsc::UnboundedReceiver<String>,
    handle: JoinHandle<()>,
}

impl StdinWrapper {
    pub fn new() -> Self {
        let (tx, rx) = mpsc::unbounded_channel::<String>();
        let handle = tokio::spawn(async move {
            let stdin = io::stdin();
            let mut lines = io::BufReader::new(stdin).lines();
            loop {
                match lines.next_line().await {
                    Ok(Some(line)) => {
                        // Stop if the receiver is gone
                        if tx.send(line).is_err() {
                            break;
                        }
                    }
                    Ok(None) => break, // EOF
                    Err(_) => break,   // I/O error
                }
            }
        });
        Self { rx, handle }
    }
}

impl Stream for StdinWrapper {
    type Item = String;
    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        Pin::new(&mut self.rx).poll_recv(cx)
    }
}

impl Drop for StdinWrapper {
    fn drop(&mut self) {
        eprintln!("aborting!");
        self.handle.abort();
    }
}

/// A wrapper for [`tokio_stream::wrappers::LinesStream`] with an
/// inner [`tokio::io::Stdin`].
///
/// 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(StdinWrapper),
    /// 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_terminal() {
            Some(UserStream::Real(StdinWrapper::new()))
        } 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(stdin_wrapper) => {
                let next = Pin::new(stdin_wrapper).poll_next(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 is_terminal::IsTerminal;
11	use pin_project::pin_project;
12	use std::pin::Pin;
13	use std::task::{Context, Poll};
14	use tokio::io::{self, AsyncBufReadExt};
15	use tokio::sync::mpsc;
16	use tokio::task::JoinHandle;
17	use tokio_stream::Stream;
18	use tracing::instrument;
19
20	#[derive(Debug)]
21	pub struct StdinWrapper {
22	rx: mpsc::UnboundedReceiver<String>,
23	handle: JoinHandle<()>,
24	}
25
26	impl StdinWrapper {
NEW 27	pub fn new() -> Self {	×
NEW 28	let (tx, rx) = mpsc::unbounded_channel::<String>();	×
NEW 29	let handle = tokio::spawn(async move {	×
NEW 30	let stdin = io::stdin();	×
NEW 31	let mut lines = io::BufReader::new(stdin).lines();	×
32	loop {
NEW 33	match lines.next_line().await {	×
NEW 34	Ok(Some(line)) => {	×
35	// Stop if the receiver is gone
NEW 36	if tx.send(line).is_err() {	×
NEW 37	break;	×
NEW 38	}	×
39	}
NEW 40	Ok(None) => break, // EOF	×
NEW 41	Err(_) => break, // I/O error	×
42	}
43	}
NEW 44	});	×
NEW 45	Self { rx, handle }	×
NEW 46	}	×
47	}
48
49	impl Stream for StdinWrapper {
50	type Item = String;
NEW 51	fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {	×
NEW 52	Pin::new(&mut self.rx).poll_recv(cx)	×
NEW 53	}	×
54	}
55
56	impl Drop for StdinWrapper {
NEW 57	fn drop(&mut self) {	×
NEW 58	eprintln!("aborting!");	×
NEW 59	self.handle.abort();	×
NEW 60	}	×
61	}
62
63	/// A wrapper for [`tokio_stream::wrappers::LinesStream`] with an
64	/// inner [`tokio::io::Stdin`].
65	///
66	/// Also provides a virtual implementation for use in tests.
67	#[pin_project(project = UserStreamProj)]
68	#[derive(Debug, Default)]
69	pub enum UserStream {
70	/// A real implementation that reads a line from stdin.
71	Real(StdinWrapper),
72	/// A virtual implementation that doesn't do anything.
73	#[default]
74	Virtual,
75	}
76
77	impl UserStream {
UNCOV 78	pub fn new_real() -> Option<UserStream> {	×
79	let stdin = io::stdin();	×
80	if stdin.is_terminal() {	×
NEW 81	Some(UserStream::Real(StdinWrapper::new()))	×
82	} else {
UNCOV 83	None	×
84	}
85	}	×
86
87	pub fn new_virtual() -> UserStream {	3✔
88	UserStream::Virtual	3✔
89	}	3✔
90	}
91
92	impl Stream for UserStream {
93	type Item = String;
94
95	#[instrument(level = "debug", ret, skip(cx))]
96	fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {	3✔
97	let this = self.project();
98	match this {
99	UserStreamProj::Real(stdin_wrapper) => {
100	let next = Pin::new(stdin_wrapper).poll_next(cx);
101	match next {
102	Poll::Ready(Some(s)) => Poll::Ready(Some(s)),
103	Poll::Ready(None) => Poll::Ready(None),
104	Poll::Pending => Poll::Pending,
105	}
106	}
107	UserStreamProj::Virtual {} => Poll::Ready(None),
108	}
109	}	3✔
110	}

lpenz / ogle / 20974679380

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