Improve tests

src/lib.rs

@@ -288,6 +288,8 @@ mod tests {
     use tokio_core::reactor::Core;
     use tokio_core::io::{self, Io};
     use futures::{Stream, Future};
+    use futures::sync::oneshot;
+    use std::thread;
 
     use super::Endpoint;
     use super::IpcConnection;
@@ -304,37 +306,53 @@ mod tests {
         format!(r"\\.\pipe\my-pipe-{}", num)
     }
 
+    fn run_server(path: &str) {
+        let path = path.to_owned();
+        let (ok_signal, ok_rx) = oneshot::channel();
+        thread::spawn(|| {
+            let mut core = Core::new().expect("failed to spawn an event loop");
+            let endpoint = Endpoint::new(path, &core.handle()).expect("failed to open endpoint");
+            ok_signal.send(()).expect("failed to send ok");
+            let srv = endpoint.incoming()
+                .for_each(|(stream, _)| {
+                    let (reader, writer) = stream.split();
+                    let buf = [0u8; 5];
+                    io::read_exact(reader,buf).and_then(move |(_reader, buf)| {
+                        let mut reply = vec![];
+                        reply.extend(&buf[..]);
+                        io::write_all(writer, reply)
+                    })
+                    .map_err(|e| {trace!("io error: {:?}", e); e })
+                    .map(|_| ())
+                })
+                .map_err(|_| ());
+            core.run(srv).expect("server failed");
+        });
+        ok_rx.wait().expect("failed to receive handle")
+    }
+
     #[test]
     fn smoke_test() {
         let path = random_pipe_path();
-        let mut core = Core::new().expect("failed to start event loop");
+        run_server(&path);
+        let mut core = Core::new().expect("failed to spawn an event loop");
         let handle = core.handle();
-        let endpoint = Endpoint::new(path.to_owned(), &core.handle()).expect("failed to open a server socket");
-        let srv = endpoint.incoming()
-            .for_each(|(stream, _)| {
-                let (reader, writer) = stream.split();
-                let buf = [0u8; 5];
-                io::read_exact(reader,buf).and_then(move |(_reader, buf)| {
-                    let mut reply = vec![];
-                    reply.extend(&buf[..]);
-                    io::write_all(writer, reply)
-                })
-                .map_err(|e| {trace!("io error: {:?}", e); e })
-                .map(|_| ())
-            })
-            .map_err(|_| ())
-                ;
-        handle.spawn(srv);
+
         let client = IpcConnection::connect(&path, &handle).expect("failed to open a client");
+        let other_client = IpcConnection::connect(&path, &handle).expect("failed to open a client");
         let msg = b"hello";
+
         let mut rx_buf = vec![0u8; msg.len()];
         let client_fut = io::write_all(client, &msg).and_then(|(client, _)| {
-            io::read_exact(client, &mut rx_buf).map(|(_, buf)| {
-                buf
-            })
+            io::read_exact(client, &mut rx_buf).map(|(_, buf)| buf)
         });
 
-        let received_message = core.run(client_fut).expect("failed to read from server");
-        assert_eq!(received_message, msg);
+        let mut rx_buf2 = vec![0u8; msg.len()];
+        let other_client_fut = io::write_all(other_client, &msg).and_then(|(client, _)| {
+            io::read_exact(client, &mut rx_buf2).map(|(_, buf)| buf)
+        });
+        let (rx_msg, other_rx_msg) = core.run(client_fut.join(other_client_fut)).expect("failed to read from server");
+        assert_eq!(rx_msg,  msg);
+        assert_eq!(other_rx_msg,  msg);
     }
 }