[go: up one dir, main page]

tower 0.4.13

Tower is a library of modular and reusable components for building robust clients and servers.
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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190

use crate::load;
use futures_util::pin_mut;
use tokio_test::{assert_pending, assert_ready, assert_ready_ok, task};
use tower_test::{assert_request_eq, mock};

use super::*;

#[tokio::test]
async fn basic() {
    // start the pool
    let (mock, handle) = mock::pair::<(), load::Constant<mock::Mock<(), &'static str>, usize>>();
    pin_mut!(handle);

    let mut pool = mock::Spawn::new(Builder::new().build(mock, ()));
    assert_pending!(pool.poll_ready());

    // give the pool a backing service
    let (svc1_m, svc1) = mock::pair();
    pin_mut!(svc1);

    assert_request_eq!(handle, ()).send_response(load::Constant::new(svc1_m, 0));
    assert_ready_ok!(pool.poll_ready());

    // send a request to the one backing service
    let mut fut = task::spawn(pool.call(()));

    assert_pending!(fut.poll());
    assert_request_eq!(svc1, ()).send_response("foobar");
    assert_eq!(assert_ready_ok!(fut.poll()), "foobar");
}

#[tokio::test]
async fn high_load() {
    // start the pool
    let (mock, handle) = mock::pair::<(), load::Constant<mock::Mock<(), &'static str>, usize>>();
    pin_mut!(handle);

    let pool = Builder::new()
        .urgency(1.0) // so _any_ Pending will add a service
        .underutilized_below(0.0) // so no Ready will remove a service
        .max_services(Some(2))
        .build(mock, ());
    let mut pool = mock::Spawn::new(pool);
    assert_pending!(pool.poll_ready());

    // give the pool a backing service
    let (svc1_m, svc1) = mock::pair();
    pin_mut!(svc1);

    svc1.allow(1);
    assert_request_eq!(handle, ()).send_response(load::Constant::new(svc1_m, 0));
    assert_ready_ok!(pool.poll_ready());

    // make the one backing service not ready
    let mut fut1 = task::spawn(pool.call(()));

    // if we poll_ready again, pool should notice that load is increasing
    // since urgency == 1.0, it should immediately enter high load
    assert_pending!(pool.poll_ready());
    // it should ask the maker for another service, so we give it one
    let (svc2_m, svc2) = mock::pair();
    pin_mut!(svc2);

    svc2.allow(1);
    assert_request_eq!(handle, ()).send_response(load::Constant::new(svc2_m, 0));

    // the pool should now be ready again for one more request
    assert_ready_ok!(pool.poll_ready());
    let mut fut2 = task::spawn(pool.call(()));

    assert_pending!(pool.poll_ready());

    // the pool should _not_ try to add another service
    // sicen we have max_services(2)
    assert_pending!(handle.as_mut().poll_request());

    // let see that each service got one request
    assert_request_eq!(svc1, ()).send_response("foo");
    assert_request_eq!(svc2, ()).send_response("bar");
    assert_eq!(assert_ready_ok!(fut1.poll()), "foo");
    assert_eq!(assert_ready_ok!(fut2.poll()), "bar");
}

#[tokio::test]
async fn low_load() {
    // start the pool
    let (mock, handle) = mock::pair::<(), load::Constant<mock::Mock<(), &'static str>, usize>>();
    pin_mut!(handle);

    let pool = Builder::new()
        .urgency(1.0) // so any event will change the service count
        .build(mock, ());

    let mut pool = mock::Spawn::new(pool);

    assert_pending!(pool.poll_ready());

    // give the pool a backing service
    let (svc1_m, svc1) = mock::pair();
    pin_mut!(svc1);

    svc1.allow(1);
    assert_request_eq!(handle, ()).send_response(load::Constant::new(svc1_m, 0));
    assert_ready_ok!(pool.poll_ready());

    // cycling a request should now work
    let mut fut = task::spawn(pool.call(()));

    assert_request_eq!(svc1, ()).send_response("foo");
    assert_eq!(assert_ready_ok!(fut.poll()), "foo");
    // and pool should now not be ready (since svc1 isn't ready)
    // it should immediately try to add another service
    // which we give it
    assert_pending!(pool.poll_ready());
    let (svc2_m, svc2) = mock::pair();
    pin_mut!(svc2);

    svc2.allow(1);
    assert_request_eq!(handle, ()).send_response(load::Constant::new(svc2_m, 0));
    // pool is now ready
    // which (because of urgency == 1.0) should immediately cause it to drop a service
    // it'll drop svc1, so it'll still be ready
    assert_ready_ok!(pool.poll_ready());
    // and even with another ready, it won't drop svc2 since its now the only service
    assert_ready_ok!(pool.poll_ready());

    // cycling a request should now work on svc2
    let mut fut = task::spawn(pool.call(()));

    assert_request_eq!(svc2, ()).send_response("foo");
    assert_eq!(assert_ready_ok!(fut.poll()), "foo");

    // and again (still svc2)
    svc2.allow(1);
    assert_ready_ok!(pool.poll_ready());
    let mut fut = task::spawn(pool.call(()));

    assert_request_eq!(svc2, ()).send_response("foo");
    assert_eq!(assert_ready_ok!(fut.poll()), "foo");
}

#[tokio::test]
async fn failing_service() {
    // start the pool
    let (mock, handle) = mock::pair::<(), load::Constant<mock::Mock<(), &'static str>, usize>>();
    pin_mut!(handle);

    let pool = Builder::new()
        .urgency(1.0) // so _any_ Pending will add a service
        .underutilized_below(0.0) // so no Ready will remove a service
        .build(mock, ());

    let mut pool = mock::Spawn::new(pool);

    assert_pending!(pool.poll_ready());

    // give the pool a backing service
    let (svc1_m, svc1) = mock::pair();
    pin_mut!(svc1);

    svc1.allow(1);
    assert_request_eq!(handle, ()).send_response(load::Constant::new(svc1_m, 0));
    assert_ready_ok!(pool.poll_ready());

    // one request-response cycle
    let mut fut = task::spawn(pool.call(()));

    assert_request_eq!(svc1, ()).send_response("foo");
    assert_eq!(assert_ready_ok!(fut.poll()), "foo");

    // now make svc1 fail, so it has to be removed
    svc1.send_error("ouch");
    // polling now should recognize the failed service,
    // try to create a new one, and then realize the maker isn't ready
    assert_pending!(pool.poll_ready());
    // then we release another service
    let (svc2_m, svc2) = mock::pair();
    pin_mut!(svc2);

    svc2.allow(1);
    assert_request_eq!(handle, ()).send_response(load::Constant::new(svc2_m, 0));

    // the pool should now be ready again
    assert_ready_ok!(pool.poll_ready());
    // and a cycle should work (and go through svc2)
    let mut fut = task::spawn(pool.call(()));

    assert_request_eq!(svc2, ()).send_response("bar");
    assert_eq!(assert_ready_ok!(fut.poll()), "bar");
}