mirror of https://github.com/fafhrd91/actix-web
use safe solution for connection data
This commit is contained in:
parent
1aee8a1a58
commit
bcd3101a5b
|
@ -6,11 +6,10 @@ use actix_service::{IntoServiceFactory, Service, ServiceFactory};
|
|||
use crate::{
|
||||
body::{BoxBody, MessageBody},
|
||||
config::{KeepAlive, ServiceConfig},
|
||||
extensions::CloneableExtensions,
|
||||
h1::{self, ExpectHandler, H1Service, UpgradeHandler},
|
||||
h2::H2Service,
|
||||
service::HttpService,
|
||||
ConnectCallback, Request, Response,
|
||||
ConnectCallback, Extensions, Request, Response,
|
||||
};
|
||||
|
||||
/// A HTTP service builder
|
||||
|
@ -168,7 +167,7 @@ where
|
|||
/// and handlers.
|
||||
pub fn on_connect_ext<F>(mut self, f: F) -> Self
|
||||
where
|
||||
F: Fn(&T, &mut CloneableExtensions) + 'static,
|
||||
F: Fn(&T, &mut Extensions) + 'static,
|
||||
{
|
||||
self.on_connect_ext = Some(Rc::new(f));
|
||||
self
|
||||
|
|
|
@ -122,13 +122,6 @@ impl Extensions {
|
|||
pub fn extend(&mut self, other: Extensions) {
|
||||
self.map.extend(other.map);
|
||||
}
|
||||
|
||||
/// Sets (or overrides) items from cloneable extensions map into this map.
|
||||
pub(crate) fn clone_from(&mut self, other: &CloneableExtensions) {
|
||||
for (k, val) in &other.map {
|
||||
self.map.insert(*k, (**val).clone_to_any());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl fmt::Debug for Extensions {
|
||||
|
@ -141,104 +134,6 @@ fn downcast_owned<T: 'static>(boxed: Box<dyn Any>) -> Option<T> {
|
|||
boxed.downcast().ok().map(|boxed| *boxed)
|
||||
}
|
||||
|
||||
#[doc(hidden)]
|
||||
pub trait CloneToAny {
|
||||
/// Cast `self` into an `Any` reference.
|
||||
#[cfg(test)]
|
||||
fn any_ref(&self) -> &dyn Any;
|
||||
|
||||
/// Clone `self` to a new `Box<Any>` object.
|
||||
fn clone_to_any(&self) -> Box<dyn Any>;
|
||||
|
||||
/// Clone `self` to a new `Box<CloneAny>` object.
|
||||
fn clone_to_clone_any(&self) -> Box<dyn CloneAny>;
|
||||
}
|
||||
|
||||
impl<T: Clone + Any> CloneToAny for T {
|
||||
#[cfg(test)]
|
||||
fn any_ref(&self) -> &dyn Any {
|
||||
&*self
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn clone_to_any(&self) -> Box<dyn Any> {
|
||||
Box::new(self.clone())
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn clone_to_clone_any(&self) -> Box<dyn CloneAny> {
|
||||
Box::new(self.clone())
|
||||
}
|
||||
}
|
||||
|
||||
/// An [`Any`] trait with an additional [`Clone`] requirement.
|
||||
pub trait CloneAny: CloneToAny + Any {}
|
||||
impl<T: Any + Clone> CloneAny for T {}
|
||||
|
||||
impl Clone for Box<dyn CloneAny> {
|
||||
#[inline]
|
||||
fn clone(&self) -> Self {
|
||||
(**self).clone_to_clone_any()
|
||||
}
|
||||
}
|
||||
|
||||
trait UncheckedAnyExt {
|
||||
/// # Safety
|
||||
/// Caller must ensure type `T` is true type.
|
||||
#[inline]
|
||||
unsafe fn downcast_unchecked<T: 'static>(self: Box<Self>) -> Box<T> {
|
||||
Box::from_raw(Box::into_raw(self) as *mut T)
|
||||
}
|
||||
}
|
||||
|
||||
impl UncheckedAnyExt for dyn CloneAny {}
|
||||
|
||||
/// A type map for `on_connect` extensions.
|
||||
///
|
||||
/// All entries into this map must be owned types and implement `Clone` trait.
|
||||
///
|
||||
/// Many requests can be processed for each connection but the `on_connect` will only be run once
|
||||
/// when the connection is opened. Therefore, items added to this special map type need to be cloned
|
||||
/// into the regular extensions map for each request. Most useful connection information types are
|
||||
/// cloneable already but you can use reference counted wrappers if not.
|
||||
#[derive(Default)]
|
||||
pub struct CloneableExtensions {
|
||||
/// Use AHasher with a std HashMap with for faster lookups on the small `TypeId` keys.
|
||||
map: AHashMap<TypeId, Box<dyn CloneAny>>,
|
||||
}
|
||||
|
||||
impl CloneableExtensions {
|
||||
/// Insert an item into the map.
|
||||
///
|
||||
/// If an item of this type was already stored, it will be replaced and returned.
|
||||
///
|
||||
/// # Examples
|
||||
/// ```
|
||||
/// # use actix_http::Extensions;
|
||||
/// let mut map = Extensions::new();
|
||||
/// assert_eq!(map.insert(""), None);
|
||||
/// assert_eq!(map.insert(1u32), None);
|
||||
/// assert_eq!(map.insert(2u32), Some(1u32));
|
||||
/// assert_eq!(*map.get::<u32>().unwrap(), 2u32);
|
||||
/// ```
|
||||
pub fn insert<T: CloneAny>(&mut self, val: T) -> Option<T> {
|
||||
self.map
|
||||
.insert(TypeId::of::<T>(), Box::new(val))
|
||||
.map(|boxed| {
|
||||
// Safety:
|
||||
// Box is owned and `T` is known to be true type from map.
|
||||
*unsafe { UncheckedAnyExt::downcast_unchecked::<T>(boxed) }
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
fn get<T: CloneAny>(&self) -> Option<&T> {
|
||||
self.map
|
||||
.get(&TypeId::of::<T>())
|
||||
.and_then(|boxed| boxed.as_ref().any_ref().downcast_ref())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
@ -382,43 +277,4 @@ mod tests {
|
|||
assert_eq!(extensions.get(), Some(&20u8));
|
||||
assert_eq!(extensions.get_mut(), Some(&mut 20u8));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_clone_from() {
|
||||
#[derive(Clone)]
|
||||
struct NonCopy {
|
||||
num: u8,
|
||||
}
|
||||
|
||||
let mut ext = Extensions::new();
|
||||
ext.insert(2isize);
|
||||
|
||||
assert_eq!(ext.get::<isize>(), Some(&2isize));
|
||||
|
||||
let mut more_ext = CloneableExtensions::default();
|
||||
more_ext.insert(3isize);
|
||||
more_ext.insert(3usize);
|
||||
more_ext.insert(NonCopy { num: 8 });
|
||||
|
||||
ext.clone_from(&more_ext);
|
||||
|
||||
assert_eq!(ext.get::<isize>(), Some(&3isize));
|
||||
assert_eq!(ext.get::<usize>(), Some(&3usize));
|
||||
assert_eq!(more_ext.get::<isize>(), Some(&3isize));
|
||||
assert_eq!(more_ext.get::<usize>(), Some(&3usize));
|
||||
|
||||
assert!(ext.get::<NonCopy>().is_some());
|
||||
assert!(more_ext.get::<NonCopy>().is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn boxes_not_aliased() {
|
||||
let a: Box<dyn CloneAny> = Box::new(42);
|
||||
let b = a.clone_to_clone_any();
|
||||
assert_ne!(Box::into_raw(a) as *const (), Box::into_raw(b) as *const ());
|
||||
|
||||
let a: Box<dyn CloneAny> = Box::new(42);
|
||||
let b = a.clone_to_any();
|
||||
assert_ne!(Box::into_raw(a) as *const (), Box::into_raw(b) as *const ());
|
||||
}
|
||||
}
|
||||
|
|
|
@ -22,7 +22,7 @@ use crate::{
|
|||
config::ServiceConfig,
|
||||
error::{DispatchError, ParseError, PayloadError},
|
||||
service::HttpFlow,
|
||||
OnConnectData, Request, Response, StatusCode,
|
||||
Extensions, OnConnectData, Request, Response, StatusCode,
|
||||
};
|
||||
|
||||
use super::{
|
||||
|
@ -100,9 +100,9 @@ where
|
|||
U::Error: fmt::Display,
|
||||
{
|
||||
flow: Rc<HttpFlow<S, X, U>>,
|
||||
on_connect_data: OnConnectData,
|
||||
flags: Flags,
|
||||
peer_addr: Option<net::SocketAddr>,
|
||||
conn_data: Option<Rc<Extensions>>,
|
||||
error: Option<DispatchError>,
|
||||
|
||||
#[pin]
|
||||
|
@ -179,10 +179,10 @@ where
|
|||
/// Create HTTP/1 dispatcher.
|
||||
pub(crate) fn new(
|
||||
io: T,
|
||||
config: ServiceConfig,
|
||||
flow: Rc<HttpFlow<S, X, U>>,
|
||||
on_connect_data: OnConnectData,
|
||||
config: ServiceConfig,
|
||||
peer_addr: Option<net::SocketAddr>,
|
||||
conn_data: OnConnectData,
|
||||
) -> Self {
|
||||
let flags = if config.keep_alive_enabled() {
|
||||
Flags::KEEPALIVE
|
||||
|
@ -198,20 +198,23 @@ where
|
|||
|
||||
Dispatcher {
|
||||
inner: DispatcherState::Normal(InnerDispatcher {
|
||||
read_buf: BytesMut::with_capacity(HW_BUFFER_SIZE),
|
||||
write_buf: BytesMut::with_capacity(HW_BUFFER_SIZE),
|
||||
payload: None,
|
||||
state: State::None,
|
||||
error: None,
|
||||
messages: VecDeque::new(),
|
||||
io: Some(io),
|
||||
codec: Codec::new(config),
|
||||
flow,
|
||||
on_connect_data,
|
||||
flags,
|
||||
peer_addr,
|
||||
conn_data: conn_data.0.map(Rc::new),
|
||||
error: None,
|
||||
|
||||
state: State::None,
|
||||
payload: None,
|
||||
messages: VecDeque::new(),
|
||||
|
||||
ka_expire,
|
||||
ka_timer,
|
||||
|
||||
io: Some(io),
|
||||
read_buf: BytesMut::with_capacity(HW_BUFFER_SIZE),
|
||||
write_buf: BytesMut::with_capacity(HW_BUFFER_SIZE),
|
||||
codec: Codec::new(config),
|
||||
}),
|
||||
|
||||
#[cfg(test)]
|
||||
|
@ -593,8 +596,8 @@ where
|
|||
Message::Item(mut req) => {
|
||||
req.head_mut().peer_addr = *this.peer_addr;
|
||||
|
||||
// merge on_connect_ext data into request extensions
|
||||
this.on_connect_data.merge_into(&mut req);
|
||||
req.conn_data =
|
||||
this.conn_data.as_ref().map(|data| Rc::clone(data));
|
||||
|
||||
match this.codec.message_type() {
|
||||
// Request is upgradable. add upgrade message and break.
|
||||
|
@ -1100,10 +1103,10 @@ mod tests {
|
|||
|
||||
let h1 = Dispatcher::<_, _, _, _, UpgradeHandler>::new(
|
||||
buf,
|
||||
ServiceConfig::default(),
|
||||
services,
|
||||
OnConnectData::default(),
|
||||
ServiceConfig::default(),
|
||||
None,
|
||||
OnConnectData::default(),
|
||||
);
|
||||
|
||||
actix_rt::pin!(h1);
|
||||
|
@ -1140,10 +1143,10 @@ mod tests {
|
|||
|
||||
let h1 = Dispatcher::<_, _, _, _, UpgradeHandler>::new(
|
||||
buf,
|
||||
cfg,
|
||||
services,
|
||||
OnConnectData::default(),
|
||||
cfg,
|
||||
None,
|
||||
OnConnectData::default(),
|
||||
);
|
||||
|
||||
actix_rt::pin!(h1);
|
||||
|
@ -1194,10 +1197,10 @@ mod tests {
|
|||
|
||||
let h1 = Dispatcher::<_, _, _, _, UpgradeHandler>::new(
|
||||
buf,
|
||||
cfg,
|
||||
services,
|
||||
OnConnectData::default(),
|
||||
cfg,
|
||||
None,
|
||||
OnConnectData::default(),
|
||||
);
|
||||
|
||||
actix_rt::pin!(h1);
|
||||
|
@ -1244,10 +1247,10 @@ mod tests {
|
|||
|
||||
let h1 = Dispatcher::<_, _, _, _, UpgradeHandler>::new(
|
||||
buf.clone(),
|
||||
cfg,
|
||||
services,
|
||||
OnConnectData::default(),
|
||||
cfg,
|
||||
None,
|
||||
OnConnectData::default(),
|
||||
);
|
||||
|
||||
buf.extend_read_buf(
|
||||
|
@ -1316,10 +1319,10 @@ mod tests {
|
|||
|
||||
let h1 = Dispatcher::<_, _, _, _, UpgradeHandler>::new(
|
||||
buf.clone(),
|
||||
cfg,
|
||||
services,
|
||||
OnConnectData::default(),
|
||||
cfg,
|
||||
None,
|
||||
OnConnectData::default(),
|
||||
);
|
||||
|
||||
buf.extend_read_buf(
|
||||
|
@ -1393,10 +1396,10 @@ mod tests {
|
|||
|
||||
let h1 = Dispatcher::<_, _, _, _, TestUpgrade>::new(
|
||||
buf.clone(),
|
||||
cfg,
|
||||
services,
|
||||
OnConnectData::default(),
|
||||
cfg,
|
||||
None,
|
||||
OnConnectData::default(),
|
||||
);
|
||||
|
||||
buf.extend_read_buf(
|
||||
|
|
|
@ -365,15 +365,7 @@ where
|
|||
}
|
||||
|
||||
fn call(&self, (io, addr): (T, Option<net::SocketAddr>)) -> Self::Future {
|
||||
let on_connect_data =
|
||||
OnConnectData::from_io(&io, self.on_connect_ext.as_deref());
|
||||
|
||||
Dispatcher::new(
|
||||
io,
|
||||
self.cfg.clone(),
|
||||
self.flow.clone(),
|
||||
on_connect_data,
|
||||
addr,
|
||||
)
|
||||
let conn_data = OnConnectData::from_io(&io, self.on_connect_ext.as_deref());
|
||||
Dispatcher::new(io, self.flow.clone(), self.cfg.clone(), addr, conn_data)
|
||||
}
|
||||
}
|
||||
|
|
|
@ -27,7 +27,7 @@ use crate::{
|
|||
body::{BodySize, BoxBody, MessageBody},
|
||||
config::ServiceConfig,
|
||||
service::HttpFlow,
|
||||
OnConnectData, Payload, Request, Response, ResponseHead,
|
||||
Extensions, OnConnectData, Payload, Request, Response, ResponseHead,
|
||||
};
|
||||
|
||||
const CHUNK_SIZE: usize = 16_384;
|
||||
|
@ -37,7 +37,7 @@ pin_project! {
|
|||
pub struct Dispatcher<T, S, B, X, U> {
|
||||
flow: Rc<HttpFlow<S, X, U>>,
|
||||
connection: Connection<T, Bytes>,
|
||||
on_connect_data: OnConnectData,
|
||||
conn_data: Option<Rc<Extensions>>,
|
||||
config: ServiceConfig,
|
||||
peer_addr: Option<net::SocketAddr>,
|
||||
ping_pong: Option<H2PingPong>,
|
||||
|
@ -50,11 +50,11 @@ where
|
|||
T: AsyncRead + AsyncWrite + Unpin,
|
||||
{
|
||||
pub(crate) fn new(
|
||||
flow: Rc<HttpFlow<S, X, U>>,
|
||||
mut conn: Connection<T, Bytes>,
|
||||
on_connect_data: OnConnectData,
|
||||
flow: Rc<HttpFlow<S, X, U>>,
|
||||
config: ServiceConfig,
|
||||
peer_addr: Option<net::SocketAddr>,
|
||||
conn_data: OnConnectData,
|
||||
timer: Option<Pin<Box<Sleep>>>,
|
||||
) -> Self {
|
||||
let ping_pong = config.keep_alive().map(|dur| H2PingPong {
|
||||
|
@ -74,7 +74,7 @@ where
|
|||
config,
|
||||
peer_addr,
|
||||
connection: conn,
|
||||
on_connect_data,
|
||||
conn_data: conn_data.0.map(Rc::new),
|
||||
ping_pong,
|
||||
_phantom: PhantomData,
|
||||
}
|
||||
|
@ -119,8 +119,7 @@ where
|
|||
head.headers = parts.headers.into();
|
||||
head.peer_addr = this.peer_addr;
|
||||
|
||||
// merge on_connect_ext data into request extensions
|
||||
this.on_connect_data.merge_into(&mut req);
|
||||
req.conn_data = this.conn_data.as_ref().map(|data| Rc::clone(data));
|
||||
|
||||
let fut = this.flow.service.call(req);
|
||||
let config = this.config.clone();
|
||||
|
|
|
@ -1,7 +1,7 @@
|
|||
use std::{
|
||||
future::Future,
|
||||
marker::PhantomData,
|
||||
net,
|
||||
mem, net,
|
||||
pin::Pin,
|
||||
rc::Rc,
|
||||
task::{Context, Poll},
|
||||
|
@ -339,21 +339,24 @@ where
|
|||
ref mut srv,
|
||||
ref mut config,
|
||||
ref peer_addr,
|
||||
ref mut on_connect_data,
|
||||
ref mut conn_data,
|
||||
ref mut handshake,
|
||||
) => match ready!(Pin::new(handshake).poll(cx)) {
|
||||
Ok((conn, timer)) => {
|
||||
let on_connect_data = std::mem::take(on_connect_data);
|
||||
let on_connect_data = mem::take(conn_data);
|
||||
|
||||
self.state = State::Incoming(Dispatcher::new(
|
||||
srv.take().unwrap(),
|
||||
conn,
|
||||
on_connect_data,
|
||||
srv.take().unwrap(),
|
||||
config.take().unwrap(),
|
||||
*peer_addr,
|
||||
on_connect_data,
|
||||
timer,
|
||||
));
|
||||
|
||||
self.poll(cx)
|
||||
}
|
||||
|
||||
Err(err) => {
|
||||
trace!("H2 handshake error: {}", err);
|
||||
Poll::Ready(Err(err))
|
||||
|
|
|
@ -53,7 +53,7 @@ pub mod ws;
|
|||
pub use self::builder::HttpServiceBuilder;
|
||||
pub use self::config::{KeepAlive, ServiceConfig};
|
||||
pub use self::error::Error;
|
||||
pub use self::extensions::{CloneableExtensions, Extensions};
|
||||
pub use self::extensions::Extensions;
|
||||
pub use self::header::ContentEncoding;
|
||||
pub use self::http_message::HttpMessage;
|
||||
pub use self::message::ConnectionType;
|
||||
|
@ -76,14 +76,14 @@ pub enum Protocol {
|
|||
Http3,
|
||||
}
|
||||
|
||||
type ConnectCallback<IO> = dyn Fn(&IO, &mut CloneableExtensions);
|
||||
type ConnectCallback<IO> = dyn Fn(&IO, &mut Extensions);
|
||||
|
||||
/// Container for data that extract with ConnectCallback.
|
||||
///
|
||||
/// # Implementation Details
|
||||
/// Uses Option to reduce necessary allocations when merging with request extensions.
|
||||
#[derive(Default)]
|
||||
pub(crate) struct OnConnectData(Option<CloneableExtensions>);
|
||||
pub(crate) struct OnConnectData(Option<Extensions>);
|
||||
|
||||
impl OnConnectData {
|
||||
/// Construct by calling the on-connect callback with the underlying transport I/O.
|
||||
|
@ -92,19 +92,11 @@ impl OnConnectData {
|
|||
on_connect_ext: Option<&ConnectCallback<T>>,
|
||||
) -> Self {
|
||||
let ext = on_connect_ext.map(|handler| {
|
||||
let mut extensions = CloneableExtensions::default();
|
||||
let mut extensions = Extensions::default();
|
||||
handler(io, &mut extensions);
|
||||
extensions
|
||||
});
|
||||
|
||||
Self(ext)
|
||||
}
|
||||
|
||||
/// Merge self into given request's extensions.
|
||||
#[inline]
|
||||
pub(crate) fn merge_into(&mut self, req: &mut Request) {
|
||||
if let Some(ref ext) = self.0 {
|
||||
req.head.extensions.get_mut().clone_from(ext);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
@ -2,7 +2,9 @@
|
|||
|
||||
use std::{
|
||||
cell::{Ref, RefMut},
|
||||
fmt, net, str,
|
||||
fmt, net,
|
||||
rc::Rc,
|
||||
str,
|
||||
};
|
||||
|
||||
use http::{header, Method, Uri, Version};
|
||||
|
@ -19,6 +21,7 @@ use crate::{
|
|||
pub struct Request<P = PayloadStream> {
|
||||
pub(crate) payload: Payload<P>,
|
||||
pub(crate) head: Message<RequestHead>,
|
||||
pub(crate) conn_data: Option<Rc<Extensions>>,
|
||||
}
|
||||
|
||||
impl<P> HttpMessage for Request<P> {
|
||||
|
@ -51,6 +54,7 @@ impl From<Message<RequestHead>> for Request<PayloadStream> {
|
|||
Request {
|
||||
head,
|
||||
payload: Payload::None,
|
||||
conn_data: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -61,6 +65,7 @@ impl Request<PayloadStream> {
|
|||
Request {
|
||||
head: Message::new(),
|
||||
payload: Payload::None,
|
||||
conn_data: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -71,16 +76,19 @@ impl<P> Request<P> {
|
|||
Request {
|
||||
payload,
|
||||
head: Message::new(),
|
||||
conn_data: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Create new Request instance
|
||||
pub fn replace_payload<P1>(self, payload: Payload<P1>) -> (Request<P1>, Payload<P>) {
|
||||
let pl = self.payload;
|
||||
|
||||
(
|
||||
Request {
|
||||
payload,
|
||||
head: self.head,
|
||||
conn_data: self.conn_data,
|
||||
},
|
||||
pl,
|
||||
)
|
||||
|
@ -170,6 +178,26 @@ impl<P> Request<P> {
|
|||
pub fn peer_addr(&self) -> Option<net::SocketAddr> {
|
||||
self.head().peer_addr
|
||||
}
|
||||
|
||||
/// Returns a reference a piece of connection data set in an [on-connect] callback.
|
||||
///
|
||||
/// ```ignore
|
||||
/// let opt_t = req.conn_data::<PeerCertificate>();
|
||||
/// ```
|
||||
///
|
||||
/// [on-connect]: crate::HttpServiceBuilder::on_connect_ext
|
||||
pub fn conn_data<T: 'static>(&self) -> Option<&T> {
|
||||
self.conn_data
|
||||
.as_deref()
|
||||
.and_then(|container| container.get::<T>())
|
||||
}
|
||||
|
||||
/// Returns the connection data container if an [on-connect] callback was registered.
|
||||
///
|
||||
/// [on-connect]: crate::HttpServiceBuilder::on_connect_ext
|
||||
pub fn take_conn_data(&mut self) -> Option<Rc<Extensions>> {
|
||||
self.conn_data.take()
|
||||
}
|
||||
}
|
||||
|
||||
impl<P> fmt::Debug for Request<P> {
|
||||
|
|
|
@ -507,8 +507,7 @@ where
|
|||
&self,
|
||||
(io, proto, peer_addr): (T, Protocol, Option<net::SocketAddr>),
|
||||
) -> Self::Future {
|
||||
let on_connect_data =
|
||||
OnConnectData::from_io(&io, self.on_connect_ext.as_deref());
|
||||
let conn_data = OnConnectData::from_io(&io, self.on_connect_ext.as_deref());
|
||||
|
||||
match proto {
|
||||
Protocol::Http2 => HttpServiceHandlerResponse {
|
||||
|
@ -517,7 +516,7 @@ where
|
|||
h2::handshake_with_timeout(io, &self.cfg),
|
||||
self.cfg.clone(),
|
||||
self.flow.clone(),
|
||||
on_connect_data,
|
||||
conn_data,
|
||||
peer_addr,
|
||||
)),
|
||||
},
|
||||
|
@ -527,10 +526,10 @@ where
|
|||
state: State::H1 {
|
||||
dispatcher: h1::Dispatcher::new(
|
||||
io,
|
||||
self.cfg.clone(),
|
||||
self.flow.clone(),
|
||||
on_connect_data,
|
||||
self.cfg.clone(),
|
||||
peer_addr,
|
||||
conn_data,
|
||||
),
|
||||
},
|
||||
},
|
||||
|
@ -627,17 +626,12 @@ where
|
|||
StateProj::H2Handshake { handshake: data } => {
|
||||
match ready!(Pin::new(&mut data.as_mut().unwrap().0).poll(cx)) {
|
||||
Ok((conn, timer)) => {
|
||||
let (_, config, flow, on_connect_data, peer_addr) =
|
||||
let (_, config, flow, conn_data, peer_addr) =
|
||||
data.take().unwrap();
|
||||
|
||||
self.as_mut().project().state.set(State::H2 {
|
||||
dispatcher: h2::Dispatcher::new(
|
||||
flow,
|
||||
conn,
|
||||
on_connect_data,
|
||||
config,
|
||||
peer_addr,
|
||||
timer,
|
||||
conn, flow, config, peer_addr, conn_data, timer,
|
||||
),
|
||||
});
|
||||
self.poll(cx)
|
||||
|
|
|
@ -430,7 +430,7 @@ async fn test_h2_on_connect() {
|
|||
data.insert(20isize);
|
||||
})
|
||||
.h2(|req: Request| {
|
||||
assert!(req.extensions().contains::<isize>());
|
||||
assert!(req.conn_data::<isize>().is_some());
|
||||
ok::<_, Infallible>(Response::ok())
|
||||
})
|
||||
.openssl(tls_config())
|
||||
|
|
|
@ -6,8 +6,10 @@
|
|||
|
||||
use std::{any::Any, io, net::SocketAddr};
|
||||
|
||||
use actix_http::CloneableExtensions;
|
||||
use actix_web::{rt::net::TcpStream, web, App, HttpServer};
|
||||
use actix_web::{
|
||||
dev::Extensions, rt::net::TcpStream, web, App, HttpRequest, HttpResponse, HttpServer,
|
||||
Responder,
|
||||
};
|
||||
|
||||
#[allow(dead_code)]
|
||||
#[derive(Debug, Clone)]
|
||||
|
@ -17,14 +19,19 @@ struct ConnectionInfo {
|
|||
ttl: Option<u32>,
|
||||
}
|
||||
|
||||
async fn route_whoami(conn_info: web::ReqData<ConnectionInfo>) -> String {
|
||||
format!(
|
||||
"Here is some info about your connection:\n\n{:#?}",
|
||||
conn_info
|
||||
)
|
||||
async fn route_whoami(req: HttpRequest) -> impl Responder {
|
||||
match req.conn_data::<ConnectionInfo>() {
|
||||
Some(info) => HttpResponse::Ok().body(format!(
|
||||
"Here is some info about your connection:\n\n{:#?}",
|
||||
info
|
||||
)),
|
||||
None => {
|
||||
HttpResponse::InternalServerError().body("Missing expected request extension data")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn get_conn_info(connection: &dyn Any, data: &mut CloneableExtensions) {
|
||||
fn get_conn_info(connection: &dyn Any, data: &mut Extensions) {
|
||||
if let Some(sock) = connection.downcast_ref::<TcpStream>() {
|
||||
data.insert(ConnectionInfo {
|
||||
bind: sock.local_addr().unwrap(),
|
||||
|
@ -40,9 +47,12 @@ fn get_conn_info(connection: &dyn Any, data: &mut CloneableExtensions) {
|
|||
async fn main() -> io::Result<()> {
|
||||
env_logger::init_from_env(env_logger::Env::new().default_filter_or("info"));
|
||||
|
||||
let bind = ("127.0.0.1", 8080);
|
||||
log::info!("staring server at http://{}:{}", &bind.0, &bind.1);
|
||||
|
||||
HttpServer::new(|| App::new().default_service(web::to(route_whoami)))
|
||||
.on_connect(get_conn_info)
|
||||
.bind(("127.0.0.1", 8080))?
|
||||
.bind(bind)?
|
||||
.workers(1)
|
||||
.run()
|
||||
.await
|
||||
|
|
|
@ -197,7 +197,8 @@ where
|
|||
|
||||
actix_service::forward_ready!(service);
|
||||
|
||||
fn call(&self, req: Request) -> Self::Future {
|
||||
fn call(&self, mut req: Request) -> Self::Future {
|
||||
let conn_data = req.take_conn_data();
|
||||
let (head, payload) = req.into_parts();
|
||||
|
||||
let req = if let Some(mut req) = self.app_state.pool().pop() {
|
||||
|
@ -205,6 +206,7 @@ where
|
|||
inner.path.get_mut().update(&head.uri);
|
||||
inner.path.reset();
|
||||
inner.head = head;
|
||||
inner.conn_data = conn_data;
|
||||
req
|
||||
} else {
|
||||
HttpRequest::new(
|
||||
|
@ -212,6 +214,7 @@ where
|
|||
head,
|
||||
self.app_state.clone(),
|
||||
self.app_data.clone(),
|
||||
conn_data,
|
||||
)
|
||||
};
|
||||
self.service.call(ServiceRequest::new(req, payload))
|
||||
|
|
|
@ -14,10 +14,7 @@ pub use crate::types::form::UrlEncoded;
|
|||
pub use crate::types::json::JsonBody;
|
||||
pub use crate::types::readlines::Readlines;
|
||||
|
||||
pub use actix_http::{
|
||||
CloneableExtensions, Extensions, Payload, PayloadStream, RequestHead, Response,
|
||||
ResponseHead,
|
||||
};
|
||||
pub use actix_http::{Extensions, Payload, PayloadStream, RequestHead, Response, ResponseHead};
|
||||
pub use actix_router::{Path, ResourceDef, ResourcePath, Url};
|
||||
pub use actix_server::{Server, ServerHandle};
|
||||
pub use actix_service::{
|
||||
|
|
|
@ -37,6 +37,7 @@ pub(crate) struct HttpRequestInner {
|
|||
pub(crate) head: Message<RequestHead>,
|
||||
pub(crate) path: Path<Url>,
|
||||
pub(crate) app_data: SmallVec<[Rc<Extensions>; 4]>,
|
||||
pub(crate) conn_data: Option<Rc<Extensions>>,
|
||||
app_state: Rc<AppInitServiceState>,
|
||||
}
|
||||
|
||||
|
@ -47,6 +48,7 @@ impl HttpRequest {
|
|||
head: Message<RequestHead>,
|
||||
app_state: Rc<AppInitServiceState>,
|
||||
app_data: Rc<Extensions>,
|
||||
conn_data: Option<Rc<Extensions>>,
|
||||
) -> HttpRequest {
|
||||
let mut data = SmallVec::<[Rc<Extensions>; 4]>::new();
|
||||
data.push(app_data);
|
||||
|
@ -57,6 +59,7 @@ impl HttpRequest {
|
|||
path,
|
||||
app_state,
|
||||
app_data: data,
|
||||
conn_data,
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
@ -165,6 +168,20 @@ impl HttpRequest {
|
|||
self.head().extensions_mut()
|
||||
}
|
||||
|
||||
/// Returns a reference a piece of connection data set in an [on-connect] callback.
|
||||
///
|
||||
/// ```ignore
|
||||
/// let opt_t = req.conn_data::<PeerCertificate>();
|
||||
/// ```
|
||||
///
|
||||
/// [on-connect]: crate::HttpServiceBuilder::on_connect_ext
|
||||
pub fn conn_data<T: 'static>(&self) -> Option<&T> {
|
||||
self.inner
|
||||
.conn_data
|
||||
.as_deref()
|
||||
.and_then(|container| container.get::<T>())
|
||||
}
|
||||
|
||||
/// Generates URL for a named resource.
|
||||
///
|
||||
/// This substitutes in sequence all URL parameters that appear in the resource itself and in
|
||||
|
|
|
@ -6,9 +6,7 @@ use std::{
|
|||
sync::{Arc, Mutex},
|
||||
};
|
||||
|
||||
use actix_http::{
|
||||
body::MessageBody, CloneableExtensions, HttpService, KeepAlive, Request, Response,
|
||||
};
|
||||
use actix_http::{body::MessageBody, Extensions, HttpService, KeepAlive, Request, Response};
|
||||
use actix_server::{Server, ServerBuilder};
|
||||
use actix_service::{
|
||||
map_config, IntoServiceFactory, Service, ServiceFactory, ServiceFactoryExt as _,
|
||||
|
@ -65,7 +63,7 @@ where
|
|||
backlog: u32,
|
||||
sockets: Vec<Socket>,
|
||||
builder: ServerBuilder,
|
||||
on_connect_fn: Option<Arc<dyn Fn(&dyn Any, &mut CloneableExtensions) + Send + Sync>>,
|
||||
on_connect_fn: Option<Arc<dyn Fn(&dyn Any, &mut Extensions) + Send + Sync>>,
|
||||
_phantom: PhantomData<(S, B)>,
|
||||
}
|
||||
|
||||
|
@ -102,9 +100,8 @@ where
|
|||
}
|
||||
|
||||
/// Sets function that will be called once before each connection is handled.
|
||||
/// It will receive a `&std::any::Any`, which contains underlying connection type and a
|
||||
/// [CloneableExtensions] container so that request-local data can be passed to middleware
|
||||
/// and handlers.
|
||||
/// It will receive a `&std::any::Any`, which contains underlying connection type and an
|
||||
/// [Extensions] container so that connection data can be accessed in middleware and handlers.
|
||||
///
|
||||
/// # Connection Types
|
||||
/// - `actix_tls::accept::openssl::TlsStream<actix_web::rt::net::TcpStream>` when using openssl.
|
||||
|
@ -114,7 +111,7 @@ where
|
|||
/// See the `on_connect` example for additional details.
|
||||
pub fn on_connect<CB>(self, f: CB) -> HttpServer<F, I, S, B>
|
||||
where
|
||||
CB: Fn(&dyn Any, &mut CloneableExtensions) + Send + Sync + 'static,
|
||||
CB: Fn(&dyn Any, &mut Extensions) + Send + Sync + 'static,
|
||||
{
|
||||
HttpServer {
|
||||
factory: self.factory,
|
||||
|
|
24
src/test.rs
24
src/test.rs
|
@ -581,7 +581,13 @@ impl TestRequest {
|
|||
let app_state = AppInitServiceState::new(Rc::new(self.rmap), self.config.clone());
|
||||
|
||||
ServiceRequest::new(
|
||||
HttpRequest::new(self.path, head, app_state, Rc::new(self.app_data)),
|
||||
HttpRequest::new(
|
||||
self.path,
|
||||
head,
|
||||
app_state,
|
||||
Rc::new(self.app_data),
|
||||
Default::default(),
|
||||
),
|
||||
payload,
|
||||
)
|
||||
}
|
||||
|
@ -599,7 +605,13 @@ impl TestRequest {
|
|||
|
||||
let app_state = AppInitServiceState::new(Rc::new(self.rmap), self.config.clone());
|
||||
|
||||
HttpRequest::new(self.path, head, app_state, Rc::new(self.app_data))
|
||||
HttpRequest::new(
|
||||
self.path,
|
||||
head,
|
||||
app_state,
|
||||
Rc::new(self.app_data),
|
||||
Default::default(),
|
||||
)
|
||||
}
|
||||
|
||||
/// Complete request creation and generate `HttpRequest` and `Payload` instances
|
||||
|
@ -610,7 +622,13 @@ impl TestRequest {
|
|||
|
||||
let app_state = AppInitServiceState::new(Rc::new(self.rmap), self.config.clone());
|
||||
|
||||
let req = HttpRequest::new(self.path, head, app_state, Rc::new(self.app_data));
|
||||
let req = HttpRequest::new(
|
||||
self.path,
|
||||
head,
|
||||
app_state,
|
||||
Rc::new(self.app_data),
|
||||
Default::default(),
|
||||
);
|
||||
|
||||
(req, payload)
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue