use std::ops::Deref; use std::time; use crossbeam_channel as chan; use crate::prelude::*; use crate::storage::RefUpdate; /// A service event. #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "kebab-case", tag = "type")] pub enum Event { RefsFetched { remote: NodeId, rid: Id, updated: Vec, }, RefsSynced { remote: NodeId, rid: Id, }, SeedDiscovered { rid: Id, nid: NodeId, }, SeedDropped { rid: Id, nid: NodeId, }, PeerConnected { nid: NodeId, }, } /// Events feed. pub struct Events(chan::Receiver); impl IntoIterator for Events { type Item = Event; type IntoIter = chan::IntoIter; fn into_iter(self) -> Self::IntoIter { self.0.into_iter() } } impl From> for Events { fn from(value: chan::Receiver) -> Self { Self(value) } } impl Deref for Events { type Target = chan::Receiver; fn deref(&self) -> &Self::Target { &self.0 } } impl Events { /// Listen for events, and wait for the given predicate to return something, /// or timeout if the specified amount of time has elapsed. pub fn wait( &self, mut f: F, timeout: time::Duration, ) -> Result where F: FnMut(&Event) -> bool, { let start = time::Instant::now(); loop { if let Some(timeout) = timeout.checked_sub(start.elapsed()) { match self.recv_timeout(timeout) { Ok(event) => { if f(&event) { return Ok(event); } } Err(err @ chan::RecvTimeoutError::Disconnected) => { return Err(err); } Err(chan::RecvTimeoutError::Timeout) => { // Keep trying until our timeout reaches zero. continue; } } } else { return Err(chan::RecvTimeoutError::Timeout); } } } }