berkus
/
actix-web
mirror of https://github.com/fafhrd91/actix-web
1
0
Fork 0
Code Issues Releases Wiki Activity

Merge branch 'master' into default_error_handler

This commit is contained in:
e-rhodes 2022-06-27 08:46:18 -06:00 committed by GitHub
parent 0c1778da33 0dba6310c6
commit 2781d70849
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
20 changed files with 353 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
actix-http/src/body/message_body.rs
View File

@ -481,6 +481,7 @@ mod tests {
assert_poll_next_none!(pl);
}
#[allow(clippy::let_unit_value)]
#[actix_rt::test]
async fn test_unit() {
let pl = ();

3
actix-http/src/lib.rs
View File

@ -25,6 +25,7 @@
)]
#![doc(html_logo_url = "https://actix.rs/img/logo.png")]
#![doc(html_favicon_url = "https://actix.rs/favicon.ico")]
#![cfg_attr(docsrs, feature(doc_cfg))]
pub use ::http::{uri, uri::Uri};
pub use ::http::{Method, StatusCode, Version};
@ -69,6 +70,8 @@ pub use self::payload::{BoxedPayloadStream, Payload, PayloadStream};
pub use self::requests::{Request, RequestHead, RequestHeadType};
pub use self::responses::{Response, ResponseBuilder, ResponseHead};
pub use self::service::HttpService;
#[cfg(any(feature = "openssl", feature = "rustls"))]
pub use self::service::TlsAcceptorConfig;
/// A major HTTP protocol version.
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]

4
actix-http/src/responses/head.rs
View File

