actix_http/h1/
service.rs

1use std::{
2    fmt,
3    marker::PhantomData,
4    net,
5    rc::Rc,
6    task::{Context, Poll},
7};
8
9use actix_codec::{AsyncRead, AsyncWrite, Framed};
10use actix_service::{
11    fn_service, IntoServiceFactory, Service, ServiceFactory, ServiceFactoryExt as _,
12};
13#[expect(deprecated, reason = "preserve the public ready future type")]
14use actix_utils::future::ready;
15use futures_core::future::LocalBoxFuture;
16use tokio::net::TcpStream;
17use tracing::error;
18
19use super::{codec::Codec, dispatcher::Dispatcher, ExpectHandler, UpgradeHandler};
20use crate::{
21    body::{BoxBody, MessageBody},
22    config::ServiceConfig,
23    error::DispatchError,
24    service::HttpServiceHandler,
25    ConnectCallback, OnConnectData, Request, Response,
26};
27
28/// `ServiceFactory` implementation for HTTP1 transport
29pub struct H1Service<T, S, B, X = ExpectHandler, U = UpgradeHandler> {
30    srv: S,
31    cfg: ServiceConfig,
32    expect: X,
33    upgrade: Option<U>,
34    on_connect_ext: Option<Rc<ConnectCallback<T>>>,
35    _phantom: PhantomData<B>,
36}
37
38impl<T, S, B> H1Service<T, S, B>
39where
40    S: ServiceFactory<Request, Config = ()>,
41    S::Error: Into<Response<BoxBody>>,
42    S::InitError: fmt::Debug,
43    S::Response: Into<Response<B>>,
44    B: MessageBody,
45{
46    /// Create new `HttpService` instance with config.
47    pub(crate) fn with_config<F: IntoServiceFactory<S, Request>>(
48        cfg: ServiceConfig,
49        service: F,
50    ) -> Self {
51        H1Service {
52            cfg,
53            srv: service.into_factory(),
54            expect: ExpectHandler,
55            upgrade: None,
56            on_connect_ext: None,
57            _phantom: PhantomData,
58        }
59    }
60}
61
62impl<S, B, X, U> H1Service<TcpStream, S, B, X, U>
63where
64    S: ServiceFactory<Request, Config = ()>,
65    S::Future: 'static,
66    S::Error: Into<Response<BoxBody>>,
67    S::InitError: fmt::Debug,
68    S::Response: Into<Response<B>>,
69
70    B: MessageBody,
71
72    X: ServiceFactory<Request, Config = (), Response = Request>,
73    X::Future: 'static,
74    X::Error: Into<Response<BoxBody>>,
75    X::InitError: fmt::Debug,
76
77    U: ServiceFactory<(Request, Framed<TcpStream, Codec>), Config = (), Response = ()>,
78    U::Future: 'static,
79    U::Error: fmt::Display + Into<Response<BoxBody>>,
80    U::InitError: fmt::Debug,
81{
82    /// Create simple tcp stream service
83    #[expect(deprecated, reason = "preserve the public ready future type")]
84    pub fn tcp(
85        self,
86    ) -> impl ServiceFactory<TcpStream, Config = (), Response = (), Error = DispatchError, InitError = ()>
87    {
88        fn_service(|io: TcpStream| {
89            let peer_addr = io.peer_addr().ok();
90            ready(Ok((io, peer_addr)))
91        })
92        .and_then(self)
93    }
94}
95
96#[cfg(feature = "openssl")]
97mod openssl {
98    use actix_tls::accept::{
99        openssl::{
100            reexports::{Error as SslError, SslAcceptor},
101            Acceptor, TlsStream,
102        },
103        TlsError,
104    };
105
106    use super::*;
107
108    impl<S, B, X, U> H1Service<TlsStream<TcpStream>, S, B, X, U>
109    where
110        S: ServiceFactory<Request, Config = ()>,
111        S::Future: 'static,
112        S::Error: Into<Response<BoxBody>>,
113        S::InitError: fmt::Debug,
114        S::Response: Into<Response<B>>,
115
116        B: MessageBody,
117
118        X: ServiceFactory<Request, Config = (), Response = Request>,
119        X::Future: 'static,
120        X::Error: Into<Response<BoxBody>>,
121        X::InitError: fmt::Debug,
122
123        U: ServiceFactory<
124            (Request, Framed<TlsStream<TcpStream>, Codec>),
125            Config = (),
126            Response = (),
127        >,
128        U::Future: 'static,
129        U::Error: fmt::Display + Into<Response<BoxBody>>,
130        U::InitError: fmt::Debug,
131    {
132        /// Create OpenSSL based service.
133        pub fn openssl(
134            self,
135            acceptor: SslAcceptor,
136        ) -> impl ServiceFactory<
137            TcpStream,
138            Config = (),
139            Response = (),
140            Error = TlsError<SslError, DispatchError>,
141            InitError = (),
142        > {
143            Acceptor::new(acceptor)
144                .map_init_err(|_| {
145                    unreachable!("TLS acceptor service factory does not error on init")
146                })
147                .map_err(TlsError::into_service_error)
148                .map(|io: TlsStream<TcpStream>| {
149                    let peer_addr = io.get_ref().peer_addr().ok();
150                    (io, peer_addr)
151                })
152                .and_then(self.map_err(TlsError::Service))
153        }
154    }
155}
156
157#[cfg(feature = "rustls-0_20")]
158mod rustls_0_20 {
159    use std::io;
160
161    use actix_tls::accept::{
162        rustls_0_20::{reexports::ServerConfig, Acceptor, TlsStream},
163        TlsError,
164    };
165
166    use super::*;
167
168    impl<S, B, X, U> H1Service<TlsStream<TcpStream>, S, B, X, U>
169    where
170        S: ServiceFactory<Request, Config = ()>,
171        S::Future: 'static,
172        S::Error: Into<Response<BoxBody>>,
173        S::InitError: fmt::Debug,
174        S::Response: Into<Response<B>>,
175
176        B: MessageBody,
177
178        X: ServiceFactory<Request, Config = (), Response = Request>,
179        X::Future: 'static,
180        X::Error: Into<Response<BoxBody>>,
181        X::InitError: fmt::Debug,
182
183        U: ServiceFactory<
184            (Request, Framed<TlsStream<TcpStream>, Codec>),
185            Config = (),
186            Response = (),
187        >,
188        U::Future: 'static,
189        U::Error: fmt::Display + Into<Response<BoxBody>>,
190        U::InitError: fmt::Debug,
191    {
192        /// Create Rustls v0.20 based service.
193        pub fn rustls(
194            self,
195            config: ServerConfig,
196        ) -> impl ServiceFactory<
197            TcpStream,
198            Config = (),
199            Response = (),
200            Error = TlsError<io::Error, DispatchError>,
201            InitError = (),
202        > {
203            Acceptor::new(config)
204                .map_init_err(|_| {
205                    unreachable!("TLS acceptor service factory does not error on init")
206                })
207                .map_err(TlsError::into_service_error)
208                .map(|io: TlsStream<TcpStream>| {
209                    let peer_addr = io.get_ref().0.peer_addr().ok();
210                    (io, peer_addr)
211                })
212                .and_then(self.map_err(TlsError::Service))
213        }
214    }
215}
216
217#[cfg(feature = "rustls-0_21")]
218mod rustls_0_21 {
219    use std::io;
220
221    use actix_tls::accept::{
222        rustls_0_21::{reexports::ServerConfig, Acceptor, TlsStream},
223        TlsError,
224    };
225
226    use super::*;
227
228    impl<S, B, X, U> H1Service<TlsStream<TcpStream>, S, B, X, U>
229    where
230        S: ServiceFactory<Request, Config = ()>,
231        S::Future: 'static,
232        S::Error: Into<Response<BoxBody>>,
233        S::InitError: fmt::Debug,
234        S::Response: Into<Response<B>>,
235
236        B: MessageBody,
237
238        X: ServiceFactory<Request, Config = (), Response = Request>,
239        X::Future: 'static,
240        X::Error: Into<Response<BoxBody>>,
241        X::InitError: fmt::Debug,
242
243        U: ServiceFactory<
244            (Request, Framed<TlsStream<TcpStream>, Codec>),
245            Config = (),
246            Response = (),
247        >,
248        U::Future: 'static,
249        U::Error: fmt::Display + Into<Response<BoxBody>>,
250        U::InitError: fmt::Debug,
251    {
252        /// Create Rustls v0.21 based service.
253        pub fn rustls_021(
254            self,
255            config: ServerConfig,
256        ) -> impl ServiceFactory<
257            TcpStream,
258            Config = (),
259            Response = (),
260            Error = TlsError<io::Error, DispatchError>,
261            InitError = (),
262        > {
263            Acceptor::new(config)
264                .map_init_err(|_| {
265                    unreachable!("TLS acceptor service factory does not error on init")
266                })
267                .map_err(TlsError::into_service_error)
268                .map(|io: TlsStream<TcpStream>| {
269                    let peer_addr = io.get_ref().0.peer_addr().ok();
270                    (io, peer_addr)
271                })
272                .and_then(self.map_err(TlsError::Service))
273        }
274    }
275}
276
277#[cfg(feature = "rustls-0_22")]
278mod rustls_0_22 {
279    use std::io;
280
281    use actix_tls::accept::{
282        rustls_0_22::{reexports::ServerConfig, Acceptor, TlsStream},
283        TlsError,
284    };
285
286    use super::*;
287
288    impl<S, B, X, U> H1Service<TlsStream<TcpStream>, S, B, X, U>
289    where
290        S: ServiceFactory<Request, Config = ()>,
291        S::Future: 'static,
292        S::Error: Into<Response<BoxBody>>,
293        S::InitError: fmt::Debug,
294        S::Response: Into<Response<B>>,
295
296        B: MessageBody,
297
298        X: ServiceFactory<Request, Config = (), Response = Request>,
299        X::Future: 'static,
300        X::Error: Into<Response<BoxBody>>,
301        X::InitError: fmt::Debug,
302
303        U: ServiceFactory<
304            (Request, Framed<TlsStream<TcpStream>, Codec>),
305            Config = (),
306            Response = (),
307        >,
308        U::Future: 'static,
309        U::Error: fmt::Display + Into<Response<BoxBody>>,
310        U::InitError: fmt::Debug,
311    {
312        /// Create Rustls v0.22 based service.
313        pub fn rustls_0_22(
314            self,
315            config: ServerConfig,
316        ) -> impl ServiceFactory<
317            TcpStream,
318            Config = (),
319            Response = (),
320            Error = TlsError<io::Error, DispatchError>,
321            InitError = (),
322        > {
323            Acceptor::new(config)
324                .map_init_err(|_| {
325                    unreachable!("TLS acceptor service factory does not error on init")
326                })
327                .map_err(TlsError::into_service_error)
328                .map(|io: TlsStream<TcpStream>| {
329                    let peer_addr = io.get_ref().0.peer_addr().ok();
330                    (io, peer_addr)
331                })
332                .and_then(self.map_err(TlsError::Service))
333        }
334    }
335}
336
337#[cfg(feature = "rustls-0_23")]
338mod rustls_0_23 {
339    use std::io;
340
341    use actix_tls::accept::{
342        rustls_0_23::{reexports::ServerConfig, Acceptor, TlsStream},
343        TlsError,
344    };
345
346    use super::*;
347
348    impl<S, B, X, U> H1Service<TlsStream<TcpStream>, S, B, X, U>
349    where
350        S: ServiceFactory<Request, Config = ()>,
351        S::Future: 'static,
352        S::Error: Into<Response<BoxBody>>,
353        S::InitError: fmt::Debug,
354        S::Response: Into<Response<B>>,
355
356        B: MessageBody,
357
358        X: ServiceFactory<Request, Config = (), Response = Request>,
359        X::Future: 'static,
360        X::Error: Into<Response<BoxBody>>,
361        X::InitError: fmt::Debug,
362
363        U: ServiceFactory<
364            (Request, Framed<TlsStream<TcpStream>, Codec>),
365            Config = (),
366            Response = (),
367        >,
368        U::Future: 'static,
369        U::Error: fmt::Display + Into<Response<BoxBody>>,
370        U::InitError: fmt::Debug,
371    {
372        /// Create Rustls v0.23 based service.
373        pub fn rustls_0_23(
374            self,
375            config: ServerConfig,
376        ) -> impl ServiceFactory<
377            TcpStream,
378            Config = (),
379            Response = (),
380            Error = TlsError<io::Error, DispatchError>,
381            InitError = (),
382        > {
383            Acceptor::new(config)
384                .map_init_err(|_| {
385                    unreachable!("TLS acceptor service factory does not error on init")
386                })
387                .map_err(TlsError::into_service_error)
388                .map(|io: TlsStream<TcpStream>| {
389                    let peer_addr = io.get_ref().0.peer_addr().ok();
390                    (io, peer_addr)
391                })
392                .and_then(self.map_err(TlsError::Service))
393        }
394    }
395}
396
397impl<T, S, B, X, U> H1Service<T, S, B, X, U>
398where
399    S: ServiceFactory<Request, Config = ()>,
400    S::Error: Into<Response<BoxBody>>,
401    S::Response: Into<Response<B>>,
402    S::InitError: fmt::Debug,
403    B: MessageBody,
404{
405    pub fn expect<X1>(self, expect: X1) -> H1Service<T, S, B, X1, U>
406    where
407        X1: ServiceFactory<Request, Response = Request>,
408        X1::Error: Into<Response<BoxBody>>,
409        X1::InitError: fmt::Debug,
410    {
411        H1Service {
412            expect,
413            cfg: self.cfg,
414            srv: self.srv,
415            upgrade: self.upgrade,
416            on_connect_ext: self.on_connect_ext,
417            _phantom: PhantomData,
418        }
419    }
420
421    pub fn upgrade<U1>(self, upgrade: Option<U1>) -> H1Service<T, S, B, X, U1>
422    where
423        U1: ServiceFactory<(Request, Framed<T, Codec>), Response = ()>,
424        U1::Error: fmt::Display,
425        U1::InitError: fmt::Debug,
426    {
427        H1Service {
428            upgrade,
429            cfg: self.cfg,
430            srv: self.srv,
431            expect: self.expect,
432            on_connect_ext: self.on_connect_ext,
433            _phantom: PhantomData,
434        }
435    }
436
437    /// Set on connect callback.
438    pub(crate) fn on_connect_ext(mut self, f: Option<Rc<ConnectCallback<T>>>) -> Self {
439        self.on_connect_ext = f;
440        self
441    }
442}
443
444impl<T, S, B, X, U> ServiceFactory<(T, Option<net::SocketAddr>)> for H1Service<T, S, B, X, U>
445where
446    T: AsyncRead + AsyncWrite + Unpin + 'static,
447
448    S: ServiceFactory<Request, Config = ()>,
449    S::Future: 'static,
450    S::Error: Into<Response<BoxBody>>,
451    S::Response: Into<Response<B>>,
452    S::InitError: fmt::Debug,
453
454    B: MessageBody,
455
456    X: ServiceFactory<Request, Config = (), Response = Request>,
457    X::Future: 'static,
458    X::Error: Into<Response<BoxBody>>,
459    X::InitError: fmt::Debug,
460
461    U: ServiceFactory<(Request, Framed<T, Codec>), Config = (), Response = ()>,
462    U::Future: 'static,
463    U::Error: fmt::Display + Into<Response<BoxBody>>,
464    U::InitError: fmt::Debug,
465{
466    type Response = ();
467    type Error = DispatchError;
468    type Config = ();
469    type Service = H1ServiceHandler<T, S::Service, B, X::Service, U::Service>;
470    type InitError = ();
471    type Future = LocalBoxFuture<'static, Result<Self::Service, Self::InitError>>;
472
473    fn new_service(&self, _: ()) -> Self::Future {
474        let service = self.srv.new_service(());
475        let expect = self.expect.new_service(());
476        let upgrade = self.upgrade.as_ref().map(|s| s.new_service(()));
477        let on_connect_ext = self.on_connect_ext.clone();
478        let cfg = self.cfg.clone();
479
480        Box::pin(async move {
481            let expect = expect.await.map_err(|err| {
482                tracing::error!("Initialization of HTTP expect service error: {err:?}");
483            })?;
484
485            let upgrade = match upgrade {
486                Some(upgrade) => {
487                    let upgrade = upgrade.await.map_err(|err| {
488                        tracing::error!("Initialization of HTTP upgrade service error: {err:?}");
489                    })?;
490                    Some(upgrade)
491                }
492                None => None,
493            };
494
495            let service = service
496                .await
497                .map_err(|err| error!("Initialization of HTTP service error: {err:?}"))?;
498
499            Ok(H1ServiceHandler::new(
500                cfg,
501                service,
502                expect,
503                upgrade,
504                on_connect_ext,
505            ))
506        })
507    }
508}
509
510/// `Service` implementation for HTTP/1 transport
511pub type H1ServiceHandler<T, S, B, X, U> = HttpServiceHandler<T, S, B, X, U>;
512
513impl<T, S, B, X, U> Service<(T, Option<net::SocketAddr>)> for HttpServiceHandler<T, S, B, X, U>
514where
515    T: AsyncRead + AsyncWrite + Unpin,
516
517    S: Service<Request>,
518    S::Error: Into<Response<BoxBody>>,
519    S::Response: Into<Response<B>>,
520
521    B: MessageBody,
522
523    X: Service<Request, Response = Request>,
524    X::Error: Into<Response<BoxBody>>,
525
526    U: Service<(Request, Framed<T, Codec>), Response = ()>,
527    U::Error: fmt::Display + Into<Response<BoxBody>>,
528{
529    type Response = ();
530    type Error = DispatchError;
531    type Future = Dispatcher<T, S, B, X, U>;
532
533    fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
534        self._poll_ready(cx).map_err(|err| {
535            error!("HTTP/1 service readiness error: {:?}", err);
536            DispatchError::Service(err)
537        })
538    }
539
540    fn call(&self, (io, addr): (T, Option<net::SocketAddr>)) -> Self::Future {
541        let conn_data = OnConnectData::from_io(&io, self.on_connect_ext.as_deref());
542        Dispatcher::new(io, Rc::clone(&self.flow), self.cfg.clone(), addr, conn_data)
543    }
544}
actix_http/h1/service.rs

actix_http/h1/
service.rs
