ruyi 0.1.6

An event-driven framework for non-blocking, asynchronous I/O in Rust
Documentation 
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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141

use std::io;
use std::mem;

use futures::{Poll, Future, Async};

use io::{AsyncRead, AsyncWrite};
use buf::ByteBuf;

enum ReadState<R> {
    Reading(R),
    Done(R),
    Dead,
}

pub struct Copy<R, W> {
    r: ReadState<R>,
    w: Option<W>,
    buf: Option<ByteBuf>,
    len: u64,
}

#[inline]
pub fn copy<R, W>(r: R, w: W) -> Copy<R, W>
where
    R: AsyncRead,
    W: AsyncWrite,
{
    Copy {
        r: ReadState::Reading(r),
        w: Some(w),
        buf: None,
        len: 0,
    }
}

impl<R, W> Copy<R, W>
where
    R: AsyncRead,
    W: AsyncWrite,
{
    #[inline]
    fn poll_copy(&mut self) -> Result<bool, io::Error> {
        let mut done_read = false;
        let mut eof = false;
        match self.r {
            ReadState::Reading(ref mut r) => {
                if r.is_readable() {
                    match super::read(r) {
                        Ok(Some((data, _))) => {
                            match self.buf {
                                Some(ref mut buf) => buf.extend(data),
                                None => self.buf = Some(data),
                            };
                            r.need_read()?;
                        }
                        Ok(None) => eof = true,
                        Err(e) => {
                            match e.kind() {
                                io::ErrorKind::WouldBlock => r.need_read()?,
                                _ => return Err(e),
                            }
                        }
                    }
                }
            }
            ReadState::Done(..) => done_read = true,
            _ => panic!("Attempted to poll Copy after completion"),
        }
        if eof {
            match mem::replace(&mut self.r, ReadState::Dead) {
                ReadState::Reading(r) => {
                    self.r = ReadState::Done(r);
                    done_read = true;
                }
                _ => ::unreachable(),
            }
        }

        match self.w {
            Some(ref mut w) => {
                if !w.is_writable() {
                    return Ok(false);
                }
                if let Some(ref mut buf) = self.buf {
                    match buf.write_out(w) {
                        Ok(n) => self.len += n as u64,
                        Err(e) => {
                            return match e.kind() {
                                io::ErrorKind::WouldBlock => {
                                    w.need_write()?;
                                    Ok(false)
                                }
                                _ => Err(e),
                            };
                        }
                    }
                    buf.compact();
                    if buf.is_empty() {
                        w.no_need_write()?;
                    } else {
                        w.need_write()?;
                        return Ok(false);
                    }
                }
            }
            None => ::unreachable(),
        }

        Ok(done_read)
    }
}

impl<R, W> Future for Copy<R, W>
where
    R: AsyncRead,
    W: AsyncWrite,
{
    type Item = (u64, R, W);
    type Error = io::Error;

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        match self.poll_copy() {
            Ok(false) => Ok(Async::NotReady),
            Ok(true) => {
                match mem::replace(&mut self.r, ReadState::Dead) {
                    ReadState::Done(r) => {
                        match mem::replace(&mut self.w, None) {
                            Some(w) => Ok(Async::Ready((self.len, r, w))),
                            _ => ::unreachable(),
                        }
                    }
                    _ => ::unreachable(),
                }
            }
            Err(e) => {
                self.r = ReadState::Dead;
                Err(e)
            }
        }
    }
}