hyper-client-pool 0.6.0

Pooled Hyper Async Clients
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

use crate::deliverable::Deliverable;
use hyper::client::{
    connect::{
        dns::{GaiResolver, Name},
        Connect,
    },
    HttpConnector,
};
use hyper_tls::HttpsConnector;
use std::sync::{Arc, RwLock};
use std::{marker::PhantomData, net::SocketAddr};
use tower_service::Service;

use super::Pool;
use crate::config::Config;
use crate::error::SpawnError;
use crate::executor::TransactionCounter;

pub type PoolConnector<R> = HttpsConnector<HttpConnector<R>>;

/// A trait used to wrap the PoolConnector used by default into a
/// different type that implements Connect
pub trait ConnectorAdaptor<R> {
    type Connect: Connect;

    fn wrap(connector: PoolConnector<R>) -> Self::Connect;
}

/// A trait used to create a DNS resolver. DefaultResolver uses the default DNS
/// resolver built into hyper.
pub trait CreateResolver {
    type Resolver: 'static
        + Clone
        + Send
        + Sync
        + Service<Name, Error = Self::Error, Future = Self::Future, Response = Self::Response>;

    type Error: 'static + Send + Sync + std::error::Error;
    type Future: Send + std::future::Future<Output = Result<Self::Response, Self::Error>>;
    type Response: Iterator<Item = SocketAddr>;

    fn create_resolver() -> Self::Resolver;
}

struct DefaultResolver;

impl CreateResolver for DefaultResolver {
    type Resolver = GaiResolver;

    type Error = <Self::Resolver as Service<Name>>::Error;
    type Future = <Self::Resolver as Service<Name>>::Future;
    type Response = <Self::Resolver as Service<Name>>::Response;

    fn create_resolver() -> Self::Resolver {
        GaiResolver::new()
    }
}

/// Default type that implemented ConnectorAdaptor, just passes through the connector
pub struct DefaultConnectorAdapator;

pub struct PoolBuilder<D: Deliverable> {
    pub(in crate::pool) config: Config,
    pub(in crate::pool) transaction_counters: Option<Arc<RwLock<Vec<TransactionCounter>>>>,

    _d: PhantomData<D>,
}

impl<D: Deliverable> PoolBuilder<D> {
    pub(in crate::pool) fn new(config: Config) -> PoolBuilder<D> {
        PoolBuilder {
            config,
            transaction_counters: None,

            _d: PhantomData,
        }
    }

    pub fn build(self) -> Result<Pool<D>, SpawnError> {
        self.build_with_adaptor::<DefaultConnectorAdapator>()
    }

    /// Create the pool with a ConnectorAdaptor, a type that is used to
    /// wrap the hyper::Client's connector
    pub fn build_with_adaptor<A>(self) -> Result<Pool<D>, SpawnError>
    where
        A: ConnectorAdaptor<GaiResolver>,
        A::Connect: 'static + Clone + Send + Sync,
    {
        self.build_with_adaptor_and_resolver::<A, DefaultResolver>()
    }

    /// Create the pool with a ConnectorAdaptor, a type that is used to
    /// wrap the hyper::Client's connector
    pub fn build_with_adaptor_and_resolver<A, CR>(self) -> Result<Pool<D>, SpawnError>
    where
        A: ConnectorAdaptor<CR::Resolver>,
        A::Connect: 'static + Clone + Send + Sync,
        CR: CreateResolver,
        CR::Resolver: 'static + Clone + Send + Sync + Service<Name>,
        CR::Error: 'static + Send + Sync + std::error::Error,
        CR::Future: Send + std::future::Future<Output = Result<CR::Response, CR::Error>>,
        CR::Response: Iterator<Item = SocketAddr>,
    {
        Pool::new::<A, CR>(self)
    }

    /// Pass in an synchronized Vec<Weak<WeakCounter>> that will be populated
    /// with transaction counters for each of the workers spawned.
    ///
    /// You can check that the WeakCounter is still valid by ensuring that the
    /// Arc::strong_count on the Weak reference
    pub fn transaction_counters(mut self, value: Arc<RwLock<Vec<TransactionCounter>>>) -> Self {
        self.transaction_counters = Some(value);
        self
    }
}

impl<R> ConnectorAdaptor<R> for DefaultConnectorAdapator
where
    R: 'static + Clone + Send + Sync + Service<Name>,
    R::Error: 'static + Send + Sync + std::error::Error,
    R::Future: Send + std::future::Future<Output = Result<R::Response, R::Error>>,
    R::Response: Iterator<Item = SocketAddr>,
{
    type Connect = PoolConnector<R>;

    fn wrap(connector: PoolConnector<R>) -> Self::Connect {
        connector
    }
}