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    inner: ServerInner<Pub, Ctl>,
125    cfg: Cfg<ServiceConfig>,
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, shared, inner }
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,
141            shared: self.shared,
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.publish.create(self.shared).await.map_err(|e| {
163                log::error!("Publish service init error: {e:?}");
164                ServerError::PublishServiceError
165            })?;
166
167            // create control service
168            let ctl_srv = inner.control.create(self.shared).await.map_err(|e| {
169                log::error!("Control service init error: {e:?}");
170                ServerError::ControlServiceError
171            })?;
172
173            Ok::<_, ServerError<()>>((ctl_srv, pub_srv))
174        })
175        .await
176        .map_err(|_| ServerError::HandshakeTimeout)??;
177
178        // create h2 codec
179        let codec = Codec::default();
180        let con = Connection::new(
181            true,
182            io.get_ref(),
183            codec.clone(),
184            self.cfg,
185            true,
186            false,
187            self.inner.pool.clone(),
188        );
189        let con2 = con.clone();
190
191        // start protocol dispatcher
192        let mut fut = IoDispatcher::new(io, codec, Dispatcher::new(con, ctl_srv, pub_srv));
193        poll_fn(|cx| {
194            if con2.config().is_shutdown() {
195                con2.disconnect_when_ready();
196            }
197            Pin::new(&mut fut).poll(cx)
198        })
199        .await
200        .map_err(|_| ServerError::Dispatcher)
201    }
202}
203
204impl<Pub, Ctl> Service<IoBoxed> for ServerHandler<Pub, Ctl>
205where
206    Ctl: ServiceFactory<Control<Pub::Error>, SharedCfg, Response = ControlAck> + 'static,
207    Ctl::Error: fmt::Debug,
208    Ctl::InitError: fmt::Debug,
209    Pub: ServiceFactory<Message, SharedCfg, Response = ()> + 'static,
210    Pub::Error: fmt::Debug,
211    Pub::InitError: fmt::Debug,
212{
213    type Response = ();
214    type Error = ServerError<()>;
215
216    async fn call(
217        &self,
218        io: IoBoxed,
219        _: ServiceCtx<'_, Self>,
220    ) -> Result<Self::Response, Self::Error> {
221        self.run(io).await
222    }
223}
224
225impl<F, Pub, Ctl> Service<Io<F>> for ServerHandler<Pub, Ctl>
226where
227    F: Filter,
228    Ctl: ServiceFactory<Control<Pub::Error>, SharedCfg, Response = ControlAck> + 'static,
229    Ctl::Error: fmt::Debug,
230    Ctl::InitError: fmt::Debug,
231    Pub: ServiceFactory<Message, SharedCfg, Response = ()> + 'static,
232    Pub::Error: fmt::Debug,
233    Pub::InitError: fmt::Debug,
234{
235    type Response = ();
236    type Error = ServerError<()>;
237
238    async fn call(
239        &self,
240        req: Io<F>,
241        _: ServiceCtx<'_, Self>,
242    ) -> Result<Self::Response, Self::Error> {
243        self.run(req.into()).await
244    }
245}
246
247async fn read_preface(io: &IoBoxed) -> Result<(), ServerError<()>> {
248    loop {
249        let ready = io.with_read_buf(|buf| {
250            if buf.len() >= consts::PREFACE.len() {
251                if buf[..consts::PREFACE.len()] == consts::PREFACE {
252                    buf.advance_to(consts::PREFACE.len());
253                    Ok(true)
254                } else {
255                    log::trace!("read_preface: invalid preface {buf:?}");
256                    Err(ServerError::<()>::Frame(frame::FrameError::InvalidPreface))
257                }
258            } else {
259                Ok(false)
260            }
261        })?;
262
263        if ready {
264            log::debug!("Preface has been received");
265            return Ok::<_, ServerError<_>>(());
266        } else {
267            io.read_ready()
268                .await?
269                .ok_or(ServerError::Disconnected(None))?;
270        }
271    }
272}
273
274/// Handle io object.
275pub async fn handle_one<Pub, Ctl>(
276    io: IoBoxed,
277    pub_svc: Pub,
278    ctl_svc: Ctl,
279) -> Result<(), ServerError<()>>
280where
281    Ctl: Service<Control<Pub::Error>, Response = ControlAck> + 'static,
282    Ctl::Error: fmt::Debug,
283    Pub: Service<Message, Response = ()> + 'static,
284    Pub::Error: fmt::Debug,
285{
286    let config: Cfg<ServiceConfig> = io.shared().get();
287
288    // read preface
289    timeout_checked(config.handshake_timeout, async { read_preface(&io).await })
290        .await
291        .map_err(|_| ServerError::HandshakeTimeout)??;
292
293    // create h2 codec
294    let codec = Codec::default();
295    let con = Connection::new(
296        true,
297        io.get_ref(),
298        codec.clone(),
299        config,
300        true,
301        false,
302        pool::new(),
303    );
304    let con2 = con.clone();
305
306    // start protocol dispatcher
307    let mut fut = IoDispatcher::new(io, codec, Dispatcher::new(con, ctl_svc, pub_svc));
308
309    poll_fn(|cx| {
310        if con2.config().is_shutdown() {
311            con2.disconnect_when_ready();
312        }
313        Pin::new(&mut fut).poll(cx)
314    })
315    .await
316    .map_err(|_| ServerError::Dispatcher)
317}
ntex_h2/server/service.rs

ntex_h2/server/
service.rs
