mountos_native/
watch.rs

1//! Change watches. A watch is a plain call on a pooled connection. Its change notices arrive
2//! on that connection and are routed to the `Watch` handle by watch id.
3
4use crate::client::{Client, Slot};
5use crate::error::{Error, Result};
6use crate::events::{Chan, Event, Hub};
7use crate::wire::*;
8use std::sync::atomic::{AtomicBool, Ordering};
9use std::sync::Arc;
10use std::time::Duration;
11
12/// Events kept per watch before the oldest are dropped.
13const WATCH_QUEUE: usize = 256;
14
15/// A live watch. Iterating blocks until the next event. Only `Change` and `Overflow` events of
16/// this watch arrive here. When more than 256 events wait, the oldest are dropped and one
17/// `Overflow` event is delivered, which tells the caller to list the path again.
18///
19/// The iterator ends after `close`, when the connection ends, or when the client closes, after
20/// the events already queued. All methods take `&self`, so another thread can call `close`
21/// while one blocks in `next` (iterate a shared `&Watch` or an `Arc<Watch>`). Dropping the
22/// handle closes it. The watch lives on one connection, and the server ends it with that
23/// connection.
24pub struct Watch {
25  id: u64,
26  slot: Arc<Slot>,
27  epoch: u64,
28  chan: Arc<Chan>,
29  hub: Arc<Hub>,
30  closed: AtomicBool,
31}
32
33impl Client {
34  /// Starts a watch on `path`. `since` asks for events after that sequence. The server keeps
35  /// no history, so an older value yields one Overflow event. Pass 0 for new events only.
36  pub fn watch(&self, path: &str, recursive: bool, since: u64) -> Result<Watch> {
37    let chan = Arc::new(Chan::new(WATCH_QUEUE, 0));
38    let req = WatchRequest {
39      path: path.into(),
40      recursive,
41      since_sequence: since,
42    };
43    let (r, slot, epoch) = self.run_any(|c| c.call_watch(&req, chan.clone()))?;
44    Ok(Watch {
45      id: r.watch_id,
46      slot: slot.clone(),
47      epoch,
48      chan,
49      hub: self.inner.hub.clone(),
50      closed: AtomicBool::new(false),
51    })
52  }
53}
54
55impl Watch {
56  /// The server's watch id, which is the `ref_id` of its events.
57  pub fn id(&self) -> u64 {
58    self.id
59  }
60
61  /// Waits up to `d` for an event. `None` means the wait timed out or the watch has ended.
62  pub fn recv_timeout(&self, d: Duration) -> Option<Event> {
63    self.chan.pop(Some(d))
64  }
65
66  /// Stops the watch: sends Unwatch on the connection that made the watch and ends a blocked
67  /// iterator. A connection that is already gone, or an EBADF reply, is not an error because
68  /// the watch ended with the connection. A second call does nothing.
69  pub fn close(&self) -> Result<()> {
70    if self.closed.swap(true, Ordering::SeqCst) {
71      return Ok(());
72    }
73    self.hub.remove_watch(self.id);
74    self.chan.close(false);
75    let req = UnwatchRequest { watch_id: self.id };
76    match self
77      .slot
78      .run(Some(self.epoch), |c| c.call_msg::<Empty>(method::UNWATCH, &req, &[]))
79    {
80      Ok(_) | Err(Error::Io(_)) | Err(Error::Timeout(_)) => Ok(()),
81      Err(Error::Server { code, .. }) if code == "EBADF" => Ok(()),
82      Err(e) => Err(e),
83    }
84  }
85}
86
87impl Iterator for Watch {
88  type Item = Event;
89
90  fn next(&mut self) -> Option<Event> {
91    self.chan.pop(None)
92  }
93}
94
95impl Iterator for &Watch {
96  type Item = Event;
97
98  fn next(&mut self) -> Option<Event> {
99    self.chan.pop(None)
100  }
101}
102
103impl Drop for Watch {
104  fn drop(&mut self) {
105    let _ = self.close();
106  }
107}
mountos_native/watch.rs

mountos_native/
watch.rs