@ -237,7 +237,7 @@ mod tests {
.await;
let mut stream = net::TcpStream::connect(srv.addr()).unwrap();
let _ = stream
stream
.write_all(b"GET /camel HTTP/1.1\r\nConnection: Close\r\n\r\n")
.unwrap();
let mut data = vec![];
@ -251,7 +251,7 @@ mod tests {
assert!(memmem::find(&data, b"content-length").is_none());
let mut stream = net::TcpStream::connect(srv.addr()).unwrap();
let _ = stream
stream
.write_all(b"GET /lower HTTP/1.1\r\nConnection: Close\r\n\r\n")
.unwrap();
let mut data = vec![];

65
actix-http/src/service.rs
View File

@ -181,6 +181,25 @@ where
}
}
/// Configuration options used when accepting TLS connection.
#[cfg(any(feature = "openssl", feature = "rustls"))]
#[cfg_attr(docsrs, doc(cfg(any(feature = "openssl", feature = "rustls"))))]
#[derive(Debug, Default)]
pub struct TlsAcceptorConfig {
pub(crate) handshake_timeout: Option<std::time::Duration>,
}
#[cfg(any(feature = "openssl", feature = "rustls"))]
impl TlsAcceptorConfig {
/// Set TLS handshake timeout duration.
pub fn handshake_timeout(self, dur: std::time::Duration) -> Self {
Self {
handshake_timeout: Some(dur),
// ..self
}
}
}
#[cfg(feature = "openssl")]
mod openssl {
use actix_service::ServiceFactoryExt as _;
@ -230,7 +249,28 @@ mod openssl {
Error = TlsError<SslError, DispatchError>,
InitError = (),
> {
Acceptor::new(acceptor)
self.openssl_with_config(acceptor, TlsAcceptorConfig::default())
}
/// Create OpenSSL based service with custom TLS acceptor configuration.
pub fn openssl_with_config(
self,
acceptor: SslAcceptor,
tls_acceptor_config: TlsAcceptorConfig,
) -> impl ServiceFactory<
TcpStream,
Config = (),
Response = (),
Error = TlsError<SslError, DispatchError>,
InitError = (),
> {
let mut acceptor = Acceptor::new(acceptor);
if let Some(handshake_timeout) = tls_acceptor_config.handshake_timeout {
acceptor.set_handshake_timeout(handshake_timeout);
}
acceptor
.map_init_err(|_| {
unreachable!("TLS acceptor service factory does not error on init")
})
@ -293,8 +333,23 @@ mod rustls {
{
/// Create Rustls based service.
pub fn rustls(
self,
config: ServerConfig,
) -> impl ServiceFactory<
TcpStream,
Config = (),
Response = (),
Error = TlsError<io::Error, DispatchError>,
InitError = (),
> {
self.rustls_with_config(config, TlsAcceptorConfig::default())
}
/// Create Rustls based service with custom TLS acceptor configuration.
pub fn rustls_with_config(
self,
mut config: ServerConfig,
tls_acceptor_config: TlsAcceptorConfig,
) -> impl ServiceFactory<
TcpStream,
Config = (),
@ -306,7 +361,13 @@ mod rustls {
protos.extend_from_slice(&config.alpn_protocols);
config.alpn_protocols = protos;
Acceptor::new(config)
let mut acceptor = Acceptor::new(config);
if let Some(handshake_timeout) = tls_acceptor_config.handshake_timeout {
acceptor.set_handshake_timeout(handshake_timeout);
}
acceptor
.map_init_err(|_| {
unreachable!("TLS acceptor service factory does not error on init")
})

9
actix-http/tests/test_openssl.rs
View File

@ -2,13 +2,13 @@
extern crate tls_openssl as openssl;
use std::{convert::Infallible, io};
use std::{convert::Infallible, io, time::Duration};
use actix_http::{
body::{BodyStream, BoxBody, SizedStream},
error::PayloadError,
header::{self, HeaderValue},
Error, HttpService, Method, Request, Response, StatusCode, Version,
Error, HttpService, Method, Request, Response, StatusCode, TlsAcceptorConfig, Version,
};
use actix_http_test::test_server;
use actix_service::{fn_service, ServiceFactoryExt};
@ -89,7 +89,10 @@ async fn h2_1() -> io::Result<()> {
assert_eq!(req.version(), Version::HTTP_2);
ok::<_, Error>(Response::ok())
})
.openssl(tls_config())
.openssl_with_config(
tls_config(),
TlsAcceptorConfig::default().handshake_timeout(Duration::from_secs(5)),
)
.map_err(|_| ())
})
.await;

8
actix-http/tests/test_rustls.rs
View File

@ -8,13 +8,14 @@ use std::{
net::{SocketAddr, TcpStream as StdTcpStream},
sync::Arc,
task::Poll,
time::Duration,
};
use actix_http::{
body::{BodyStream, BoxBody, SizedStream},
error::PayloadError,
header::{self, HeaderName, HeaderValue},
Error, HttpService, Method, Request, Response, StatusCode, Version,
Error, HttpService, Method, Request, Response, StatusCode, TlsAcceptorConfig, Version,
};
use actix_http_test::test_server;
use actix_rt::pin;
@ -160,7 +161,10 @@ async fn h2_1() -> io::Result<()> {
assert_eq!(req.version(), Version::HTTP_2);
ok::<_, Error>(Response::ok())
})
.rustls(tls_config())
.rustls_with_config(
tls_config(),
TlsAcceptorConfig::default().handshake_timeout(Duration::from_secs(5)),
)
})
.await;

3
actix-web-actors/Cargo.toml
View File

@ -29,6 +29,9 @@ tokio = { version = "1.13.1", features = ["sync"] }
actix-rt = "2.2"
actix-test = "0.1.0-beta.13"
awc = { version = "3", default-features = false }
actix-web = { version = "4", features = ["macros"] }
mime = "0.3"
env_logger = "0.9"
futures-util = { version = "0.3.7", default-features = false }

56
actix-web-actors/src/context.rs
View File

