ntex_h2/server/
service.rs

1use std::{fmt, future::Future, future::poll_fn, pin::Pin, rc::Rc};
2
3use ntex_dispatcher::Dispatcher as IoDispatcher;
4use ntex_io::{Filter, Io, IoBoxed};
5use ntex_service::cfg::{Cfg, SharedCfg};
6use ntex_service::{IntoServiceFactory, Service, ServiceCtx, ServiceFactory};
7use ntex_util::{channel::pool, time::timeout_checked};
8
9use crate::control::{Control, ControlAck};
10use crate::{codec::Codec, connection::Connection, default::DefaultControlService};
11use crate::{config::ServiceConfig, consts, dispatcher::Dispatcher, frame, message::Message};
12
13use super::ServerError;
14
15#[derive(Debug)]
16/// Http/2 server factory
17pub struct Server<Pub, Ctl>(ServerInner<Pub, Ctl>);
18
19#[derive(Debug)]
20struct ServerInner<Pub, Ctl> {
21    control: Rc<Ctl>,
22    publish: Rc<Pub>,
23    pool: pool::Pool<()>,
24}
25
26impl<Pub, Ctl> Clone for ServerInner<Pub, Ctl> {
27    fn clone(&self) -> Self {
28        Self {
29            control: self.control.clone(),
30            publish: self.publish.clone(),
31            pool: self.pool.clone(),
32        }
33    }
34}
35
36impl<Pub> Server<Pub, DefaultControlService>
37where
38    Pub: ServiceFactory<Message, SharedCfg, Response = ()> + 'static,
39    Pub::Error: fmt::Debug,
40    Pub::InitError: fmt::Debug,
41{
42    /// Create new instance of Server factory
43    pub fn new(publish: Pub) -> Self {
44        Self(ServerInner {
45            publish: Rc::new(publish),
46            control: Rc::new(DefaultControlService),
47            pool: pool::new(),
48        })
49    }
50}
51
52impl<Pub, Ctl> Server<Pub, Ctl>
53where
54    Ctl: ServiceFactory<Control<Pub::Error>, SharedCfg, Response = ControlAck> + 'static,
55    Ctl::Error: fmt::Debug,
56    Ctl::InitError: fmt::Debug,
57    Pub: ServiceFactory<Message, SharedCfg, Response = ()> + 'static,
58    Pub::Error: fmt::Debug,
59    Pub::InitError: fmt::Debug,
60{
61    /// Service to handle control frames
62    pub fn control<S, F>(&self, service: F) -> Server<Pub, S>
63    where
64        F: IntoServiceFactory<S, Control<Pub::Error>, SharedCfg>,
65        S: ServiceFactory<Control<Pub::Error>, SharedCfg, Response = ControlAck> + 'static,
66        S::Error: fmt::Debug,
67        S::InitError: fmt::Debug,
68    {
69        Server(ServerInner {
70            control: Rc::new(service.into_factory()),
71            publish: self.0.publish.clone(),
72            pool: self.0.pool.clone(),
73        })
74    }
75
76    /// Construct service handler
77    pub fn handler(&self, cfg: SharedCfg) -> ServerHandler<Pub, Ctl> {
78        ServerHandler::new(cfg, self.0.clone())
79    }
80}
81
82impl<Pub, Ctl> ServiceFactory<IoBoxed, SharedCfg> for Server<Pub, Ctl>
83where
84    Ctl: ServiceFactory<Control<Pub::Error>, SharedCfg, Response = ControlAck> + 'static,
85    Ctl::Error: fmt::Debug,
86    Ctl::InitError: fmt::Debug,
87    Pub: ServiceFactory<Message, SharedCfg, Response = ()> + 'static,
88    Pub::Error: fmt::Debug,
89    Pub::InitError: fmt::Debug,
90{
91    type Response = ();
92    type Error = ServerError<()>;
93    type Service = ServerHandler<Pub, Ctl>;
94    type InitError = ();
95
96    async fn create(&self, cfg: SharedCfg) -> Result<Self::Service, Self::InitError> {
97        Ok(ServerHandler::new(cfg, self.0.clone()))
98    }
99}
100
101impl<F, Pub, Ctl> ServiceFactory<Io<F>, SharedCfg> for Server<Pub, Ctl>
102where
103    F: Filter,
104    Ctl: ServiceFactory<Control<Pub::Error>, SharedCfg, Response = ControlAck> + 'static,
105    Ctl::Error: fmt::Debug,
106    Ctl::InitError: fmt::Debug,
107    Pub: ServiceFactory<Message, SharedCfg, Response = ()> + 'static,
108    Pub::Error: fmt::Debug,
109    Pub::InitError: fmt::Debug,
110{
111    type Response = ();
112    type Error = ServerError<()>;
113    type Service = ServerHandler<Pub, Ctl>;
114    type InitError = ();
115
116    async fn create(&self, cfg: SharedCfg) -> Result<Self::Service, Self::InitError> {
117        Ok(ServerHandler::new(cfg, self.0.clone()))
118    }
119}
120
121#[derive(Debug)]
122/// Http2 connections handler
123pub struct ServerHandler<Pub, Ctl> {
124    cfg: Cfg<ServiceConfig>,
125    inner: ServerInner<Pub, Ctl>,
126    shared: SharedCfg,
127}
128
129impl<Pub, Ctl> ServerHandler<Pub, Ctl> {
130    fn new(shared: SharedCfg, inner: ServerInner<Pub, Ctl>) -> Self {
131        let cfg = shared.get();
132        Self { cfg, inner, shared }
133    }
134}
135
136impl<Pub, Ctl> Clone for ServerHandler<Pub, Ctl> {
137    fn clone(&self) -> Self {
138        Self {
139            inner: self.inner.clone(),
140            cfg: self.cfg.clone(),
141            shared: self.shared.clone(),
142        }
143    }
144}
145
146impl<Pub, Ctl> ServerHandler<Pub, Ctl>
147where
148    Ctl: ServiceFactory<Control<Pub::Error>, SharedCfg, Response = ControlAck> + 'static,
149    Ctl::Error: fmt::Debug,
150    Ctl::InitError: fmt::Debug,
151    Pub: ServiceFactory<Message, SharedCfg, Response = ()> + 'static,
152    Pub::Error: fmt::Debug,
153    Pub::InitError: fmt::Debug,
154{
155    pub async fn run(&self, io: IoBoxed) -> Result<(), ServerError<()>> {
156        let inner = &self.inner;
157
158        let (ctl_srv, pub_srv) = timeout_checked(self.cfg.handshake_timeout, async {
159            read_preface(&io).await?;
160
161            // create publish service
162            let pub_srv = inner
163                .publish
164                .create(self.shared.clone())
165                .await
166                .map_err(|e| {
167                    log::error!("Publish service init error: {e:?}");
168                    ServerError::PublishServiceError
169                })?;
170
171            // create control service
172            let ctl_srv = inner
173                .control
174                .create(self.shared.clone())
175                .await
176                .map_err(|e| {
177                    log::error!("Control service init error: {e:?}");
178                    ServerError::ControlServiceError
179                })?;
180
181            Ok::<_, ServerError<()>>((ctl_srv, pub_srv))
182        })
183        .await
184        .map_err(|()| ServerError::HandshakeTimeout)??;
185
186        // create h2 codec
187        let codec = Codec::default();
188        codec.set_max_headers(self.cfg.max_headers);
189
190        let con = Connection::new(
191            true,
192            io.get_ref(),
193            codec.clone(),
194            self.cfg.clone(),
195            true,
196            false,
197            self.inner.pool.clone(),
198        );
199        let con2 = con.clone();
200
201        // start protocol dispatcher
202        let mut fut = IoDispatcher::new(io, codec, Dispatcher::new(con, ctl_srv, pub_srv));
203        poll_fn(|cx| {
204            if con2.config().is_shutdown() {
205                con2.disconnect_when_ready();
206            }
207            Pin::new(&mut fut).poll(cx)
208        })
209        .await
210        .map_err(|()| ServerError::Dispatcher)
211    }
212}
213
214impl<Pub, Ctl> Service<IoBoxed> for ServerHandler<Pub, Ctl>
215where
216    Ctl: ServiceFactory<Control<Pub::Error>, SharedCfg, Response = ControlAck> + 'static,
217    Ctl::Error: fmt::Debug,
218    Ctl::InitError: fmt::Debug,
219    Pub: ServiceFactory<Message, SharedCfg, Response = ()> + 'static,
220    Pub::Error: fmt::Debug,
221    Pub::InitError: fmt::Debug,
222{
223    type Response = ();
224    type Error = ServerError<()>;
225
226    async fn call(
227        &self,
228        io: IoBoxed,
229        _: ServiceCtx<'_, Self>,
230    ) -> Result<Self::Response, Self::Error> {
231        self.run(io).await
232    }
233}
234
235impl<F, Pub, Ctl> Service<Io<F>> for ServerHandler<Pub, Ctl>
236where
237    F: Filter,
238    Ctl: ServiceFactory<Control<Pub::Error>, SharedCfg, Response = ControlAck> + 'static,
239    Ctl::Error: fmt::Debug,
240    Ctl::InitError: fmt::Debug,
241    Pub: ServiceFactory<Message, SharedCfg, Response = ()> + 'static,
242    Pub::Error: fmt::Debug,
243    Pub::InitError: fmt::Debug,
244{
245    type Response = ();
246    type Error = ServerError<()>;
247
248    async fn call(
249        &self,
250        req: Io<F>,
251        _: ServiceCtx<'_, Self>,
252    ) -> Result<Self::Response, Self::Error> {
253        self.run(req.into()).await
254    }
255}
256
257async fn read_preface(io: &IoBoxed) -> Result<(), ServerError<()>> {
258    let mut buf = [0; consts::PREFACE_LEN];
259    io.read(&mut buf).await?;
260
261    if buf == consts::PREFACE {
262        log::debug!("Preface has been received");
263        Ok(())
264    } else {
265        log::trace!("read_preface: invalid preface {buf:?}");
266        Err(ServerError::<()>::Frame(frame::FrameError::InvalidPreface))
267    }
268}
269
270/// Handle io object.
271pub async fn handle_one<Pub, Ctl>(
272    io: IoBoxed,
273    pub_svc: Pub,
274    ctl_svc: Ctl,
275) -> Result<(), ServerError<()>>
276where
277    Ctl: Service<Control<Pub::Error>, Response = ControlAck> + 'static,
278    Ctl::Error: fmt::Debug,
279    Pub: Service<Message, Response = ()> + 'static,
280    Pub::Error: fmt::Debug,
281{
282    let config: Cfg<ServiceConfig> = io.shared().get();
283
284    // read preface
285    timeout_checked(config.handshake_timeout, async { read_preface(&io).await })
286        .await
287        .map_err(|()| ServerError::HandshakeTimeout)??;
288
289    // create h2 codec
290    let codec = Codec::default();
291    codec.set_max_headers(config.max_headers);
292    let con = Connection::new(
293        true,
294        io.get_ref(),
295        codec.clone(),
296        config,
297        true,
298        false,
299        pool::new(),
300    );
301    let con2 = con.clone();
302
303    // start protocol dispatcher
304    let mut fut = IoDispatcher::new(io, codec, Dispatcher::new(con, ctl_svc, pub_svc));
305
306    poll_fn(|cx| {
307        if con2.config().is_shutdown() {
308            con2.disconnect_when_ready();
309        }
310        Pin::new(&mut fut).poll(cx)
311    })
312    .await
313    .map_err(|()| ServerError::Dispatcher)
314}
ntex_h2/server/service.rs

ntex_h2/server/
service.rs
