chore: prepare dquic v0.6.0 release

Cargo.toml

@@ -31,7 +31,7 @@ default-members = [
 ]
 
 [workspace.package]
-version = "0.5.1"
+version = "0.6.0"
 edition = "2024"
 readme = "README.md"
 repository = "https://github.com/genmeta/dquic"
@@ -97,17 +97,17 @@ tracing-appender = "0.2"
 
 # members
 qmacro = { path = "./qmacro", version = "0.5.1" }
-qbase = { path = "./qbase", version = "0.5.1" }
-qevent = { path = "./qevent", version = "0.5.1" }
-qudp = { path = "./qudp", version = "0.5.1" }
-qinterface = { path = "./qinterface", version = "0.5.1" }
-qdatagram = { path = "./qdatagram", version = "0.5.1" }
-qresolve = { path = "./qresolve", version = "0.5.1" }
-qrecovery = { path = "./qrecovery", version = "0.5.1" }
-qtraversal = { path = "./qtraversal", version = "0.5.1" }
-qcongestion = { path = "./qcongestion", version = "0.5.1" }
-qconnection = { path = "./qconnection", version = "0.5.2" }
-dquic = { path = "./dquic", version = "0.5.1" }
+qbase = { path = "./qbase", version = "0.6.0" }
+qevent = { path = "./qevent", version = "0.6.0" }
+qudp = { path = "./qudp", version = "0.6.0" }
+qinterface = { path = "./qinterface", version = "0.6.0" }
+qdatagram = { path = "./qdatagram", version = "0.6.0" }
+qresolve = { path = "./qresolve", version = "0.6.0" }
+qrecovery = { path = "./qrecovery", version = "0.6.0" }
+qtraversal = { path = "./qtraversal", version = "0.6.0" }
+qcongestion = { path = "./qcongestion", version = "0.6.0" }
+qconnection = { path = "./qconnection", version = "0.6.0" }
+dquic = { path = "./dquic", version = "0.6.0" }
 h3-shim = { path = "./h3-shim", version = "0.5.1" }
 
 

dquic/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dquic"
-version = "0.5.1"
+version = "0.6.0"
 edition.workspace = true
 description = "An IETF quic transport protocol implemented natively using async Rust"
 readme = "README.md"

dquic/src/client.rs

