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

use std::{fmt, rc::Rc};

use ntex_bytes::ByteString;
use ntex_http::{uri::Scheme, HeaderMap, Method};
use ntex_io::{Dispatcher as IoDispatcher, IoBoxed, OnDisconnect};
use ntex_service::{IntoService, Service};
use ntex_util::time::Seconds;

use crate::connection::Connection;
use crate::default::DefaultControlService;
use crate::dispatcher::Dispatcher;
use crate::{codec::Codec, config::Config, Message, OperationError, Stream};

/// Http2 client
#[derive(Clone)]
pub struct Client(Rc<ClientRef>);

/// Http2 client
struct ClientRef {
    con: Connection,
    authority: ByteString,
}

/// Http2 client connection
pub struct ClientConnection {
    io: IoBoxed,
    client: Rc<ClientRef>,
}

impl Client {
    #[inline]
    /// Send request to the peer
    pub async fn send_request(
        &self,
        method: Method,
        path: ByteString,
        headers: HeaderMap,
        eof: bool,
    ) -> Result<Stream, OperationError> {
        self.0
            .con
            .send_request(self.0.authority.clone(), method, path, headers, eof)
            .await
    }

    #[inline]
    /// Check if client is allowed to send new request
    ///
    /// Readiness depends on number of opened streams and max concurrency setting
    pub fn is_ready(&self) -> bool {
        self.0.con.can_create_new_stream()
    }

    #[doc(hidden)]
    #[inline]
    /// Set client's secure state
    pub fn set_scheme(&self, scheme: Scheme) {
        if scheme == Scheme::HTTPS {
            self.0.con.set_secure(true)
        } else {
            self.0.con.set_secure(false)
        }
    }

    #[doc(hidden)]
    /// Set client's authority
    pub fn set_authority(&self, _: ByteString) {}

    #[inline]
    /// Check client readiness
    ///
    /// Client is ready when it is possible to start new stream
    pub async fn ready(&self) -> Result<(), OperationError> {
        self.0.con.ready().await
    }

    #[inline]
    /// Gracefully close connection
    pub fn close(&self) {
        log::debug!("Closing client");
        self.0.con.disconnect_when_ready()
    }

    #[inline]
    /// Close connection
    pub fn force_close(&self) {
        self.0.con.close()
    }

    #[inline]
    /// Check if connection is closed
    pub fn is_closed(&self) -> bool {
        self.0.con.is_closed()
    }

    #[inline]
    /// Notify when connection get closed
    pub fn on_disconnect(&self) -> OnDisconnect {
        self.0.con.state().io.on_disconnect()
    }
}

impl Drop for Client {
    fn drop(&mut self) {
        if Rc::strong_count(&self.0) == 1 {
            log::debug!("Last h2 client has been dropped, disconnecting");
            self.0.con.disconnect_when_ready()
        }
    }
}

impl fmt::Debug for Client {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ntex_h2::Client")
            .field("authority", &self.0.authority)
            .field("connection", &self.0.con)
            .finish()
    }
}

impl fmt::Debug for ClientRef {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ntex_h2::Client")
            .field("authority", &self.authority)
            .field("connection", &self.con)
            .finish()
    }
}

impl ClientConnection {
    /// Construct new `ClientConnection` instance.
    pub fn new<T>(io: T, config: Config) -> Self
    where
        IoBoxed: From<T>,
    {
        Self::with_params(io, config, false, ByteString::new())
    }

    /// Construct new `ClientConnection` instance.
    pub fn with_params<T>(io: T, config: Config, secure: bool, authority: ByteString) -> Self
    where
        IoBoxed: From<T>,
    {
        let io: IoBoxed = io.into();
        let codec = Codec::default();
        let con = Connection::new(io.get_ref(), codec, config, false);
        con.set_secure(secure);

        ClientConnection {
            io,
            client: Rc::new(ClientRef { con, authority }),
        }
    }

    #[inline]
    /// Get client
    pub fn client(&self) -> Client {
        Client(self.client.clone())
    }

    /// Run client with provided control messages handler
    pub async fn start<F, S>(self, service: F) -> Result<(), ()>
    where
        F: IntoService<S, Message> + 'static,
        S: Service<Message, Response = ()> + 'static,
        S::Error: fmt::Debug,
    {
        let disp = Dispatcher::new(
            self.client.con.clone(),
            DefaultControlService,
            service.into_service(),
        );

        IoDispatcher::new(self.io, self.client.con.state().codec.clone(), disp)
            .keepalive_timeout(Seconds::ZERO)
            .disconnect_timeout(self.client.con.config().disconnect_timeout.get())
            .await
    }
}

impl fmt::Debug for ClientConnection {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ntex_h2::ClientConnection")
            .field("authority", &self.client.authority)
            .field("config", &self.client.con.config())
            .finish()
    }
}