use std::future::Future;
use std::io;
use std::thread::JoinHandle;
use tokio::sync::oneshot;
pub struct Watch {
runtime: tokio::runtime::Runtime,
stopped: oneshot::Receiver<io::Result<()>>,
ended: Option<io::Result<()>>,
reader: JoinHandle<()>,
}
impl Watch {
pub(crate) fn start() -> io::Result<Self> {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_io()
.build()?;
let (stop, stopped) = oneshot::channel();
let reader = std::thread::Builder::new()
.name("stdin-close".to_owned())
.spawn(move || {
let read = io::copy(&mut io::stdin().lock(), &mut io::sink()).map(|_| ());
if stop.send(read).is_err() {
eprintln!("stdin_watch_receiver_closed: listener already ended");
}
})?;
Ok(Self {
runtime,
stopped,
ended: None,
reader,
})
}
pub(crate) fn register<T>(&self, bind: impl FnOnce() -> io::Result<T>) -> io::Result<T> {
let entered = self.runtime.enter();
let result = bind();
drop(entered);
result
}
pub(crate) fn next<T>(
&mut self,
incoming: impl Future<Output = io::Result<T>>,
) -> Option<io::Result<T>> {
self.runtime.block_on(async {
tokio::select! {
biased;
ended = &mut self.stopped => {
self.ended = Some(ended.map_err(io::Error::other).and_then(|ended| ended));
None
}
accepted = incoming => Some(accepted),
}
})
}
pub(crate) fn finish(self) -> io::Result<()> {
let ended = self.ended.ok_or_else(|| {
io::Error::other("stdin_watch_not_observed: listener ended before stdin")
})?;
let reader = self
.reader
.join()
.map_err(|_| io::Error::other("stdin_watch_failed: reader panicked"));
match (ended, reader) {
(Ok(()), Ok(())) => Ok(()),
(Err(error), Ok(())) | (Ok(()), Err(error)) => Err(error),
(Err(ended), Err(reader)) => Err(io::Error::other(format!("{ended}; {reader}"))),
}
}
}