@@ -16,7 +16,10 @@ use qbase::{net::Family, param::ClientParameters, token::TokenSink};
 use qconnection::{
     self,
     qbase::net::AddrFamily,
-    qinterface::{component::location::Locations, io::IO},
+    qinterface::{
+        component::local_endpoint::{InterfaceEndpointUpdate, LocalEndpoints},
+        io::IO,
+    },
 };
 use qevent::telemetry::QLog;
 use qinterface::{
@@ -29,6 +32,7 @@ use rustls::{
     client::{ResolvesClientCert, WantsClientCert},
 };
 use thiserror::Error;
+use tracing::Instrument as _;
 
 use crate::{prelude::*, *};
 
@@ -123,19 +127,68 @@ impl QuicClient {
     /// This is useful for advanced scenarios where you need fine-grained control
     /// over which interfaces and paths are used for the connection.
     pub fn new_connection(&self, server_name: impl Into<String>) -> Arc<Connection> {
-        Connection::new_client(server_name.into(), self.token_sink.clone())
+        let connection = Connection::new_client(server_name.into(), self.token_sink.clone())
             .with_parameters(self.parameters.clone())
             .with_tls_config(self.tls_config.clone())
             .with_streams_concurrency_strategy(self.stream_strategy_factory.as_ref())
             .with_zero_rtt(self.tls_config.enable_early_data)
             .with_iface_factory(self.network.iface_factory.clone())
             .with_iface_manager(self.network.iface_manager.clone())
             .with_quic_router(self.network.quic_router.clone())
-            .with_locations(self.network.locations.clone())
             .with_defer_idle_timeout(self.defer_idle_timeout)
             .with_cids(ConnectionId::random_gen(8))
             .with_qlog(self.qlogger.clone())
-            .run()
+            .run();
+        self.subscribe_connection_local_endpoints(connection.clone());
+        connection
+    }
+
+    fn subscribe_connection_local_endpoints(&self, connection: Arc<Connection>) {
+        let mut subscriber = self.network.local_endpoints.subscribe();
+        let weak = Arc::downgrade(&connection);
+
+        // Inherent termination: this task exits when the connection drops,
+        // the connection terminates, or the LocalEndpoints subscriber ends.
+        tokio::spawn(
+            async move {
+                loop {
+                    let Some(terminated) = weak.upgrade().map(|connection| connection.terminated())
+                    else {
+                        break;
+                    };
+                    tokio::select! {
+                        biased;
+                        _ = terminated => break,
+                        update = subscriber.recv() => {
+                            let Some((bind_uri, update)) = update else { break };
+                            let Some(connection) = weak.upgrade() else { break };
+                            Self::feed_connection_local_endpoint(&connection, bind_uri, update);
+                        }
+                    }
+                }
+            }
+            .in_current_span(),
+        );
+    }
+
+    fn feed_connection_local_endpoint(
+        connection: &Connection,
+        bind_uri: BindUri,
+        update: InterfaceEndpointUpdate,
+    ) {
+        let result = match update {
+            InterfaceEndpointUpdate::Upsert { key, endpoint } => {
+                connection.upsert_local_endpoint(bind_uri, key, endpoint)
+            }
+            InterfaceEndpointUpdate::Remove { key } => {
+                connection.remove_local_endpoint(&bind_uri, &key)
+            }
+            InterfaceEndpointUpdate::Close => connection.close_local_endpoints(&bind_uri),
+        };
+
+        if let Err(error) = result {
+            tracing::trace!(target: "quic", ?error, "failed to feed local endpoint update to client connection");
+        }
     }
 
     /// Builds a [`BindUri`] from the DNS [`Source`] and endpoint address.
@@ -249,7 +302,7 @@ impl QuicClient {
     /// let paths = quic_client.probe(server_addresses).await?;
     /// let connection = quic_client.new_connection("genmeta.net");
     /// for (iface, link, pathway) in paths {
-    ///     connection.add_path(iface.bind_uri(), link, pathway)?;
+    ///     connection.add_path((iface.bind_uri(), pathway, link))?;
     /// }
     /// # Ok(())
     /// # }
@@ -303,7 +356,7 @@ impl QuicClient {
         let paths = self.probe([(source, server_ep)]).await?;
         let has_direct_path = !paths.is_empty();
         for (iface, link, pathway) in paths {
-            _ = connection.add_path(iface.bind_uri(), link, pathway);
+            _ = connection.add_path((iface.bind_uri(), pathway, link));
         }
         Ok(has_direct_path)
     }
@@ -329,7 +382,6 @@ impl QuicClient {
         server_eps: impl IntoIterator<Item = (Source, EndpointAddr)>,
     ) -> Result<Arc<Connection>, ConnectServerError> {
         let connection = self.new_connection(server_name);
-        _ = connection.subscribe_local_address();
         for (source, server_ep) in server_eps {
             self.setup_server_endpoint(&connection, source, server_ep)
                 .await
@@ -360,10 +412,6 @@ impl QuicClient {
             .map_err(|source| ConnectServerError::Dns { source })?;
 
         let connection = self.new_connection(server);
-        if connection.subscribe_local_address().is_err() {
-            // connection already closed, return immediately (not connect error)
-            return Ok(connection);
-        }
 
         let mut last_error: Option<ConnectServerError> = None;
 
@@ -527,11 +575,11 @@ impl<T> QuicClientBuilder<T> {
         self
     }
 
-    /// Specify the locations for interface sharing.
+    /// Specify the local endpoints for interface sharing.
     ///
-    /// The given locations is shared by all connections created by this client.
-    pub fn with_locations(mut self, locations: Arc<Locations>) -> Self {
-        self.network.locations = locations;
+    /// The given local endpoints hub is shared by all connections created by this client.
+    pub fn with_local_endpoints(mut self, local_endpoints: Arc<LocalEndpoints>) -> Self {
+        self.network.local_endpoints = local_endpoints;
         self
     }
 
@@ -855,3 +903,19 @@ impl QuicClientBuilder<TlsClientConfig> {
         }
     }
 }
+
+#[cfg(test)]
+mod local_endpoint_subscription_tests {
+    #[test]
+    fn quic_client_subscribes_per_connection_without_qconnection_subscription_api() {
+        let source = include_str!("client.rs");
+        let production = source
+            .split("#[cfg(test)]")
+            .next()
+            .expect("test module boundary");
+        assert!(production.contains(concat!("fn subscribe_connection_", "local_endpoints")));
+        assert!(production.contains(concat!("local_endpoints", ".subscribe()")));
+        assert!(production.contains(concat!("InterfaceEndpoint", "Update::Upsert")));
+        assert!(!production.contains(concat!("subscribe_local_", "address_events")));
+    }
+}

dquic/src/common.rs

@@ -9,7 +9,7 @@ use qconnection::{
         component::{
             Components,
             alive::RebindOnNetworkChangedComponent,
-            location::{Locations, LocationsComponent},
+            local_endpoint::{LocalEndpoints, LocalEndpointsComponent},
             route::{QuicRouter, QuicRouterComponent},
         },
         device::Devices,
@@ -31,7 +31,7 @@ pub struct Network {
     pub iface_manager: Arc<InterfaceManager>,
     pub quic_router: Arc<QuicRouter>,
     pub stun_server: Option<Arc<str>>,
-    pub locations: Arc<Locations>,
+    pub local_endpoints: Arc<LocalEndpoints>,
 }
 
 impl Default for Network {
@@ -43,7 +43,7 @@ impl Default for Network {
             iface_manager: InterfaceManager::global().clone(),
             quic_router: QuicRouter::global().clone(),
             stun_server: None,
-            locations: Arc::new(Locations::new()),
+            local_endpoints: Arc::new(LocalEndpoints::new()),
         }
     }
 }
@@ -58,8 +58,10 @@ impl Network {
                 .init_with(|| QuicRouterComponent::new(self.quic_router.clone()))
                 .router();
 
-            let locations = components
-                .init_with(|| LocationsComponent::new(iface.downgrade(), self.locations.clone()))
+            let local_endpoints = components
+                .init_with(|| {
+                    LocalEndpointsComponent::new(iface.downgrade(), self.local_endpoints.clone())
+                })
                 .clone();
 
             match &stun_server {
@@ -79,7 +81,7 @@ impl Network {
                                 self.resolver.clone(),
                                 stun_server,
                                 iter::empty(),
-                                Some(locations.clone()),
+                                Some(local_endpoints.clone()),
                             )
                         })
                         .clone();

dquic/src/server.rs

@@ -19,7 +19,12 @@ use qbase::{
 };
 use qconnection::{
     self,
-    qinterface::{self, bind_uri::BindUri, component::location::Locations, device::Devices},
+    qinterface::{
+        self,
+        bind_uri::BindUri,
+        component::local_endpoint::{InterfaceEndpointUpdate, LocalEndpoints},
+        device::Devices,
+    },
     tls::AcceptAllClientAuther,
 };
 use qevent::telemetry::QLog;
@@ -374,6 +379,57 @@ impl AuthClient for ServerAuther {
 
 // internal methods
 impl QuicListeners {
+    fn subscribe_connection_local_endpoints(
+        local_endpoints: Arc<LocalEndpoints>,
+        connection: Arc<Connection>,
+    ) {
+        let mut subscriber = local_endpoints.subscribe();
+        let weak = Arc::downgrade(&connection);
+
+        // Inherent termination: this task exits when the connection drops,
+        // the connection terminates, or the LocalEndpoints subscriber ends.
+        tokio::spawn(
+            async move {
+                loop {
+                    let Some(terminated) = weak.upgrade().map(|connection| connection.terminated())
+                    else {
+                        break;
+                    };
+                    tokio::select! {
+                        biased;
+                        _ = terminated => break,
+                        update = subscriber.recv() => {
+                            let Some((bind_uri, update)) = update else { break };
+                            let Some(connection) = weak.upgrade() else { break };
+                            Self::feed_connection_local_endpoint(&connection, bind_uri, update);
+                        }
+                    }
+                }
+            }
+            .in_current_span(),
+        );
+    }
+
+    fn feed_connection_local_endpoint(
+        connection: &Connection,
+        bind_uri: BindUri,
+        update: InterfaceEndpointUpdate,
+    ) {
+        let result = match update {
+            InterfaceEndpointUpdate::Upsert { key, endpoint } => {
+                connection.upsert_local_endpoint(bind_uri, key, endpoint)
+            }
+            InterfaceEndpointUpdate::Remove { key } => {
+                connection.remove_local_endpoint(&bind_uri, &key)
+            }
+            InterfaceEndpointUpdate::Close => connection.close_local_endpoints(&bind_uri),
+        };
+
+        if let Err(error) = result {
+            tracing::trace!(target: "quic_listeners", ?error, "failed to feed local endpoint update to accepted connection");
+        }
+    }
+
     #[tracing::instrument(
         target = "quic_listeners", level = "debug", skip_all, 
         fields(%bind_uri, %pathway, %link, odcid=tracing::field::Empty, server_name=tracing::field::Empty)
@@ -416,7 +472,6 @@ impl QuicListeners {
             .with_iface_factory(self.network.iface_factory.clone())
             .with_iface_manager(self.network.iface_manager.clone())
             .with_quic_router(self.network.quic_router.clone())
-            .with_locations(self.network.locations.clone())
             // todo
             // .with_stun_servers()
             .with_cids(origin_dcid)
@@ -425,14 +480,15 @@ impl QuicListeners {
 
         let incomings = self.incomings.clone();
         let quic_router = self.network.quic_router.clone();
+        let local_endpoints = self.network.local_endpoints.clone();
 
         let try_accept_connection = async move {
             quic_router.deliver(packet, (bind_uri, pathway, link)).await;
 
             match connection.server_name().await {
                 Ok(server_name) => {
                     tracing::Span::current().record("server_name", &server_name);
-                    _ = connection.subscribe_local_address();
+                    Self::subscribe_connection_local_endpoints(local_endpoints, connection.clone());
                     let incoming = (connection, server_name, pathway, link);
                     match incomings.send((incoming, premit)).await {
                         Ok(..) => {
@@ -565,11 +621,11 @@ impl<T> QuicListenersBuilder<T> {
         self
     }
 
-    /// Specify the locations for interface sharing.
+    /// Specify the local endpoints for interface sharing.
     ///
-    /// The given locations is shared by all connections created by this listeners.
-    pub fn with_locations(mut self, locations: Arc<Locations>) -> Self {
-        self.network.locations = locations;
+    /// The given local endpoints hub is shared by all connections created by these listeners.
+    pub fn with_local_endpoints(mut self, local_endpoints: Arc<LocalEndpoints>) -> Self {
+        self.network.local_endpoints = local_endpoints;
         self
     }
 
@@ -833,3 +889,21 @@ impl QuicListenersBuilder<TlsServerConfig> {
         Ok(quic_listeners)
     }
 }
+
+#[cfg(test)]
+mod local_endpoint_subscription_tests {
+    #[test]
+    fn quic_listeners_subscribe_accepted_connections_without_qconnection_subscription_api() {
+        let source = include_str!("server.rs");
+        let production = source
+            .split("#[cfg(test)]")
+            .next()
+            .expect("test module boundary");
+        assert!(production.contains(concat!("fn subscribe_connection_", "local_endpoints")));
+        assert!(production.contains(concat!("local_endpoints", ".subscribe()")));
+        assert!(production.contains(concat!("InterfaceEndpoint", "Update::Upsert")));
+        assert!(!production.contains(concat!("subscribe_local_", "address_events")));
+        assert!(!production.contains("let listeners = self.clone();"));
+        assert!(!production.contains("listeners.subscribe_connection_local_endpoints"));
+    }
+}