tower-util 0.1.0

Utilities for working with `Service`. 
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
142
143

use futures::{
    self,
    future::{ok, FutureResult},
    stream, Async, Poll, Stream,
};
use std::{cell::Cell, rc::Rc};
use tower::ServiceExt;
use tower_service::*;
use tower_test::{assert_request_eq, mock};

type Error = Box<dyn std::error::Error + Send + Sync>;

#[derive(Debug, Eq, PartialEq)]
struct Srv {
    admit: Rc<Cell<bool>>,
    count: Rc<Cell<usize>>,
}
impl Service<&'static str> for Srv {
    type Response = &'static str;
    type Error = Error;
    type Future = FutureResult<Self::Response, Error>;

    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
        if !self.admit.get() {
            return Ok(Async::NotReady);
        }

        self.admit.set(false);
        Ok(Async::Ready(()))
    }

    fn call(&mut self, req: &'static str) -> Self::Future {
        self.count.set(self.count.get() + 1);
        ok(req)
    }
}

macro_rules! assert_ready {
    ($e:expr) => {{
        match $e {
            Ok(futures::Async::Ready(v)) => v,
            Ok(_) => panic!("not ready"),
            Err(e) => panic!("error = {:?}", e),
        }
    }};
}

macro_rules! assert_not_ready {
    ($e:expr) => {{
        match $e {
            Ok(futures::Async::NotReady) => {}
            Ok(futures::Async::Ready(v)) => panic!("ready; value = {:?}", v),
            Err(e) => panic!("error = {:?}", e),
        }
    }};
}

#[test]
fn ordered() {
    let mut mock = tokio_mock_task::MockTask::new();

    let admit = Rc::new(Cell::new(false));
    let count = Rc::new(Cell::new(0));
    let srv = Srv {
        count: count.clone(),
        admit: admit.clone(),
    };
    let (tx, rx) = futures::unsync::mpsc::unbounded();
    let mut ca = srv.call_all(rx.map_err(|_| "nope"));

    assert_not_ready!(mock.enter(|| ca.poll()));
    tx.unbounded_send("one").unwrap();
    mock.is_notified();
    assert_not_ready!(mock.enter(|| ca.poll()));
    admit.set(true);
    let v = assert_ready!(mock.enter(|| ca.poll()));
    assert_eq!(v, Some("one"));
    assert_not_ready!(mock.enter(|| ca.poll()));
    admit.set(true);
    tx.unbounded_send("two").unwrap();
    mock.is_notified();
    tx.unbounded_send("three").unwrap();
    let v = assert_ready!(mock.enter(|| ca.poll()));
    assert_eq!(v, Some("two"));
    assert_not_ready!(mock.enter(|| ca.poll()));
    admit.set(true);
    let v = assert_ready!(mock.enter(|| ca.poll()));
    assert_eq!(v, Some("three"));
    admit.set(true);
    assert_not_ready!(mock.enter(|| ca.poll()));
    admit.set(true);
    tx.unbounded_send("four").unwrap();
    mock.is_notified();
    let v = assert_ready!(mock.enter(|| ca.poll()));
    assert_eq!(v, Some("four"));
    assert_not_ready!(mock.enter(|| ca.poll()));

    // need to be ready since impl doesn't know it'll get EOF
    admit.set(true);

    // When we drop the request stream, CallAll should return None.
    drop(tx);
    mock.is_notified();
    let v = assert_ready!(mock.enter(|| ca.poll()));
    assert!(v.is_none());
    assert_eq!(count.get(), 4);

    // We should also be able to recover the wrapped Service.
    assert_eq!(
        ca.into_inner(),
        Srv {
            count: count.clone(),
            admit
        }
    );
}

#[test]
fn unordered() {
    let (mock, mut handle) = mock::pair::<_, &'static str>();
    let mut task = tokio_mock_task::MockTask::new();
    let requests = stream::iter_ok::<_, Error>(&["one", "two"]);

    let mut svc = mock.call_all(requests).unordered();
    assert_not_ready!(task.enter(|| svc.poll()));

    let resp1 = assert_request_eq!(handle, &"one");
    let resp2 = assert_request_eq!(handle, &"two");

    resp2.send_response("resp 1");

    let v = assert_ready!(task.enter(|| svc.poll()));
    assert_eq!(v, Some("resp 1"));
    assert_not_ready!(task.enter(|| svc.poll()));

    resp1.send_response("resp 2");

    let v = assert_ready!(task.enter(|| svc.poll()));
    assert_eq!(v, Some("resp 2"));

    let v = assert_ready!(task.enter(|| svc.poll()));
    assert!(v.is_none());
}