Source code
Revision control
Copy as Markdown
Other Tools
#![warn(rust_2018_idioms)]
#![cfg(all(feature = "full", not(tokio_wasi)))] // Wasi doesn't support bind
use std::io::{Error, ErrorKind, Result};
use std::io::{Read, Write};
use std::{net, thread};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::try_join;
#[tokio::test]
async fn split() -> Result<()> {
const MSG: &[u8] = b"split";
let listener = TcpListener::bind("127.0.0.1:0").await?;
let addr = listener.local_addr()?;
let (stream1, (mut stream2, _)) = try_join! {
TcpStream::connect(&addr),
listener.accept(),
}?;
let (mut read_half, mut write_half) = stream1.into_split();
let ((), (), ()) = try_join! {
async {
let len = stream2.write(MSG).await?;
assert_eq!(len, MSG.len());
let mut read_buf = vec![0u8; 32];
let read_len = stream2.read(&mut read_buf).await?;
assert_eq!(&read_buf[..read_len], MSG);
Result::Ok(())
},
async {
let len = write_half.write(MSG).await?;
assert_eq!(len, MSG.len());
Ok(())
},
async {
let mut read_buf = vec![0u8; 32];
let peek_len1 = read_half.peek(&mut read_buf[..]).await?;
let peek_len2 = read_half.peek(&mut read_buf[..]).await?;
assert_eq!(peek_len1, peek_len2);
let read_len = read_half.read(&mut read_buf[..]).await?;
assert_eq!(peek_len1, read_len);
assert_eq!(&read_buf[..read_len], MSG);
Ok(())
},
}?;
Ok(())
}
#[tokio::test]
async fn reunite() -> Result<()> {
let listener = net::TcpListener::bind("127.0.0.1:0")?;
let addr = listener.local_addr()?;
let handle = thread::spawn(move || {
drop(listener.accept().unwrap());
drop(listener.accept().unwrap());
});
let stream1 = TcpStream::connect(&addr).await?;
let (read1, write1) = stream1.into_split();
let stream2 = TcpStream::connect(&addr).await?;
let (_, write2) = stream2.into_split();
let read1 = match read1.reunite(write2) {
Ok(_) => panic!("Reunite should not succeed"),
Err(err) => err.0,
};
read1.reunite(write1).expect("Reunite should succeed");
handle.join().unwrap();
Ok(())
}
/// Test that dropping the write half actually closes the stream.
#[tokio::test]
async fn drop_write() -> Result<()> {
const MSG: &[u8] = b"split";
let listener = net::TcpListener::bind("127.0.0.1:0")?;
let addr = listener.local_addr()?;
let handle = thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
stream.write_all(MSG).unwrap();
let mut read_buf = [0u8; 32];
let res = match stream.read(&mut read_buf) {
Ok(0) => Ok(()),
Ok(len) => Err(Error::new(
ErrorKind::Other,
format!("Unexpected read: {} bytes.", len),
)),
Err(err) => Err(err),
};
drop(stream);
res
});
let stream = TcpStream::connect(&addr).await?;
let (mut read_half, write_half) = stream.into_split();
let mut read_buf = [0u8; 32];
let read_len = read_half.read(&mut read_buf[..]).await?;
assert_eq!(&read_buf[..read_len], MSG);
// drop it while the read is in progress
std::thread::spawn(move || {
thread::sleep(std::time::Duration::from_millis(10));
drop(write_half);
});
match read_half.read(&mut read_buf[..]).await {
Ok(0) => {}
Ok(len) => panic!("Unexpected read: {} bytes.", len),
Err(err) => panic!("Unexpected error: {}.", err),
}
handle.join().unwrap().unwrap();
Ok(())
}