History

[clone] a piped iterator

Sebastian Thiel committed 6 years ago

Commit 5148c851 parent 9cfa9b74 files · +138 −100Browse files

Unified · Split

4 files
  1. git-features/src/pipe.rs +69 −47
  2. git-features/tests/pipe.rs +68 −53
  3. git-transport/tests/client/http.rs +0 −0
  4. tasks.md +1 −0
git-features/src/pipe.rs+69−47
@@ -1,60 +1,82 @@
1−use bytes::{Buf, BufMut, BytesMut};
2−use std::io;
1+mod iter {
2+ pub struct Iter<T> {
3+ channel: std::sync::mpsc::Receiver<T>,
4+ }
5+ pub fn iter<T>(in_flight_writes: impl Into<Option<usize>>) -> (std::sync::mpsc::SyncSender<T>, Iter<T>) {
6+ let (tx, rx) = std::sync::mpsc::sync_channel(in_flight_writes.into().unwrap_or(0));
7+ (tx, Iter { channel: rx })
8+ }
39
4−pub struct Writer {
5− channel: std::sync::mpsc::SyncSender<BytesMut>,
6− buf: BytesMut,
7−}
10+ impl<T> Iterator for Iter<T> {
11+ type Item = T;
812
9−pub struct Reader {
10− channel: std::sync::mpsc::Receiver<BytesMut>,
11− buf: BytesMut,
13+ fn next(&mut self) -> Option<Self::Item> {
14+ self.channel.recv().ok()
15+ }
16+ }
1217 }
18+pub use iter::{iter, Iter};
19+
20+mod io {
21+ use bytes::{Buf, BufMut, BytesMut};
22+ use std::io;
23+
24+ pub struct Writer {
25+ channel: std::sync::mpsc::SyncSender<BytesMut>,
26+ buf: BytesMut,
27+ }
1328
14−impl io::Read for Reader {
15− fn read(&mut self, mut out: &mut [u8]) -> io::Result<usize> {
16− let mut written = 0;
17− while !out.is_empty() {
18− if self.buf.is_empty() {
19− match self.channel.recv() {
20− Ok(buf) => self.buf = buf,
21− Err(_) => break,
29+ pub struct Reader {
30+ channel: std::sync::mpsc::Receiver<BytesMut>,
31+ buf: BytesMut,
32+ }
33+
34+ impl io::Read for Reader {
35+ fn read(&mut self, mut out: &mut [u8]) -> io::Result<usize> {
36+ let mut written = 0;
37+ while !out.is_empty() {
38+ if self.buf.is_empty() {
39+ match self.channel.recv() {
40+ Ok(buf) => self.buf = buf,
41+ Err(_) => break,
42+ }
2243 }
44+ let bytes_to_write = self.buf.len().min(out.len());
45+ let (to_write, rest) = out.split_at_mut(bytes_to_write);
46+ self.buf.split_to(bytes_to_write).copy_to_slice(to_write);
47+ out = rest;
48+ written += bytes_to_write;
2349 }
24− let bytes_to_write = self.buf.len().min(out.len());
25− let (to_write, rest) = out.split_at_mut(bytes_to_write);
26− self.buf.split_to(bytes_to_write).copy_to_slice(to_write);
27− out = rest;
28− written += bytes_to_write;
50+ Ok(written)
2951 }
30− Ok(written)
3152 }
32−}
3353
34−impl io::Write for Writer {
35− fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
36− self.buf.put_slice(buf);
37− self.channel
38− .send(self.buf.split())
39− .map_err(|err| io::Error::new(io::ErrorKind::BrokenPipe, err))?;
40− Ok(buf.len())
41− }
54+ impl io::Write for Writer {
55+ fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
56+ self.buf.put_slice(buf);
57+ self.channel
58+ .send(self.buf.split())
59+ .map_err(|err| io::Error::new(io::ErrorKind::BrokenPipe, err))?;
60+ Ok(buf.len())
61+ }
4262
43− fn flush(&mut self) -> io::Result<()> {
44− Ok(())
63+ fn flush(&mut self) -> io::Result<()> {
64+ Ok(())
65+ }
4566 }
46−}
4767
48−pub fn unidirectional(in_flight_writes: impl Into<Option<usize>>) -> (Writer, Reader) {
49− let (tx, rx) = std::sync::mpsc::sync_channel(in_flight_writes.into().unwrap_or(0));
50− (
51− Writer {
52− channel: tx,
53− buf: BytesMut::with_capacity(4096),
54− },
55− Reader {
56− channel: rx,
57− buf: BytesMut::new(),
58− },
59− )
68+ pub fn unidirectional(in_flight_writes: impl Into<Option<usize>>) -> (Writer, Reader) {
69+ let (tx, rx) = std::sync::mpsc::sync_channel(in_flight_writes.into().unwrap_or(0));
70+ (
71+ Writer {
72+ channel: tx,
73+ buf: BytesMut::with_capacity(4096),
74+ },
75+ Reader {
76+ channel: rx,
77+ buf: BytesMut::new(),
78+ },
79+ )
80+ }
6081 }
82+pub use io::{unidirectional, Reader, Writer};
git-features/tests/pipe.rs+68−53
@@ -1,64 +1,79 @@
1−use git_features::pipe;
2−use std::io::{ErrorKind, Read, Write};
1+mod iter {
2+ use git_features::pipe;
33
4−#[test]
5−fn threaded_read_to_end() {
6− let (mut writer, mut reader) = git_features::pipe::unidirectional(0);
4+ #[test]
5+ fn iterator_ends_when_sender_is_dropped() {
6+ let (send, mut iter) = pipe::iter(1);
7+ send.send(1).expect("buffered send successful");
8+ assert_eq!(iter.next(), Some(1), "it receives the item");
9+ drop(send);
10+ assert_eq!(iter.next(), None, "sender is dropped, so no more items for us");
11+ }
12+}
713
8− let message = "Hello, world!";
9− std::thread::spawn(move || {
10− writer
11− .write_all(message.as_bytes())
12− .expect("writes to work if reader is present")
13− });
14+mod io {
15+ use git_features::pipe;
16+ use std::io::{ErrorKind, Read, Write};
1417
15− let mut received = String::new();
16− reader.read_to_string(&mut received).unwrap();
18+ #[test]
19+ fn threaded_read_to_end() {
20+ let (mut writer, mut reader) = git_features::pipe::unidirectional(0);
1721
18− assert_eq!(&received, message);
19−}
22+ let message = "Hello, world!";
23+ std::thread::spawn(move || {
24+ writer
25+ .write_all(message.as_bytes())
26+ .expect("writes to work if reader is present")
27+ });
2028
21−#[test]
22−fn write_failure_propagates() {
23− let (mut writer, _) = pipe::unidirectional(None);
24− assert_eq!(
25− writer.write_all(b"must fail").unwrap_err().kind(),
26− ErrorKind::BrokenPipe
27− );
28−}
29+ let mut received = String::new();
30+ reader.read_to_string(&mut received).unwrap();
2931
30−#[test]
31−fn continue_on_empty_writes() {
32− let (mut writer, mut reader) = pipe::unidirectional(2);
33− writer.write(&[]).expect("write successful and non-blocking");
34− let input = b"hello";
35− writer
36− .write(input)
37− .expect("second write works as well as there is capacity");
38− let mut buf = vec![0u8; input.len()];
39− assert_eq!(reader.read(&mut buf).expect("read succeeds"), input.len());
40− assert_eq!(buf, &input[..]);
41−}
32+ assert_eq!(&received, message);
33+ }
4234
43−#[test]
44−fn small_reads() {
45− const BLOCK_SIZE: usize = 20;
46− let block_count = 20;
47− let (mut writer, mut reader) = pipe::unidirectional(Some(4));
48− std::thread::spawn(move || {
49− for _ in 0..block_count {
50− let data = &[0; BLOCK_SIZE];
51− writer.write_all(data).unwrap();
52− }
53− });
35+ #[test]
36+ fn write_failure_propagates() {
37+ let (mut writer, _) = pipe::unidirectional(None);
38+ assert_eq!(
39+ writer.write_all(b"must fail").unwrap_err().kind(),
40+ ErrorKind::BrokenPipe
41+ );
42+ }
43+
44+ #[test]
45+ fn continue_on_empty_writes() {
46+ let (mut writer, mut reader) = pipe::unidirectional(2);
47+ writer.write(&[]).expect("write successful and non-blocking");
48+ let input = b"hello";
49+ writer
50+ .write(input)
51+ .expect("second write works as well as there is capacity");
52+ let mut buf = vec![0u8; input.len()];
53+ assert_eq!(reader.read(&mut buf).expect("read succeeds"), input.len());
54+ assert_eq!(buf, &input[..]);
55+ }
56+
57+ #[test]
58+ fn small_reads() {
59+ const BLOCK_SIZE: usize = 20;
60+ let block_count = 20;
61+ let (mut writer, mut reader) = pipe::unidirectional(Some(4));
62+ std::thread::spawn(move || {
63+ for _ in 0..block_count {
64+ let data = &[0; BLOCK_SIZE];
65+ writer.write_all(data).unwrap();
66+ }
67+ });
5468
55− let mut small_read_buf = [0; BLOCK_SIZE / 2];
56− let mut bytes_read = 0;
57− while let Ok(size) = reader.read(&mut small_read_buf) {
58− if size == 0 {
59− break;
69+ let mut small_read_buf = [0; BLOCK_SIZE / 2];
70+ let mut bytes_read = 0;
71+ while let Ok(size) = reader.read(&mut small_read_buf) {
72+ if size == 0 {
73+ break;
74+ }
75+ bytes_read += size;
6076 }
61− bytes_read += size;
77+ assert_eq!(block_count * BLOCK_SIZE, bytes_read);
6278 }
63− assert_eq!(block_count * BLOCK_SIZE, bytes_read);
6479 }
git-transport/tests/client/http.rs+0−0

Empty file added.

tasks.md+1−0
@@ -24,6 +24,7 @@ · 23 unmodified lines
2424 * [ ] http://git-upload-pack
2525 * **prerequisites**
2626 * [x] pipe to link writers with readers using bytes
27+ * [x] piped iterator
2728 * [ ] HTTP trait for simple gets and post implemented for Curl
2829 * [ ] V1 handshake
2930 * **git-refs**