diff options
| -rw-r--r-- | src/swarm/links.rs | 22 |
1 files changed, 16 insertions, 6 deletions
diff --git a/src/swarm/links.rs b/src/swarm/links.rs index 8502b39..b84f57a 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)] @@ -260,6 +263,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 @@ -311,8 +315,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); @@ -324,7 +328,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 @@ -332,6 +335,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(); |