@ -14,6 +14,58 @@ use futures_core::Stream;
use tokio::sync::oneshot::Sender;
/// Execution context for HTTP actors
///
/// # Example
///
/// A demonstration of [server-sent events](https://developer.mozilla.org/docs/Web/API/Server-sent_events) using actors:
///
/// ```no_run
/// use std::time::Duration;
///
/// use actix::{Actor, AsyncContext};
/// use actix_web::{get, http::header, App, HttpResponse, HttpServer};
/// use actix_web_actors::HttpContext;
/// use bytes::Bytes;
///
/// struct MyActor {
/// count: usize,
/// }
///
/// impl Actor for MyActor {
/// type Context = HttpContext<Self>;
///
/// fn started(&mut self, ctx: &mut Self::Context) {
/// ctx.run_later(Duration::from_millis(100), Self::write);
/// }
/// }
///
/// impl MyActor {
/// fn write(&mut self, ctx: &mut HttpContext<Self>) {
/// self.count += 1;
/// if self.count > 3 {
/// ctx.write_eof()
/// } else {
/// ctx.write(Bytes::from(format!("event: count\ndata: {}\n\n", self.count)));
/// ctx.run_later(Duration::from_millis(100), Self::write);
/// }
/// }
/// }
///
/// #[get("/")]
/// async fn index() -> HttpResponse {
/// HttpResponse::Ok()
/// .insert_header(header::ContentType(mime::TEXT_EVENT_STREAM))
/// .streaming(HttpContext::create(MyActor { count: 0 }))
/// }
///
/// #[actix_web::main]
/// async fn main() -> std::io::Result<()> {
/// HttpServer::new(|| App::new().service(index))
/// .bind(("127.0.0.1", 8080))?
/// .run()
/// .await
/// }
/// ```
pub struct HttpContext<A>
where
A: Actor<Context = HttpContext<A>>,
@ -210,7 +262,7 @@ mod tests {
type Context = HttpContext<Self>;
fn started(&mut self, ctx: &mut Self::Context) {
ctx.run_later(Duration::from_millis(100), |slf, ctx| slf.write(ctx));
ctx.run_later(Duration::from_millis(100), Self::write);
}
}
@ -221,7 +273,7 @@ mod tests {
ctx.write_eof()
} else {
ctx.write(Bytes::from(format!("LINE-{}", self.count)));
ctx.run_later(Duration::from_millis(100), |slf, ctx| slf.write(ctx));
ctx.run_later(Duration::from_millis(100), Self::write);
}
}
}

55
actix-web-actors/src/lib.rs
View File

@ -1,4 +1,59 @@
//! Actix actors support for Actix Web.
//!
//! # Examples
//!
//! ```no_run
//! use actix::{Actor, StreamHandler};
//! use actix_web::{get, web, App, Error, HttpRequest, HttpResponse, HttpServer};
//! use actix_web_actors::ws;
//!
//! /// Define Websocket actor
//! struct MyWs;
//!
//! impl Actor for MyWs {
//! type Context = ws::WebsocketContext<Self>;
//! }
//!
//! /// Handler for ws::Message message
//! impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for MyWs {
//! fn handle(&mut self, msg: Result<ws::Message, ws::ProtocolError>, ctx: &mut Self::Context) {
//! match msg {
//! Ok(ws::Message::Ping(msg)) => ctx.pong(&msg),
//! Ok(ws::Message::Text(text)) => ctx.text(text),
//! Ok(ws::Message::Binary(bin)) => ctx.binary(bin),
//! _ => (),
//! }
//! }
//! }
//!
//! #[get("/ws")]
//! async fn index(req: HttpRequest, stream: web::Payload) -> Result<HttpResponse, Error> {
//! ws::start(MyWs, &req, stream)
//! }
//!
//! #[actix_web::main]
//! async fn main() -> std::io::Result<()> {
//! HttpServer::new(|| App::new().service(index))
//! .bind(("127.0.0.1", 8080))?
//! .run()
//! .await
//! }
//! ```
//!
//! # Documentation & Community Resources
//! In addition to this API documentation, several other resources are available:
//!
//! * [Website & User Guide](https://actix.rs/)
//! * [Documentation for `actix_web`](actix_web)
//! * [Examples Repository](https://github.com/actix/examples)
//! * [Community Chat on Discord](https://discord.gg/NWpN5mmg3x)
//!
//! To get started navigating the API docs, you may consider looking at the following pages first:
//!
//! * [`ws`]: This module provides actor support for WebSockets.
//!
//! * [`HttpContext`]: This struct provides actor support for streaming HTTP responses.
//!
#![deny(rust_2018_idioms, nonstandard_style)]
#![warn(future_incompatible)]

105
actix-web-actors/src/ws.rs
View File

