feat: Make Endpoint::node_addr watchable and add trait Watcher & combinators

iroh/bench/src/iroh.rs

@@ -7,6 +7,7 @@ use anyhow::{Context, Result};
 use bytes::Bytes;
 use iroh::{
     endpoint::{Connection, ConnectionError, RecvStream, SendStream, TransportConfig},
+    watcher::Watcher as _,
     Endpoint, NodeAddr, RelayMap, RelayMode, RelayUrl,
 };
 use tracing::{trace, warn};

iroh/examples/connect-unreliable.rs

@@ -8,7 +8,7 @@
 use std::net::SocketAddr;
 
 use clap::Parser;
-use iroh::{Endpoint, NodeAddr, RelayMode, RelayUrl, SecretKey};
+use iroh::{watcher::Watcher as _, Endpoint, NodeAddr, RelayMode, RelayUrl, SecretKey};
 use tracing::info;
 
 // An example ALPN that we are using to communicate over the `Endpoint`
@@ -50,7 +50,7 @@ async fn main() -> anyhow::Result<()> {
         .bind()
         .await?;
 
-    let node_addr = endpoint.node_addr().await?;
+    let node_addr = endpoint.node_addr().initialized().await?;
     let me = node_addr.node_id;
     println!("node id: {me}");
     println!("node listening addresses:");

iroh/examples/connect.rs

@@ -9,7 +9,7 @@ use std::net::SocketAddr;
 
 use anyhow::Context;
 use clap::Parser;
-use iroh::{Endpoint, NodeAddr, RelayMode, RelayUrl, SecretKey};
+use iroh::{watcher::Watcher as _, Endpoint, NodeAddr, RelayMode, RelayUrl, SecretKey};
 use tracing::info;
 
 // An example ALPN that we are using to communicate over the `Endpoint`

iroh/examples/echo.rs

@@ -11,6 +11,7 @@ use futures_lite::future::Boxed as BoxedFuture;
 use iroh::{
     endpoint::Connecting,
     protocol::{ProtocolHandler, Router},
+    watcher::Watcher as _,
     Endpoint, NodeAddr,
 };
 
@@ -23,7 +24,7 @@ const ALPN: &[u8] = b"iroh-example/echo/0";
 #[tokio::main]
 async fn main() -> Result<()> {
     let router = accept_side().await?;
-    let node_addr = router.endpoint().node_addr().await?;
+    let node_addr = router.endpoint().node_addr().initialized().await?;
 
     connect_side(node_addr).await?;
 

iroh/examples/listen-unreliable.rs

@@ -3,7 +3,7 @@
 //! This example uses the default relay servers to attempt to holepunch, and will use that relay server to relay packets if the two devices cannot establish a direct UDP connection.
 //! run this example from the project root:
 //!     $ cargo run --example listen-unreliable
-use iroh::{Endpoint, RelayMode, SecretKey};
+use iroh::{watcher::Watcher as _, Endpoint, RelayMode, SecretKey};
 use tracing::{info, warn};
 
 // An example ALPN that we are using to communicate over the `Endpoint`
@@ -35,7 +35,7 @@ async fn main() -> anyhow::Result<()> {
     println!("node id: {me}");
     println!("node listening addresses:");
 
-    let node_addr = endpoint.node_addr().await?;
+    let node_addr = endpoint.node_addr().initialized().await?;
     let local_addrs = node_addr
         .direct_addresses
         .into_iter()

iroh/examples/listen.rs

@@ -5,7 +5,7 @@
 //!     $ cargo run --example listen
 use std::time::Duration;
 
-use iroh::{endpoint::ConnectionError, Endpoint, RelayMode, SecretKey};
+use iroh::{endpoint::ConnectionError, watcher::Watcher as _, Endpoint, RelayMode, SecretKey};
 use tracing::{debug, info, warn};
 
 // An example ALPN that we are using to communicate over the `Endpoint`
@@ -37,7 +37,7 @@ async fn main() -> anyhow::Result<()> {
     println!("node id: {me}");
     println!("node listening addresses:");
 
-    let node_addr = endpoint.node_addr().await?;
+    let node_addr = endpoint.node_addr().initialized().await?;
     let local_addrs = node_addr
         .direct_addresses
         .into_iter()

iroh/examples/transfer.rs

@@ -8,7 +8,8 @@ use bytes::Bytes;
 use clap::{Parser, Subcommand};
 use indicatif::HumanBytes;
 use iroh::{
-    endpoint::ConnectionError, Endpoint, NodeAddr, RelayMap, RelayMode, RelayUrl, SecretKey,
+    endpoint::ConnectionError, watcher::Watcher as _, Endpoint, NodeAddr, RelayMap, RelayMode,
+    RelayUrl, SecretKey,
 };
 use iroh_base::ticket::NodeTicket;
 use tracing::info;

iroh/src/discovery.rs

@@ -450,7 +450,7 @@ mod tests {
     use tokio_util::task::AbortOnDropHandle;
 
     use super::*;
-    use crate::RelayMode;
+    use crate::{watcher::Watcher as _, RelayMode};
 
     type InfoStore = HashMap<NodeId, (Option<RelayUrl>, BTreeSet<SocketAddr>, u64)>;
 
@@ -580,7 +580,7 @@ mod tests {
         };
         let ep1_addr = NodeAddr::new(ep1.node_id());
         // wait for our address to be updated and thus published at least once
-        ep1.node_addr().await?;
+        ep1.node_addr().initialized().await?;
         let _conn = ep2.connect(ep1_addr, TEST_ALPN).await?;
         Ok(())
     }
@@ -606,7 +606,7 @@ mod tests {
         };
         let ep1_addr = NodeAddr::new(ep1.node_id());
         // wait for out address to be updated and thus published at least once
-        ep1.node_addr().await?;
+        ep1.node_addr().initialized().await?;
         let _conn = ep2.connect(ep1_addr, TEST_ALPN).await?;
         Ok(())
     }
@@ -636,7 +636,7 @@ mod tests {
         };
         let ep1_addr = NodeAddr::new(ep1.node_id());
         // wait for out address to be updated and thus published at least once
-        ep1.node_addr().await?;
+        ep1.node_addr().initialized().await?;
         let _conn = ep2.connect(ep1_addr, TEST_ALPN).await?;
         Ok(())
     }
@@ -659,7 +659,7 @@ mod tests {
         };
         let ep1_addr = NodeAddr::new(ep1.node_id());
         // wait for out address to be updated and thus published at least once
-        ep1.node_addr().await?;
+        ep1.node_addr().initialized().await?;
         let res = ep2.connect(ep1_addr, TEST_ALPN).await;
         assert!(res.is_err());
         Ok(())
@@ -682,7 +682,7 @@ mod tests {
             new_endpoint(secret, disco).await
         };
         // wait for out address to be updated and thus published at least once
-        ep1.node_addr().await?;
+        ep1.node_addr().initialized().await?;
         let ep1_wrong_addr = NodeAddr {
             node_id: ep1.node_id(),
             relay_url: None,

iroh/src/discovery/local_swarm_discovery.rs

@@ -54,7 +54,7 @@ use tracing::{debug, error, info_span, trace, warn, Instrument};
 
 use crate::{
     discovery::{Discovery, DiscoveryItem},
-    watchable::Watchable,
+    watcher::{Watchable, Watcher as _},
     Endpoint,
 };
 

iroh/src/discovery/pkarr.rs

@@ -61,7 +61,7 @@ use crate::{
     discovery::{Discovery, DiscoveryItem},
     dns::node_info::NodeInfo,
     endpoint::force_staging_infra,
-    watchable::{Disconnected, Watchable, Watcher},
+    watcher::{self, Disconnected, Watchable, Watcher as _},
     Endpoint,
 };
 
@@ -221,7 +221,7 @@ struct PublisherService {
     secret_key: SecretKey,
     #[debug("PkarrClient")]
     pkarr_client: PkarrRelayClient,
-    watcher: Watcher<Option<NodeInfo>>,
+    watcher: watcher::Direct<Option<NodeInfo>>,
     ttl: u32,
     republish_interval: Duration,
 }