Change a return value of reactor::poll to io::Result. (#40)

* Change a return value of reactor::poll to io::Result.

* Revert "Change a return value of reactor::poll to io::Result."

This reverts commit 281d8c32d44d8971e0aebf3833a72c02273ac3d2.

* Return a result from reactor::poll.

* Drop a reactor if any error occurs. Fix warnings in tests.

* Update a documentation for reactor::turn.

* Unwrap the last turn() call in tests.

src/reactor/global.rs

@@ -49,6 +49,6 @@ impl Drop for HelperThread {
 
 fn run(mut reactor: Reactor, done: Arc<AtomicBool>) {
     while !done.load(Ordering::SeqCst) {
-        reactor.turn(None);
+        reactor.turn(None).unwrap();
     }
 }

src/reactor/mod.rs

@@ -144,23 +144,25 @@ impl Reactor {
     ///
     /// # Return value
     ///
-    /// This function returns an instance of `Turn` which as of today has no
-    /// extra information with it and can be safely discarded. In the future
-    /// this return value may contain information about what happened while this
+    /// This function returns an instance of `Turn` or `io::Error`.
+    ///
+    /// `Turn` as of today has no extra information with it and can be safely discarded.
+    /// In the future `Turn` may contain information about what happened while this
     /// reactor blocked.
-    pub fn turn(&mut self, max_wait: Option<Duration>) -> Turn {
-        self.poll(max_wait);
-        Turn { _priv: () }
+    ///
+    /// `io::Error` is possible when there is an internal `tokio` bug or I/O error.
+    pub fn turn(&mut self, max_wait: Option<Duration>) -> io::Result<Turn> {
+        self.poll(max_wait)?;
+        Ok(Turn { _priv: () })
     }
 
-    fn poll(&mut self, max_wait: Option<Duration>) {
+    fn poll(&mut self, max_wait: Option<Duration>) -> io::Result<()> {
         // Block waiting for an event to happen, peeling out how many events
         // happened.
         match self.inner.io.poll(&mut self.events, max_wait) {
             Ok(_) => {}
-            Err(ref e) if e.kind() == ErrorKind::Interrupted => return,
-            // TODO: This should return an io::Result instead of panic.
-            Err(e) => panic!("error in poll: {}", e),
+            Err(ref e) if e.kind() == ErrorKind::Interrupted => return Ok(()),
+            Err(e) => return Err(e),
         }
 
         // Process all the events that came in, dispatching appropriately
@@ -176,6 +178,8 @@ impl Reactor {
                 self.dispatch(token, event.readiness());
             }
         }
+
+        Ok(())
     }
 
     fn dispatch(&mut self, token: mio::Token, ready: mio::Ready) {

tests/wakeup.rs

@@ -10,13 +10,13 @@ use tokio::reactor::Reactor;
 fn works() {
     let mut r = Reactor::new().unwrap();
     r.handle().wakeup();
-    r.turn(None);
+    r.turn(None).unwrap();
 
     let now = Instant::now();
     let mut n = 0;
     while now.elapsed() < Duration::from_millis(10) {
         n += 1;
-        r.turn(Some(Duration::from_millis(10)));
+        r.turn(Some(Duration::from_millis(10))).unwrap();
     }
     assert!(n < 5);
 }
@@ -37,7 +37,7 @@ fn wakes() {
 
     for _ in 0..N {
         tx.send(()).unwrap();
-        r.turn(None);
+        r.turn(None).unwrap();
     }
     t.join().unwrap();
 }