Limit the scope of borrow_mut in consume_queue

Otherwise we may accidentally hold the borrowed ref cell for too long which can
cause a borrow error.

Closes #190

src/reactor/mod.rs

@@ -411,7 +411,8 @@ impl Core {
         // TODO: can we do better than `.unwrap()` here?
         let unpark = self.rx_readiness.clone();
         loop {
-            match self.rx.borrow_mut().poll_stream(unpark.clone()).unwrap() {
+            let msg = self.rx.borrow_mut().poll_stream(unpark.clone()).unwrap();
+            match msg {
                 Async::Ready(Some(msg)) => self.notify(msg),
                 Async::NotReady |
                 Async::Ready(None) => break,

tests/spawn.rs

@@ -2,13 +2,14 @@ extern crate tokio_core;
 extern crate env_logger;
 extern crate futures;
 
-use std::time::Duration;
 use std::sync::mpsc;
+use std::thread;
+use std::time::Duration;
 
 use futures::Future;
 use futures::future;
 use futures::sync::oneshot;
-use tokio_core::reactor::Core;
+use tokio_core::reactor::{Core, Timeout};
 
 #[test]
 fn simple() {
@@ -76,3 +77,21 @@ fn spawn_in_poll() {
 
     assert_eq!(lp.run(rx1.join(rx2)).unwrap(), (1, 2));
 }
+
+#[test]
+fn drop_timeout_in_spawn() {
+    drop(env_logger::init());
+    let mut lp = Core::new().unwrap();
+
+    let (tx, rx) = oneshot::channel();
+    let remote = lp.remote();
+    thread::spawn(move || {
+        remote.spawn(|handle| {
+            drop(Timeout::new(Duration::new(1, 0), handle));
+            tx.send(()).unwrap();
+            Ok(())
+        });
+    });
+
+    lp.run(rx).unwrap();
+}