Expose API for PublicAddresses

src/protocol/libp2p/identify.rs

@@ -33,12 +33,14 @@ use crate::{
 
 use futures::{future::BoxFuture, stream::FuturesUnordered, Stream, StreamExt};
 use multiaddr::Multiaddr;
+use parking_lot::RwLock;
 use prost::Message;
 use tokio::sync::mpsc::{channel, Sender};
 use tokio_stream::wrappers::ReceiverStream;
 
 use std::{
     collections::{HashMap, HashSet},
+    sync::Arc,
     time::Duration,
 };
 
@@ -62,6 +64,54 @@ mod identify_schema {
     include!(concat!(env!("OUT_DIR"), "/identify.rs"));
 }
 
+/// Set of the public addresses of the local node.
+///
+/// These addresses are reported to the identify protocol.
+#[derive(Debug, Clone)]
+pub struct IdentifyPublicAddresses {
+    inner: Arc<RwLock<HashSet<Multiaddr>>>,
+}
+
+impl IdentifyPublicAddresses {
+    /// Create new [`IdentifyPublicAddresses`].
+    fn new() -> Self {
+        Self {
+            inner: Arc::new(RwLock::new(HashSet::new())),
+        }
+    }
+
+    /// Add public address.
+    pub fn add(&self, address: Multiaddr) {
+        if address.is_empty() {
+            return;
+        }
+
+        self.inner.write().insert(address);
+    }
+
+    /// Remove public address.
+    pub fn remove(&self, address: &Multiaddr) {
+        self.inner.write().remove(address);
+    }
+
+    /// Get public addresses.
+    pub fn get_addresses(&self) -> Vec<Multiaddr> {
+        self.inner.read().iter().cloned().collect()
+    }
+
+    /// Get public addresses in the identify raw format.
+    fn get_raw_addresses(&self) -> Vec<Vec<u8>> {
+        self.inner.read().iter().map(|address| address.to_vec()).collect()
+    }
+
+    /// Extend public addresses.
+    pub fn extend(&self, addresses: impl IntoIterator<Item = Multiaddr>) {
+        self.inner
+            .write()
+            .extend(addresses.into_iter().filter(|address| !address.is_empty()));
+    }
+}
+
 /// Identify configuration.
 pub struct Config {
     /// Protocol name.
@@ -80,7 +130,7 @@ pub struct Config {
     pub(crate) protocols: Vec<ProtocolName>,
 
     /// Public addresses.
-    pub(crate) public_addresses: Vec<Multiaddr>,
+    pub(crate) public_addresses: IdentifyPublicAddresses,
 
     /// Protocol version.
     pub(crate) protocol_version: String,
@@ -98,20 +148,27 @@ impl Config {
         protocol_version: String,
         user_agent: Option<String>,
         public_addresses: Vec<Multiaddr>,
-    ) -> (Self, Box<dyn Stream<Item = IdentifyEvent> + Send + Unpin>) {
+    ) -> (
+        Self,
+        IdentifyPublicAddresses,
+        Box<dyn Stream<Item = IdentifyEvent> + Send + Unpin>,
+    ) {
         let (tx_event, rx_event) = channel(DEFAULT_CHANNEL_SIZE);
+        let identify_public_addresses = IdentifyPublicAddresses::new();
+        identify_public_addresses.extend(public_addresses);
 
         (
             Self {
                 tx_event,
                 public: None,
-                public_addresses,
+                public_addresses: identify_public_addresses.clone(),
                 protocol_version,
                 user_agent,
                 codec: ProtocolCodec::UnsignedVarint(Some(IDENTIFY_PAYLOAD_SIZE)),
                 protocols: Vec::new(),
                 protocol: ProtocolName::from(PROTOCOL_NAME),
             },
+            identify_public_addresses,
             Box::new(ReceiverStream::new(rx_event)),
         )
     }
@@ -183,7 +240,7 @@ pub(crate) struct Identify {
     user_agent: String,
 
     /// Public addresses.
-    listen_addresses: HashSet<Multiaddr>,
+    listen_addresses: IdentifyPublicAddresses,
 
     /// Protocols supported by the local node, filled by `Litep2p`.
     protocols: Vec<String>,
@@ -205,11 +262,13 @@ impl Identify {
         config: Config,
         listen_addresses: Vec<Multiaddr>,
     ) -> Self {
+        config.public_addresses.extend(listen_addresses);
+
         Self {
             service,
             tx: config.tx_event,
             peers: HashMap::new(),
-            listen_addresses: config.public_addresses.into_iter().chain(listen_addresses).collect(),
+            listen_addresses: config.public_addresses,
             public: config.public.expect("public key to be supplied"),
             protocol_version: config.protocol_version,
             user_agent: config.user_agent.unwrap_or(DEFAULT_AGENT.to_string()),
@@ -269,11 +328,7 @@ impl Identify {
             protocol_version: Some(self.protocol_version.clone()),
             agent_version: Some(self.user_agent.clone()),
             public_key: Some(self.public.to_protobuf_encoding()),
-            listen_addrs: self
-                .listen_addresses
-                .iter()
-                .map(|address| address.to_vec())
-                .collect::<Vec<_>>(),
+            listen_addrs: self.listen_addresses.get_raw_addresses(),
             observed_addr,
             protocols: self.protocols.clone(),
         };
@@ -413,3 +468,74 @@ impl Identify {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{config::ConfigBuilder, transport::tcp::config::Config as TcpConfig, Litep2p};
+    use multiaddr::Multiaddr;
+
+    fn create_litep2p() -> (
+        Litep2p,
+        IdentifyPublicAddresses,
+        Box<dyn Stream<Item = IdentifyEvent> + Send + Unpin>,
+    ) {
+        let (identify_config, pub_addr, identify) = Config::new(
+            "1.0.0".to_string(),
+            Some("litep2p/1.0.0".to_string()),
+            vec![Multiaddr::empty()],
+        );
+
+        let config = ConfigBuilder::new()
+            .with_tcp(TcpConfig {
+                listen_addresses: vec!["/ip6/::1/tcp/0".parse().unwrap()],
+                ..Default::default()
+            })
+            .with_libp2p_identify(identify_config)
+            .build();
+
+        (Litep2p::new(config).unwrap(), pub_addr, identify)
+    }
+
+    #[tokio::test]
+    async fn update_identify_addresses() {
+        // Create two instances of litep2p
+        let (mut litep2p1, addr1, mut event_stream1) = create_litep2p();
+        let (mut litep2p2, _addr2, mut event_stream2) = create_litep2p();
+
+        let multiaddr: Multiaddr = "/ip6/::9/tcp/111".parse().unwrap();
+        // Litep2p1 is now listening on the new address.
+        addr1.add(multiaddr.clone());
+
+        // Dial `litep2p1`
+        litep2p2
+            .dial_address(litep2p1.listen_addresses().next().unwrap().clone())
+            .await
+            .unwrap();
+
+        tokio::spawn(async move {
+            loop {
+                tokio::select! {
+                    _ = litep2p1.next_event() => {}
+                    _event = event_stream1.next() => {}
+                }
+            }
+        });
+
+        loop {
+            tokio::select! {
+                _ = litep2p2.next_event() => {}
+                event = event_stream2.next() => match event {
+                    Some(IdentifyEvent::PeerIdentified {
+                        listen_addresses,
+                        ..
+                    }) => {
+                        assert!(listen_addresses.iter().any(|address| address == &multiaddr));
+                        break;
+                    }
+                    _ => {}
+                }
+            }
+        }
+    }
+}

tests/conformance/rust/identify.rs

@@ -71,7 +71,7 @@ fn initialize_litep2p() -> (
 ) {
     let keypair = Keypair::generate();
     let (ping_config, ping_event_stream) = PingConfig::default();
-    let (identify_config, identify_event_stream) =
+    let (identify_config, _, identify_event_stream) =
         IdentifyConfig::new("proto v1".to_string(), None, Vec::new());
 
     let litep2p = Litep2p::new(

tests/protocol/identify.rs

@@ -65,7 +65,7 @@ async fn identify_supported(transport1: Transport, transport2: Transport) {
         .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
         .try_init();
 
-    let (identify_config1, mut identify_event_stream1) = Config::new(
+    let (identify_config1, _, mut identify_event_stream1) = Config::new(
         "/proto/1".to_string(),
         Some("agent v1".to_string()),
         Vec::new(),
@@ -75,7 +75,7 @@ async fn identify_supported(transport1: Transport, transport2: Transport) {
         .with_libp2p_identify(identify_config1);
     let config1 = add_transport(config_builder1, transport1).build();
 
-    let (identify_config2, mut identify_event_stream2) = Config::new(
+    let (identify_config2, _, mut identify_event_stream2) = Config::new(
         "/proto/2".to_string(),
         Some("agent v2".to_string()),
         Vec::new(),
@@ -193,7 +193,7 @@ async fn identify_not_supported(transport1: Transport, transport2: Transport) {
         .with_libp2p_ping(ping_config);
     let config1 = add_transport(config_builder1, transport1).build();
 
-    let (identify_config2, mut identify_event_stream2) =
+    let (identify_config2, _, mut identify_event_stream2) =
         Config::new("litep2p".to_string(), None, Vec::new());
     let config_builder2 = ConfigBuilder::new()
         .with_keypair(Keypair::generate())