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

use std::fmt;

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::default::DefaultControlService;
use crate::dispatcher::Dispatcher;
use crate::{
    codec::Codec, config::Config, connection::Connection, Message, OperationError, Stream,
};

/// Http2 client
#[derive(Clone)]
pub struct Client {
    con: Connection,
    scheme: Scheme,
    authority: ByteString,
}

/// Http2 client connection
pub struct ClientConnection(IoBoxed, Connection);

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

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.con
            .send_request(
                self.scheme.clone(),
                self.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.con.can_create_new_stream()
    }

    #[inline]
    /// Set client's secure state
    pub fn set_scheme(&mut self, scheme: Scheme) {
        self.scheme = scheme;
    }

    /// Set client's authority
    pub fn set_authority(&mut self, authority: ByteString) {
        self.authority = authority;
    }

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

    #[inline]
    /// Gracefully close connection
    pub fn close(&self) {
        self.con.state().io.close()
    }

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

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

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

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

        ClientConnection(io, con)
    }

    #[inline]
    /// Get client
    pub fn client(&self) -> Client {
        Client {
            con: self.1.clone(),
            scheme: Scheme::HTTP,
            authority: ByteString::new(),
        }
    }

    /// 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.1.clone(),
            DefaultControlService,
            service.into_service(),
        );

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