links: wait for subscriber instead of ticker

1 files changed, 16 insertions, 6 deletions

diff --git a/src/swarm/links.rs b/src/swarm/links.rs
index e3c1b56..1d5436e 100644
--- a/src/swarm/links.rs
+++ b/src/swarm/links.rs
@@ -11,9 +11,10 @@
 //!
 //! A LINKS may subscribe (`subscribe_secs`): the member then sends a LINKS_REPLY again
 //! whenever its links change (`changed`: a peer, a transport, relayed, dormant, or a
-//! round trip that moved by a fifth), looked at every `WATCH_EVERY`, until the
-//! subscription runs out (the asker renews it with its next LINKS, so one that is gone
-//! stops it by not renewing) or a LINKS with 0 ends it.
+//! round trip that moved by a fifth), looked at every `WATCH_EVERY` while anyone
+//! subscribes (round trips change without telling), until the subscription runs out
+//! (the asker renews it with its next LINKS, so one that is gone stops it by not
+//! renewing) or a LINKS with 0 ends it.
 
 use std::collections::HashMap;
 use std::sync::Mutex;
@@ -51,6 +52,8 @@ const REPLY_WITHIN: Duration = Duration::from_secs(30);
 #[derive(Default)]
 pub(crate) struct Links {
     inner: Mutex<Inner>,
+    /// Someone subscribed (`answer`): `watch` looks again.
+    subscribed: tokio::sync::Notify,
 }
 
 #[derive(Default)]
@@ -273,6 +276,7 @@ pub(crate) async fn answer(state: &AppState, peer: &str, request: LinksRequest)
             inner
                 .subscribers
                 .insert(peer.to_owned(), (request.nonce, now + lasts));
+            state.links.subscribed.notify_one();
         }
         if inner
             .answered
@@ -324,8 +328,8 @@ pub(crate) fn received(state: &AppState, peer: &str, mut reply: LinksReply) {
 }
 
 /// For subscribers (`answer`): every `WATCH_EVERY`, while there are any, this node's
-/// links; sent to each of them when they changed since they were last sent. Runs for
-/// the node's life.
+/// links; sent to each of them when they changed since they were last sent. Without
+/// any, it waits for one. Runs for the node's life.
 pub(crate) async fn watch(state: AppState) {
     let mut every = tokio::time::interval(WATCH_EVERY);
     every.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
@@ -337,7 +341,6 @@ pub(crate) async fn watch(state: AppState) {
             inner.subscribers.retain(|_, (_, until)| now < *until);
             if inner.subscribers.is_empty() {
                 inner.published = None;
-                continue;
             }
             inner
                 .subscribers
@@ -345,6 +348,13 @@ pub(crate) async fn watch(state: AppState) {
                 .map(|(peer, (nonce, _))| (peer.clone(), *nonce))
                 .collect()
         };
+        if subscribers.is_empty() {
+            // A new subscriber was just answered (`answer`, or within `ANSWER_EVERY`):
+            // the next look is due an interval later.
+            state.links.subscribed.notified().await;
+            every.reset();
+            continue;
+        }
         let links = own_links(&state).await;
         {
             let mut inner = state.links.inner.lock_or_recover();