Extend the polling feature to cover TryStream::try_read_inner()

src/process/mod.rs

@@ -1,6 +1,6 @@
 //! This module contains a platform independent abstraction over an os process.
 
-use std::io::Result;
+use std::{io::Result, time::Duration};
 
 #[cfg(unix)]
 pub mod unix;
@@ -72,6 +72,17 @@ where
 pub trait NonBlocking {
     /// Sets a [std::io::Read]er into a non/blocking mode.
     fn set_blocking(&mut self, on: bool) -> Result<()>;
+
+    /// Wait for data to be available from the stream.
+    ///
+    /// Returns true if there is data available or if availability is unknown.
+    fn ready(&self, timeout: Duration) -> Result<bool> {
+        let _ = timeout;
+
+        // The default implementation is always ready, meaning we fallback
+        // to a busy-wait if the platform code doesn't provide an implementation.
+        Ok(true)
+    }
 }
 
 impl<T> NonBlocking for &mut T
@@ -81,6 +92,10 @@ where
     fn set_blocking(&mut self, on: bool) -> Result<()> {
         T::set_blocking(self, on)
     }
+
+    fn ready(&self, timeout: Duration) -> Result<bool> {
+        T::ready(self, timeout)
+    }
 }
 
 /// Terminal configuration trait, used for IO configuration.

src/process/unix.rs

@@ -149,6 +149,19 @@ impl NonBlocking for PtyStream {
             false => make_non_blocking(fd, true),
         }
     }
+
+    #[cfg(feature = "polling")]
+    fn ready(&self, timeout: std::time::Duration) -> Result<bool> {
+        use polling::{Event, Poller};
+
+        let poller = Poller::new()?;
+        poller.add(self.handle.as_raw_fd(), Event::readable(1))?;
+
+        let mut events = Vec::new();
+        let num_ready = poller.wait(&mut events, Some(timeout))?;
+
+        Ok(num_ready > 0)
+    }
 }
 
 impl AsRawFd for PtyStream {

src/session/sync_session.rs

@@ -362,6 +362,10 @@ where
     fn set_blocking(&mut self, on: bool) -> io::Result<()> {
         S::set_blocking(self.get_stream_mut(), on)
     }
+
+    fn ready(&self, timeout: Duration) -> io::Result<bool> {
+        S::ready(self.get_stream(), timeout)
+    }
 }
 
 #[cfg(unix)]
@@ -504,6 +508,12 @@ where
 
     // non-buffered && non-blocking read
     fn try_read_inner(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        // We don't know what the session timeout is but it is unlikely to be
+        // less than 100ms (and if it is then polling is optional anyway).
+        if !self.stream.get_mut().ready(Duration::from_millis(100))? {
+            return Err(io::Error::from(io::ErrorKind::WouldBlock));
+        }
+
         self.stream.get_mut().set_blocking(false)?;
 
         let result = self.stream.get_mut().read(buf);

src/stream/log.rs

@@ -4,6 +4,7 @@
 use std::{
     io::{self, Read, Result, Write},
     ops::{Deref, DerefMut},
+    time::Duration,
 };
 
 #[cfg(feature = "async")]
@@ -89,6 +90,10 @@ where
     fn set_blocking(&mut self, on: bool) -> Result<()> {
         self.stream.set_blocking(on)
     }
+
+    fn ready(&self, timeout: Duration) -> Result<bool> {
+        self.stream.ready(timeout)
+    }
 }
 
 impl<S, W> Deref for LogStream<S, W> {