Fixed tests

Author: thehydroimpulse

src/dispatcher.rs

@@ -152,7 +152,10 @@ mod tests {
     use event_loop::SENDER;
     use protocol::{ThriftMessage, ThriftMessageType};
     use binary_protocol::BinaryDeserializer;
+    use std::sync::mpsc::channel;
     use util;
+    use std::thread;
+    use std::time::Duration;
 
     #[test]
     fn should_create_server_dispatcher() {
@@ -164,19 +167,32 @@ mod tests {
     fn should_start_server() {
         let addr: SocketAddr = "127.0.0.1:5955".parse().unwrap();
         let (handle_server, server) = Dispatcher::spawn(Role::Server(addr.clone())).unwrap();
+        thread::sleep(Duration::from_millis(30));
         let (handle_client, client) = Dispatcher::spawn(Role::Client(addr.clone())).unwrap();
 
         let buf = util::create_empty_thrift_message("foobar123", ThriftMessageType::Call);
 
         let (res, future) = Future::<(ThriftMessage, BinaryDeserializer<Cursor<Vec<u8>>>)>::channel();
         client.send(Incoming::Call("foobar123".to_string(), buf, res)).unwrap();
 
+        let (res_tx, res_rx) = channel();
+        let cloned = res_tx.clone();
         future.and_then(move |(msg, de)| {
             println!("[test]: Received: {:?}", msg);
             SENDER.clone().send(Message::Shutdown);
+            res_tx.send(0);
             Async::Ok(())
         });
 
+        thread::spawn(move || -> Result<(), ()> {
+            thread::sleep(Duration::from_millis(3000));
+            SENDER.clone().send(Message::Shutdown);
+            panic!("Test timeout was hit. This means the Reactor did not shutdown and a response was not received.");
+            cloned.send(1);
+        });
+
         Reactor::run().join();
+
+        assert_eq!(res_rx.recv().unwrap(), 0);
     }
 }

src/reactor.rs

@@ -366,10 +366,11 @@ impl Reactor {
                 self.connections.get_mut(&id).expect("connection was not found #2").write(&*data);
             },
             Message::Shutdown => {
+                println!("Shutting down...");
                 event_loop.shutdown();
             },
             Message::Connect(addr, id_tx, tx) => {
-                let mut mio_stream = TcpStream::connect(&addr).expect("MIO ERR");
+                let mut mio_stream = TcpStream::connect(&addr)?;
                 let new_token = Token(self.current_token);
                 id_tx.send(Id(new_token));
                 let mut conn = Connection::new((mio_stream, addr), new_token, tx);