#![cfg(feature = "tokio")] extern crate tokio_core; extern crate xz2; extern crate tokio_io; extern crate futures; extern crate rand; use std::thread; use std::net::{Shutdown, TcpListener}; use std::io::{Read, Write}; use xz2::read; use xz2::write; use futures::Future; use rand::{Rng, thread_rng}; use tokio_core::net::TcpStream; use tokio_core::reactor::Core; use tokio_io::AsyncRead; use tokio_io::io::{copy, shutdown}; #[test] fn tcp_stream_echo_pattern() { const N: u8 = 16; const M: usize = 16 * 1024; let mut core = Core::new().unwrap(); let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let addr = listener.local_addr().unwrap(); let t = thread::spawn(move || { let a = listener.accept().unwrap().0; let b = a.try_clone().unwrap(); let t = thread::spawn(move || { let mut b = read::XzDecoder::new(b); let mut buf = [0; M]; for i in 0..N { b.read_exact(&mut buf).unwrap(); for byte in buf.iter() { assert_eq!(*byte, i); } } assert_eq!(b.read(&mut buf).unwrap(), 0); }); let mut a = write::XzEncoder::new(a, 6); for i in 0..N { let buf = [i; M]; a.write_all(&buf).unwrap(); } a.finish().unwrap() .shutdown(Shutdown::Write).unwrap(); t.join().unwrap(); }); let handle = core.handle(); let stream = TcpStream::connect(&addr, &handle); let copy = stream.and_then(|s| { let (a, b) = s.split(); let a = read::XzDecoder::new(a); let b = write::XzEncoder::new(b, 6); copy(a, b) }).then(|result| { let (amt, _a, b) = result.unwrap(); assert_eq!(amt, (N as u64) * (M as u64)); shutdown(b).map(|_| ()) }); core.run(copy).unwrap(); t.join().unwrap(); } #[test] fn echo_random() { let v = thread_rng().gen_iter::().take(1024 * 1024).collect::>(); let mut core = Core::new().unwrap(); let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let addr = listener.local_addr().unwrap(); let v2 = v.clone(); let t = thread::spawn(move || { let a = listener.accept().unwrap().0; let b = a.try_clone().unwrap(); let mut v3 = v2.clone(); let t = thread::spawn(move || { let mut b = read::XzDecoder::new(b); let mut buf = [0; 1024]; while v3.len() > 0 { let n = b.read(&mut buf).unwrap(); for (actual, expected) in buf[..n].iter().zip(&v3) { assert_eq!(*actual, *expected); } v3.drain(..n); } assert_eq!(b.read(&mut buf).unwrap(), 0); }); let mut a = write::XzEncoder::new(a, 6); a.write_all(&v2).unwrap(); a.finish().unwrap() .shutdown(Shutdown::Write).unwrap(); t.join().unwrap(); }); let handle = core.handle(); let stream = TcpStream::connect(&addr, &handle); let copy = stream.and_then(|s| { let (a, b) = s.split(); let a = read::XzDecoder::new(a); let b = write::XzEncoder::new(b, 6); copy(a, b) }).then(|result| { let (amt, _a, b) = result.unwrap(); assert_eq!(amt, v.len() as u64); shutdown(b).map(|_| ()) }); core.run(copy).unwrap(); t.join().unwrap(); }