@ -1,4 +1,60 @@
//! Websocket integration.
//!
//! # Examples
//!
//! ```no_run
//! use actix::{Actor, StreamHandler};
//! use actix_web::{get, web, App, Error, HttpRequest, HttpResponse, HttpServer};
//! use actix_web_actors::ws;
//!
//! /// Define Websocket actor
//! struct MyWs;
//!
//! impl Actor for MyWs {
//! type Context = ws::WebsocketContext<Self>;
//! }
//!
//! /// Handler for ws::Message message
//! impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for MyWs {
//! fn handle(&mut self, msg: Result<ws::Message, ws::ProtocolError>, ctx: &mut Self::Context) {
//! match msg {
//! Ok(ws::Message::Ping(msg)) => ctx.pong(&msg),
//! Ok(ws::Message::Text(text)) => ctx.text(text),
//! Ok(ws::Message::Binary(bin)) => ctx.binary(bin),
//! _ => (),
//! }
//! }
//! }
//!
//! #[get("/ws")]
//! async fn websocket(req: HttpRequest, stream: web::Payload) -> Result<HttpResponse, Error> {
//! ws::start(MyWs, &req, stream)
//! }
//!
//! const MAX_FRAME_SIZE: usize = 16_384; // 16KiB
//!
//! #[get("/custom-ws")]
//! async fn custom_websocket(req: HttpRequest, stream: web::Payload) -> Result<HttpResponse, Error> {
//! // Create a Websocket session with a specific max frame size, and protocols.
//! ws::WsResponseBuilder::new(MyWs, &req, stream)
//! .frame_size(MAX_FRAME_SIZE)
//! .protocols(&["A", "B"])
//! .start()
//! }
//!
//! #[actix_web::main]
//! async fn main() -> std::io::Result<()> {
//! HttpServer::new(|| {
//! App::new()
//! .service(websocket)
//! .service(custom_websocket)
//! })
//! .bind(("127.0.0.1", 8080))?
//! .run()
//! .await
//! }
//! ```
//!
use std::{
collections::VecDeque,
@ -41,20 +97,51 @@ use tokio::sync::oneshot;
///
/// # Examples
///
/// Create a Websocket session response with default configuration.
/// ```ignore
/// WsResponseBuilder::new(WsActor, &req, stream).start()
/// ```
/// ```no_run
/// # use actix::{Actor, StreamHandler};
/// # use actix_web::{get, web, App, Error, HttpRequest, HttpResponse, HttpServer};
/// # use actix_web_actors::ws;
/// #
/// # struct MyWs;
/// #
/// # impl Actor for MyWs {
/// # type Context = ws::WebsocketContext<Self>;
/// # }
/// #
/// # /// Handler for ws::Message message
/// # impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for MyWs {
/// # fn handle(&mut self, msg: Result<ws::Message, ws::ProtocolError>, ctx: &mut Self::Context) {}
/// # }
/// #
/// #[get("/ws")]
/// async fn websocket(req: HttpRequest, stream: web::Payload) -> Result<HttpResponse, Error> {
/// ws::WsResponseBuilder::new(MyWs, &req, stream).start()
/// }
///
/// Create a Websocket session with a specific max frame size, [`Codec`], and protocols.
/// ```ignore
/// const MAX_FRAME_SIZE: usize = 16_384; // 16KiB
///
/// ws::WsResponseBuilder::new(WsActor, &req, stream)
/// .codec(Codec::new())
/// .protocols(&["A", "B"])
/// #[get("/custom-ws")]
/// async fn custom_websocket(req: HttpRequest, stream: web::Payload) -> Result<HttpResponse, Error> {
/// // Create a Websocket session with a specific max frame size, codec, and protocols.
/// ws::WsResponseBuilder::new(MyWs, &req, stream)
/// .codec(actix_http::ws::Codec::new())
/// // This will overwrite the codec's max frame-size
/// .frame_size(MAX_FRAME_SIZE)
/// .protocols(&["A", "B"])
/// .start()
/// }
/// #
/// # #[actix_web::main]
/// # async fn main() -> std::io::Result<()> {
/// # HttpServer::new(|| {
/// # App::new()
/// # .service(websocket)
/// # .service(custom_websocket)
/// # })
/// # .bind(("127.0.0.1", 8080))?
/// # .run()
/// # .await
/// # }
/// ```
pub struct WsResponseBuilder<'a, A, T>
where

9
actix-web/CHANGES.md
View File

@ -1,12 +1,15 @@
# Changelog
## Unreleased - 2022-xx-xx
### Added
- Add configuration options for TLS handshake timeout via `HttpServer::{rustls, openssl}_with_config` methods. [#2752]
- Add `ErrorHandlers::default_handler()` (as well as `default_handler_{server, client}()`) to make registering handlers with the `ErrorHandlers` middleware easier. [#2784]
- Add `ServiceRequest::{parts, request}()` getter methods. [#2786]
### Changed
- Minimum supported Rust version (MSRV) is now 1.57 due to transitive `time` dependency.
### Added
- Add `ServiceRequest::{parts, request}()` getter methods. [#2786]
- Add `ErrorHandlers::default_handler()` (as well as `default_handler_{server, client}()`) to make registering handlers with the `ErrorHandlers` middleware easier. [#2784]
[#2752]: https://github.com/actix/actix-web/pull/2752
[#2784]: https://github.com/actix/actix-web/pull/2784
[#2786]: https://github.com/actix/actix-web/pull/2786

2
actix-web/src/app.rs
View File

@ -60,7 +60,7 @@ where
/// [`HttpRequest::app_data`](crate::HttpRequest::app_data) method at runtime.
///
/// # [`Data<T>`]
/// Any [`Data<T>`] type added here can utilize it's extractor implementation in handlers.
/// Any [`Data<T>`] type added here can utilize its extractor implementation in handlers.
/// Types not wrapped in `Data<T>` cannot use this extractor. See [its docs](Data<T>) for more
/// about its usage and patterns.
///

2
actix-web/src/app_service.rs
View File

@ -257,7 +257,7 @@ impl ServiceFactory<ServiceRequest> for AppRoutingFactory {
type Future = LocalBoxFuture<'static, Result<Self::Service, Self::InitError>>;
fn new_service(&self, _: ()) -> Self::Future {
// construct all services factory future with it's resource def and guards.
// construct all services factory future with its resource def and guards.
let factory_fut = join_all(self.services.iter().map(|(path, factory, guards)| {
let path = path.clone();
let guards = guards.borrow_mut().take().unwrap_or_default();

10
actix-web/src/config.rs
View File

@ -153,6 +153,16 @@ impl AppConfig {
}
impl Default for AppConfig {
/// Returns the default AppConfig.
/// Note: The included socket address is "127.0.0.1".
///
/// 127.0.0.1: non-routable meta address that denotes an unknown, invalid or non-applicable target.
/// If you need a service only accessed by itself, use a loopback address.
/// A loopback address for IPv4 is any loopback address that begins with "127".
/// Loopback addresses should be only used to test your application locally.
/// The default configuration provides a loopback address.
///
/// 0.0.0.0: if configured to use this special address, the application will listen to any IP address configured on the machine.
fn default() -> Self {
AppConfig::new(
false,

4
actix-web/src/guard.rs
View File

@ -254,7 +254,7 @@ impl Guard for AllGuard {
}
}
/// Wraps a guard and inverts the outcome of it's `Guard` implementation.
/// Wraps a guard and inverts the outcome of its `Guard` implementation.
///
/// # Examples
/// The handler below will be called for any request method apart from `GET`.
@ -459,7 +459,7 @@ impl Guard for HostGuard {
return scheme == req_host_uri_scheme;
}
// TODO: is the the correct behavior?
// TODO: is this the correct behavior?
// falls through if scheme cannot be determined
}

2
actix-web/src/handler.rs
View File

@ -37,7 +37,7 @@ use crate::{
/// Thanks to Rust's type system, Actix Web can infer the function parameter types. During the
/// extraction step, the parameter types are described as a tuple type, [`from_request`] is run on
/// that tuple, and the `Handler::call` implementation for that particular function arity
/// destructures the tuple into it's component types and calls your handler function with them.
/// destructures the tuple into its component types and calls your handler function with them.
///
/// In pseudo-code the process looks something like this:
/// ```ignore

2
actix-web/src/response/response.rs
View File

@ -343,7 +343,7 @@ mod response_fut_impl {
// Future is only implemented for BoxBody payload type because it's the most useful for making
// simple handlers without async blocks. Making it generic over all MessageBody types requires a
// future impl on Response which would cause it's body field to be, undesirably, Option<B>.
// future impl on Response which would cause its body field to be, undesirably, Option<B>.
//
// This impl is not particularly efficient due to the Response construction and should probably
// not be invoked if performance is important. Prefer an async fn/block in such cases.

39
actix-web/src/server.rs
View File

@ -18,6 +18,9 @@ use actix_tls::accept::openssl::reexports::{AlpnError, SslAcceptor, SslAcceptorB
#[cfg(feature = "rustls")]
use actix_tls::accept::rustls::reexports::ServerConfig as RustlsServerConfig;
#[cfg(any(feature = "openssl", feature = "rustls"))]
use actix_http::TlsAcceptorConfig;
use crate::{config::AppConfig, Error};
struct Socket {
@ -30,6 +33,8 @@ struct Config {
keep_alive: KeepAlive,
client_request_timeout: Duration,
client_disconnect_timeout: Duration,
#[cfg(any(feature = "openssl", feature = "rustls"))]
tls_handshake_timeout: Option<Duration>,
}
/// An HTTP Server.
@ -92,6 +97,8 @@ where
keep_alive: KeepAlive::default(),
client_request_timeout: Duration::from_secs(5),
client_disconnect_timeout: Duration::from_secs(1),
#[cfg(any(feature = "rustls", feature = "openssl"))]
tls_handshake_timeout: None,
})),
backlog: 1024,
sockets: Vec::new(),
@ -225,6 +232,24 @@ where
self
}
/// Set TLS handshake timeout.
///
/// Defines a timeout for TLS handshake. If the TLS handshake does not complete
/// within this time, the connection is closed.
///
/// By default handshake timeout is set to 3000 milliseconds.
#[cfg(any(feature = "openssl", feature = "rustls"))]
#[cfg_attr(docsrs, doc(cfg(any(feature = "openssl", feature = "rustls"))))]
pub fn tls_handshake_timeout(self, dur: Duration) -> Self {
self.config
.lock()
.unwrap()
.tls_handshake_timeout
.replace(dur);
self
}
#[doc(hidden)]
#[deprecated(since = "4.0.0", note = "Renamed to `client_disconnect_timeout`.")]
pub fn client_shutdown(self, dur: u64) -> Self {
@ -376,10 +401,15 @@ where
.into_factory()
.map_err(|err| err.into().error_response());
let acceptor_config = match c.tls_handshake_timeout {
Some(dur) => TlsAcceptorConfig::default().handshake_timeout(dur),
None => TlsAcceptorConfig::default(),
};
svc.finish(map_config(fac, move |_| {
AppConfig::new(true, host.clone(), addr)
}))
.openssl(acceptor.clone())
.openssl_with_config(acceptor.clone(), acceptor_config)
})?;
Ok(self)
@ -434,10 +464,15 @@ where
.into_factory()
.map_err(|err| err.into().error_response());
let acceptor_config = match c.tls_handshake_timeout {
Some(dur) => TlsAcceptorConfig::default().handshake_timeout(dur),
None => TlsAcceptorConfig::default(),
};
svc.finish(map_config(fac, move |_| {
AppConfig::new(true, host.clone(), addr)
}))
.rustls(config.clone())
.rustls_with_config(config.clone(), acceptor_config)
})?;
Ok(self)

1
actix-web/src/types/path.rs
View File

@ -183,6 +183,7 @@ mod tests {
assert!(Path::<MyStruct>::from_request(&req, &mut pl).await.is_err());
}
#[allow(clippy::let_unit_value)]
#[actix_rt::test]
async fn test_tuple_extract() {
let resource = ResourceDef::new("/{key}/{value}/");

4
actix-web/src/types/payload.rs
View File

@ -113,7 +113,7 @@ pub struct BytesExtractFut {
body_fut: HttpMessageBody,
}
impl<'a> Future for BytesExtractFut {
impl Future for BytesExtractFut {
type Output = Result<Bytes, Error>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
@ -167,7 +167,7 @@ pub struct StringExtractFut {
encoding: &'static Encoding,
}
impl<'a> Future for StringExtractFut {
impl Future for StringExtractFut {
type Output = Result<String, Error>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {