1#![warn(rust_2018_idioms)]
2#![cfg(all(feature = "full", not(target_os = "wasi")))] // Wasi does not support panic recovery
3
4use std::pin::Pin;
5use std::task::{Context, Poll};
6use tokio::io::{self, AsyncRead, AsyncReadExt, ReadBuf};
7use tokio_test::assert_ok;
8
9mod support {
10 pub(crate) mod leaked_buffers;
11}
12use support::leaked_buffers::LeakedBuffers;
13
14#[tokio::test]
15async fn take() {
16 let mut buf = [0; 6];
17 let rd: &[u8] = b"hello world";
18
19 let mut rd = rd.take(4);
20 let n = assert_ok!(rd.read(&mut buf).await);
21 assert_eq!(n, 4);
22 assert_eq!(&buf, &b"hell\0\0"[..]);
23}
24
25#[tokio::test]
26async fn issue_4435() {
27 let mut buf = [0; 8];
28 let rd: &[u8] = b"hello world";
29
30 let rd = rd.take(4);
31 tokio::pin!(rd);
32
33 let mut read_buf = ReadBuf::new(&mut buf);
34 read_buf.put_slice(b"AB");
35
36 futures::future::poll_fn(|cx| rd.as_mut().poll_read(cx, &mut read_buf))
37 .await
38 .unwrap();
39 assert_eq!(&buf, &b"ABhell\0\0"[..]);
40}
41
42struct BadReader {
43 leaked_buffers: LeakedBuffers,
44}
45
46impl BadReader {
47 fn new() -> Self {
48 Self {
49 leaked_buffers: LeakedBuffers::new(),
50 }
51 }
52}
53
54impl AsyncRead for BadReader {
55 fn poll_read(
56 mut self: Pin<&mut Self>,
57 _cx: &mut Context<'_>,
58 read_buf: &mut ReadBuf<'_>,
59 ) -> Poll<io::Result<()>> {
60 let mut buf = ReadBuf::new(unsafe { self.leaked_buffers.create(10) });
61 buf.put_slice(&[123; 10]);
62 *read_buf = buf;
63
64 Poll::Ready(Ok(()))
65 }
66}
67
68#[tokio::test]
69#[should_panic]
70async fn bad_reader_fails() {
71 let mut buf = Vec::with_capacity(10);
72
73 BadReader::new().take(10).read_buf(&mut buf).await.unwrap();
74}
75