Merge remote-tracking branch 'origin/master' into on-connect-fix

CHANGES.md

@@ -7,18 +7,24 @@
 * `Range` typed header. [#2485]
 * `HttpResponse::map_into_{left,right}_body` and `HttpResponse::map_into_boxed_body`. [#2468]
 * `ServiceResponse::map_into_{left,right}_body` and `HttpResponse::map_into_boxed_body`. [#2468]
+* `HttpServer::on_connect` now receives a `CloneableExtensions` object. [#2327]
+
+[#2325]: https://github.com/actix/actix-web/pull/2325
+[#2327]: https://github.com/actix/actix-web/pull/2327
 
 ### Changed
 * Rename `Accept::{mime_precedence => ranked}`. [#2480]
 * Rename `Accept::{mime_preference => preference}`. [#2480]
 * Un-deprecate `App::data_factory`. [#2484]
 * `HttpRequest::url_for` no longer constructs URLs with query or fragment components. [#2430]
+* `HttpServer::on_connect` now receives a `CloneableExtensions` object. [#2327]
 
 ### Fixed
 * Accept wildcard `*` items in `AcceptLanguage`. [#2480]
 * Re-exports `dev::{BodySize, MessageBody, SizedStream}`. They are exposed through the `body` module. [#2468]
 * Typed headers containing lists that require one or more items now enforce this minimum. [#2482]
 
+[#2327]: https://github.com/actix/actix-web/pull/2327
 [#2430]: https://github.com/actix/actix-web/pull/2430
 [#2468]: https://github.com/actix/actix-web/pull/2468
 [#2480]: https://github.com/actix/actix-web/pull/2480

actix-http/CHANGES.md

@@ -14,6 +14,7 @@
 * `header::QualityItem::{max, min}`. [#2486]
 * `header::Quality::{MAX, MIN}`. [#2486]
 * `impl Display` for `header::Quality`. [#2486]
+* `CloneableExtensions` object for use in `on_connect` handlers. [#2327]
 
 ### Changed
 * Rename `body::BoxBody::{from_body => new}`. [#2468]
@@ -23,6 +24,7 @@
 * `From` implementations on error types now return a `Response<BoxBody>`. [#2468]
 * `ResponseBuilder::body(B)` now returns `Response<EitherBody<B>>`. [#2468]
 * `ResponseBuilder::finish()` now returns `Response<EitherBody<()>>`. [#2468]
+* `on_connect_ext` methods now receive a `CloneableExtensions` object. [#2327]
 
 ### Removed
 * `ResponseBuilder::streaming`. [#2468]
@@ -34,6 +36,7 @@
 * `impl TryFrom<u16>` for `header::Quality`. [#2486]
 * `http` module. Most everything it contained is exported at the crate root. [#2488]
 
+[#2327]: https://github.com/actix/actix-web/pull/2327
 [#2483]: https://github.com/actix/actix-web/pull/2483
 [#2468]: https://github.com/actix/actix-web/pull/2468
 [#1920]: https://github.com/actix/actix-web/pull/1920

actix-http/src/builder.rs

@@ -6,10 +6,11 @@ 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, Extensions, Request, Response,
+    ConnectCallback, Request, Response,
 };
 
 /// A HTTP service builder
@@ -167,7 +168,7 @@ where
     /// and handlers.
     pub fn on_connect_ext<F>(mut self, f: F) -> Self
     where
-        F: Fn(&T, &mut Extensions) + 'static,
+        F: Fn(&T, &mut CloneableExtensions) + 'static,
     {
         self.on_connect_ext = Some(Rc::new(f));
         self

actix-http/src/extensions.rs

@@ -1,6 +1,6 @@
 use std::{
     any::{Any, TypeId},
-    fmt, mem,
+    fmt,
 };
 
 use ahash::AHashMap;
@@ -10,8 +10,7 @@ use ahash::AHashMap;
 /// All entries into this map must be owned types (or static references).
 #[derive(Default)]
 pub struct Extensions {
-    /// Use FxHasher with a std HashMap with for faster
+    /// Use AHasher with a std HashMap with for faster lookups on the small `TypeId` keys.
-    /// lookups on the small `TypeId` (u64 equivalent) keys.
     map: AHashMap<TypeId, Box<dyn Any>>,
 }
 
@@ -124,9 +123,11 @@ impl Extensions {
         self.map.extend(other.map);
     }
 
-    /// Sets (or overrides) items from `other` into this map.
+    /// Sets (or overrides) items from cloneable extensions map into this map.
-    pub(crate) fn drain_from(&mut self, other: &mut Self) {
+    pub(crate) fn clone_from(&mut self, other: &CloneableExtensions) {
-        self.map.extend(mem::take(&mut other.map));
+        for (k, val) in &other.map {
+            self.map.insert(*k, (**val).clone_to_any());
+        }
     }
 }
 
@@ -140,6 +141,104 @@ 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::*;
@@ -179,6 +278,8 @@ mod tests {
 
     #[test]
     fn test_integers() {
+        static A: u32 = 8;
+
         let mut map = Extensions::new();
 
         map.insert::<i8>(8);
@@ -191,6 +292,7 @@ mod tests {
         map.insert::<u32>(32);
         map.insert::<u64>(64);
         map.insert::<u128>(128);
+        map.insert::<&'static u32>(&A);
         assert!(map.get::<i8>().is_some());
         assert!(map.get::<i16>().is_some());
         assert!(map.get::<i32>().is_some());
@@ -201,6 +303,7 @@ mod tests {
         assert!(map.get::<u32>().is_some());
         assert!(map.get::<u64>().is_some());
         assert!(map.get::<u128>().is_some());
+        assert!(map.get::<&'static u32>().is_some());
     }
 
     #[test]
@@ -281,25 +384,41 @@ mod tests {
     }
 
     #[test]
-    fn test_drain_from() {
+    fn test_clone_from() {
+        #[derive(Clone)]
+        struct NonCopy {
+            num: u8,
+        }
+
         let mut ext = Extensions::new();
         ext.insert(2isize);
 
-        let mut more_ext = Extensions::new();
-
-        more_ext.insert(5isize);
-        more_ext.insert(5usize);
-
         assert_eq!(ext.get::<isize>(), Some(&2isize));
-        assert_eq!(ext.get::<usize>(), None);
-        assert_eq!(more_ext.get::<isize>(), Some(&5isize));
-        assert_eq!(more_ext.get::<usize>(), Some(&5usize));
 
-        ext.drain_from(&mut more_ext);
+        let mut more_ext = CloneableExtensions::default();
+        more_ext.insert(3isize);
+        more_ext.insert(3usize);
+        more_ext.insert(NonCopy { num: 8 });
 
-        assert_eq!(ext.get::<isize>(), Some(&5isize));
+        ext.clone_from(&more_ext);
-        assert_eq!(ext.get::<usize>(), Some(&5usize));
+
-        assert_eq!(more_ext.get::<isize>(), None);
+        assert_eq!(ext.get::<isize>(), Some(&3isize));
-        assert_eq!(more_ext.get::<usize>(), None);
+        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 ());
     }
 }

actix-http/src/lib.rs

@@ -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::Extensions;
+pub use self::extensions::{CloneableExtensions, 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 Extensions);
+type ConnectCallback<IO> = dyn Fn(&IO, &mut CloneableExtensions);
 
 /// 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<Extensions>);
+pub(crate) struct OnConnectData(Option<CloneableExtensions>);
 
 impl OnConnectData {
     /// Construct by calling the on-connect callback with the underlying transport I/O.
@@ -92,7 +92,7 @@ impl OnConnectData {
         on_connect_ext: Option<&ConnectCallback<T>>,
     ) -> Self {
         let ext = on_connect_ext.map(|handler| {
-            let mut extensions = Extensions::new();
+            let mut extensions = CloneableExtensions::default();
             handler(io, &mut extensions);
             extensions
         });
@@ -103,8 +103,8 @@ impl OnConnectData {
     /// Merge self into given request's extensions.
     #[inline]
     pub(crate) fn merge_into(&mut self, req: &mut Request) {
-        if let Some(ref mut ext) = self.0 {
+        if let Some(ref ext) = self.0 {
-            req.head.extensions.get_mut().drain_from(ext);
+            req.head.extensions.get_mut().clone_from(ext);
         }
     }
 }

examples/on_connect.rs

@@ -6,7 +6,8 @@
 
 use std::{any::Any, io, net::SocketAddr};
 
-use actix_web::{dev::Extensions, rt::net::TcpStream, web, App, HttpServer};
+use actix_http::CloneableExtensions;
+use actix_web::{rt::net::TcpStream, web, App, HttpServer};
 
 #[allow(dead_code)]
 #[derive(Debug, Clone)]
@@ -23,7 +24,7 @@ async fn route_whoami(conn_info: web::ReqData<ConnectionInfo>) -> String {
     )
 }
 
-fn get_conn_info(connection: &dyn Any, data: &mut Extensions) {
+fn get_conn_info(connection: &dyn Any, data: &mut CloneableExtensions) {
     if let Some(sock) = connection.downcast_ref::<TcpStream>() {
         data.insert(ConnectionInfo {
             bind: sock.local_addr().unwrap(),

src/dev.rs

@@ -14,7 +14,10 @@ pub use crate::types::form::UrlEncoded;
 pub use crate::types::json::JsonBody;
 pub use crate::types::readlines::Readlines;
 
-pub use actix_http::{Extensions, Payload, PayloadStream, RequestHead, Response, ResponseHead};
+pub use actix_http::{
+    CloneableExtensions, Extensions, Payload, PayloadStream, RequestHead, Response,
+    ResponseHead,
+};
 pub use actix_router::{Path, ResourceDef, ResourcePath, Url};
 pub use actix_server::{Server, ServerHandle};
 pub use actix_service::{

src/server.rs

@@ -6,7 +6,9 @@ use std::{
     sync::{Arc, Mutex},
 };
 
-use actix_http::{body::MessageBody, Extensions, HttpService, KeepAlive, Request, Response};
+use actix_http::{
+    body::MessageBody, CloneableExtensions, HttpService, KeepAlive, Request, Response,
+};
 use actix_server::{Server, ServerBuilder};
 use actix_service::{
     map_config, IntoServiceFactory, Service, ServiceFactory, ServiceFactoryExt as _,
@@ -63,7 +65,7 @@ where
     backlog: u32,
     sockets: Vec<Socket>,
     builder: ServerBuilder,
-    on_connect_fn: Option<Arc<dyn Fn(&dyn Any, &mut Extensions) + Send + Sync>>,
+    on_connect_fn: Option<Arc<dyn Fn(&dyn Any, &mut CloneableExtensions) + Send + Sync>>,
     _phantom: PhantomData<(S, B)>,
 }
 
@@ -100,10 +102,11 @@ 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 an
+    /// It will receive a `&std::any::Any`, which contains underlying connection type and a
-    /// [Extensions] container so that request-local data can be passed to middleware and handlers.
+    /// [CloneableExtensions] container so that request-local data can be passed to middleware
+    /// and handlers.
     ///
-    /// For example:
+    /// # Connection Types
     /// - `actix_tls::accept::openssl::TlsStream<actix_web::rt::net::TcpStream>` when using openssl.
     /// - `actix_tls::accept::rustls::TlsStream<actix_web::rt::net::TcpStream>` when using rustls.
     /// - `actix_web::rt::net::TcpStream` when no encryption is used.
@@ -111,7 +114,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 Extensions) + Send + Sync + 'static,
+        CB: Fn(&dyn Any, &mut CloneableExtensions) + Send + Sync + 'static,
     {
         HttpServer {
             factory: self.factory,