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
100
101
102
103
104
105
|
use crate::errors::Error;
use futures::{Async, Stream};
use hyper::Chunk;
use std::{
cmp,
io::{self, Read},
};
use tokio_io::AsyncRead;
/*
* The following is taken from
* https://github.com/ferristseng/rust-ipfs-api/blob/master/ipfs-api/src/read.rs.
* TODO: see with upstream author to move to a separate crate.
*/
/// The state of a stream returning Chunks.
///
enum ReadState {
/// A chunk is ready to be read from.
///
Ready(Chunk, usize),
/// The next chunk isn't ready yet.
///
NotReady,
}
/// Reads from a stream of chunks asynchronously.
///
pub struct StreamReader<S> {
stream: S,
state: ReadState,
}
impl<S> StreamReader<S>
where
S: Stream<Item = Chunk, Error = Error>,
{
#[inline]
pub fn new(stream: S) -> StreamReader<S> {
StreamReader {
stream,
state: ReadState::NotReady,
}
}
}
impl<S> Read for StreamReader<S>
where
S: Stream<Item = Chunk, Error = Error>,
{
fn read(
&mut self,
buf: &mut [u8],
) -> io::Result<usize> {
loop {
let ret;
match self.state {
// Stream yielded a Chunk to read.
//
ReadState::Ready(ref mut chunk, ref mut pos) => {
let chunk_start = *pos;
let len = cmp::min(buf.len(), chunk.len() - chunk_start);
let chunk_end = chunk_start + len;
buf[..len].copy_from_slice(&chunk[chunk_start..chunk_end]);
*pos += len;
if *pos == chunk.len() {
ret = len;
} else {
return Ok(len);
}
}
// Stream is not ready, and a Chunk needs to be read.
//
ReadState::NotReady => {
match self.stream.poll() {
// Polling stream yielded a Chunk that can be read from.
//
Ok(Async::Ready(Some(chunk))) => {
self.state = ReadState::Ready(chunk, 0);
continue;
}
// Polling stream yielded EOF.
//
Ok(Async::Ready(None)) => return Ok(0),
// Stream could not be read from.
//
Ok(Async::NotReady) => return Err(io::ErrorKind::WouldBlock.into()),
Err(e) => return Err(io::Error::new(io::ErrorKind::Other, e.to_string())),
}
}
}
self.state = ReadState::NotReady;
return Ok(ret);
}
}
}
impl<S> AsyncRead for StreamReader<S> where S: Stream<Item = Chunk, Error = Error> {}
|
