tk-pool 0.5.3

Generic connection pool for tokio
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

extern crate tk_pool;
extern crate tk_http;
extern crate futures;
extern crate abstract_ns;
extern crate futures_cpupool;
extern crate tokio_core;
extern crate env_logger;
extern crate ns_std_threaded;
extern crate ns_router;
extern crate log;

use std::env;
use std::time::Duration;

use abstract_ns::HostResolve;
use futures::{Future, Sink};
use futures::future::join_all;
use tk_pool::{pool_for};
use tk_http::client::{Proto, Config as HConfig, Client, Error};
use ns_router::SubscribeExt;

fn main() {
    if let Err(_) = env::var("RUST_LOG") {
        env::set_var("RUST_LOG", "warn");
    }
    env_logger::init();

    let mut lp = tokio_core::reactor::Core::new().unwrap();
    let h1 = lp.handle();
    let h2 = lp.handle();

    let ns = ns_router::Router::from_config(&ns_router::Config::new()
        .set_fallthrough(ns_std_threaded::ThreadedResolver::new()
            .null_service_resolver()
            .interval_subscriber(Duration::new(1, 0), &h1))
        .done(),
        &lp.handle());

    let connection_config = HConfig::new()
        .inflight_request_limit(1)
        .done();
    let mut pool =
        pool_for(move |addr| Proto::connect_tcp(addr, &connection_config, &h2))
        .connect_to(ns.subscribe_many(&["httpbin.org"], 80))
        .lazy_uniform_connections(2)
        .with_queue_size(16)
        .spawn_on(&lp.handle())
        // This is needed for Client trait, (i.e. so that `.fetch_url()` works)
        // May be fixed in tk-http in future
        .sink_map_err(|_| Error::custom("Can't send request"));

    // Alternative config (not implemented)
    // let mut pool =
    //     pool_for(|addr| Proto::connect_tcp(addr, &connection_config, &h2))
    //     .with_config_stream(ns, once_and_wait(Config::new()
    //         .set_name(&["httpbin.org:80"])
    //         .lazy_uniform_connections(2)
    //         .with_queue_size(10)))
    //     .spawn_on(&lp.handle());

    println!("We will send 16 requests over 1 connection per ip. \
              Each requests hangs for 5 seconds at the server side \
              (a feature of httpbin.org). Since httpbin nowadays has \
              many IPs, expect script to finish in 5 or 10 seconds \
              with first response coming in 5 seconds.");

    lp.run(futures::lazy(|| {
        join_all((0..16).map(move |_| {
            pool
            .fetch_url("http://httpbin.org/delay/5")
            .map(|r| {
                println!("Received {} bytes", r.body().len())
            })
        }))
    })).unwrap();
}