1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
|
use asynchronous_codec::{Bytes, BytesCodec, FramedWrite, LinesCodec};
use core::iter::Iterator;
use futures::io::{AsyncWrite, Cursor};
use futures::sink::SinkExt;
use futures::{executor, stream, stream::StreamExt};
use std::pin::Pin;
use std::task::{Context, Poll};
// An iterator which outputs a single zero byte up to limit times
struct ZeroBytes {
pub count: usize,
pub limit: usize,
}
impl Iterator for ZeroBytes {
type Item = Bytes;
fn next(&mut self) -> Option<Self::Item> {
if self.count >= self.limit {
None
} else {
self.count += 1;
Some(Bytes::from_static(b"\0"))
}
}
}
// An AsyncWrite which is always ready and just consumes the data
struct AsyncWriteNull {
// number of poll_write calls
pub num_poll_write: usize,
// size of the last poll_write
pub last_write_size: usize,
}
impl AsyncWrite for AsyncWriteNull {
fn poll_write(
mut self: Pin<&mut Self>,
_cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<std::io::Result<usize>> {
self.num_poll_write += 1;
self.last_write_size = buf.len();
Poll::Ready(Ok(buf.len()))
}
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
Poll::Ready(Ok(()))
}
fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
Poll::Ready(Ok(()))
}
}
#[test]
fn line_write() {
let curs = Cursor::new(vec![0u8; 16]);
let mut framer = FramedWrite::new(curs, LinesCodec {});
executor::block_on(framer.send("Hello\n".to_owned())).unwrap();
executor::block_on(framer.send("World\n".to_owned())).unwrap();
let curs = framer.into_inner();
assert_eq!(&curs.get_ref()[0..12], b"Hello\nWorld\n");
assert_eq!(curs.position(), 12);
}
#[test]
fn line_write_to_eof() {
let mut buf = [0u8; 16];
let curs = Cursor::new(&mut buf[..]);
let mut framer = FramedWrite::new(curs, LinesCodec {});
let _err =
executor::block_on(framer.send("This will fill up the buffer\n".to_owned())).unwrap_err();
let curs = framer.into_inner();
assert_eq!(curs.position(), 16);
assert_eq!(&curs.get_ref()[0..16], b"This will fill u");
}
#[test]
fn send_high_water_mark() {
// stream will output 999 bytes, 1 at at a time, and will always be ready
let mut stream = stream::iter(ZeroBytes {
count: 0,
limit: 999,
})
.map(Ok);
// sink will eat whatever it receives
let io = AsyncWriteNull {
num_poll_write: 0,
last_write_size: 0,
};
// expect two sends
let mut framer = FramedWrite::new(io, BytesCodec {});
framer.set_send_high_water_mark(500);
executor::block_on(framer.send_all(&mut stream)).unwrap();
let io = framer.into_inner();
assert_eq!(io.num_poll_write, 2);
assert_eq!(io.last_write_size, 499);
}
|
