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
use std::marker::PhantomData;
use futures::{Future, Async, Sink, AsyncSink};
use uniform::{FutureOk, FutureErr};
use uniform::chan::{Action, Helper};
pub(in uniform) struct SinkFuture<S, E>
where S: Sink,
{
sink: S,
task: Helper<S::SinkItem>,
phantom: PhantomData<*const E>,
}
impl<S: Sink, E> SinkFuture<S, E> {
pub fn new(sink: S, task: Helper<S::SinkItem>)
-> SinkFuture<S, E>
{
SinkFuture { sink, task, phantom: PhantomData }
}
}
impl<S: Sink, E> Future for SinkFuture<S, E> {
type Item = FutureOk<S>;
type Error = FutureErr<E, S::SinkError>;
fn poll(&mut self) -> Result<Async<Self::Item>, Self::Error> {
match self.task.take() {
Action::StartSend(item) => match self.sink.start_send(item) {
Ok(AsyncSink::Ready) => {
// We need to flush data immediately because there is
// no way to schedule a wakeup on poll_complete of parent
// schedule
//
// TODO(tailhook) we might want to fix it by wrapping the
// future into a refcell and calling
// start_send manually instead of through
// futures unordered.
match self.sink.poll_complete() {
Ok(_) => {
// By contract there is no difference in Ready and
// NotReady I.e. both of them may mean that another
// element can be pushed now. They only distinquish
// whether there is something left in the buffer.
self.task.requeue();
Ok(Async::NotReady)
}
Err(e) => {
self.task.closed();
Err(FutureErr::Disconnected(self.task.addr(), e))
}
}
}
Ok(AsyncSink::NotReady(item)) => {
self.task.backpressure(item);
Ok(Async::NotReady)
}
Err(e) => {
self.task.closed();
Err(FutureErr::Disconnected(self.task.addr(), e))
}
}
Action::Poll => match self.sink.poll_complete() {
Ok(_) => {
// By contract there is no difference in Ready and NotReady
// I.e. both of them may mean that another element can
// be pushed now. They only distinquish whether there is
// something left in the buffer.
self.task.requeue();
Ok(Async::NotReady)
}
Err(e) => {
self.task.closed();
Err(FutureErr::Disconnected(self.task.addr(), e))
}
}
Action::Close => match self.sink.close() {
Ok(Async::Ready(())) => {
Ok(Async::Ready(FutureOk::Closed(self.task.addr())))
}
Ok(Async::NotReady) => Ok(Async::NotReady),
Err(e) => {
self.task.closed();
Err(FutureErr::Disconnected(self.task.addr(), e))
}
}
}
}
}