Make UvTImer selectable and implement recv_timeout

src/libstd/rt/io/timer.rs

@@ -8,12 +8,17 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
+use comm;
+use kinds::Send;
 use option::{Option, Some, None};
 use result::{Ok, Err};
 use rt::io::{io_error};
 use rt::rtio::{IoFactory, IoFactoryObject,
                RtioTimer, RtioTimerObject};
 use rt::local::Local;
+use rt::select::SelectInner;
+use rt::sched::Scheduler;
+use rt::kill::BlockedTask;
 
 pub struct Timer(~RtioTimerObject);
 
@@ -24,6 +29,13 @@ pub fn sleep(msecs: u64) {
     timer.sleep(msecs)
 }
 
+/// Sleep the current task for `msecs` milliseconds.
+pub fn sleep_uv(msecs: u64) {
+    let mut timer = Timer::new().expect("timer::sleep: could not create a Timer");
+
+    timer.sleep_uv(msecs)
+}
+
 impl Timer {
 
     pub fn new() -> Option<Timer> {
@@ -46,24 +58,139 @@ impl Timer {
     pub fn sleep(&mut self, msecs: u64) {
         (**self).sleep(msecs);
     }
+
+    pub fn sleep_uv(&mut self, msecs: u64) {
+        (**self).sleep_uv(msecs, true);
+    }
+}
+
+impl SelectInner for Timer {
+
+    fn optimistic_check(&mut self) -> bool {
+        (**self).optimistic_check()
+    }
+
+    fn block_on(&mut self, sched: &mut Scheduler, task: BlockedTask) -> bool {
+        (**self).block_on(sched, task)
+    }
+
+    fn unblock_from(&mut self) -> bool {false}
+}
+
+trait TimedPort<T: Send> {
+
+ /**
+  * This implementation adds the required
+  * API for recv_timeout with an approximate
+  * but not safe behavior.
+  *
+  * Current implementations of this Trait for
+  * both PortOne and Port poll on the port every
+  * second to check for new messages. A correct
+  * implementation for recv_timeout should implement
+  * `SelectInner` and Select for UvTimer and re-write
+  * the sleep method around that.
+  *
+  * FIXME: (flaper87) #9195
+  */
+  fn recv_timeout(self, msecs: u64) -> Option<T>;
+}
+
+impl<T: Send> TimedPort<T> for comm::PortOne<T> {
+
+    fn recv_timeout(self, msecs: u64) -> Option<T> {
+        let mut tout = msecs;
+        let mut timer = Timer::new().unwrap();
+
+        while tout > 0 {
+            if self.peek() { return Some(self.recv()); }
+            timer.sleep(1000);
+            tout -= 1000;
+        }
+
+        None
+    }
+}
+
+
+impl<T: Send> TimedPort<T> for comm::Port<T> {
+
+    fn recv_timeout(self, msecs: u64) -> Option<T> {
+        let mut tout = msecs;
+        let mut timer = Timer::new().unwrap();
+
+        while tout > 0 {
+            if self.peek() { return Some(self.recv()); }
+            timer.sleep_uv(1000);
+            tout -= 1000;
+        }
+
+        None
+    }
 }
 
 #[cfg(test)]
 mod test {
     use super::*;
     use rt::test::*;
+    use task;
+    use comm;
+
     #[test]
     fn test_io_timer_sleep_simple() {
         do run_in_mt_newsched_task {
             let timer = Timer::new();
-            do timer.map_move |mut t| { t.sleep(1) };
+            do timer.map_move |mut t| { t.sleep(1000) };
         }
     }
 
     #[test]
     fn test_io_timer_sleep_standalone() {
         do run_in_mt_newsched_task {
-            sleep(1)
+            sleep(1000)
+        }
+    }
+
+    #[test]
+    fn test_io_timer_sleep_uv_simple() {
+        do run_in_mt_newsched_task {
+            let timer = Timer::new();
+            do timer.map_move |mut t| { t.sleep_uv(1) };
+        }
+    }
+
+    #[test]
+    fn test_io_timer_sleep_uv_standalone() {
+        do run_in_mt_newsched_task {
+            sleep_uv(1)
+        }
+    }
+
+    #[test]
+    fn test_recv_timeout() {
+        do run_in_newsched_task {
+            let (p, c) = comm::stream::<int>();
+            do task::spawn {
+                let mut t = Timer::new().unwrap();
+                t.sleep_uv(1000);
+                c.send(1);
+            }
+
+            assert!(p.recv_timeout(2000).unwrap() == 1);
+        }
+    }
+
+    #[test]
+    fn test_recv_timeout_expire() {
+        do run_in_newsched_task {
+            let (p, c) = comm::stream::<int>();
+            do task::spawn {
+                let mut t = Timer::new().unwrap();
+                t.sleep_uv(3000);
+                c.send(1);
+            }
+
+            assert!(p.recv_timeout(1000).is_none());
         }
     }
 }

src/libstd/rt/rtio.rs

@@ -132,6 +132,8 @@ pub trait RtioUdpSocket : RtioSocket {
 
 pub trait RtioTimer {
     fn sleep(&mut self, msecs: u64);
+    fn sleep_uv(&mut self, msecs: u64, deschedule: bool);
+    fn sleep_then(&mut self, msecs: u64, cb: &fn(&mut Self));
 }
 
 pub trait RtioFileStream {