dicedb_rs/
watchrpc.rs

1use crate::{
2    client::Client,
3    commands::{Command, CommandExecutor, ScalarValue},
4    errors::ClientError,
5    stream::Stream,
6    watchstream::WatchStream,
7};
8
9type Result<T> = std::result::Result<T, ClientError>;
10
11impl Client {
12    /// Get a watch stream for a key.
13    /// >[!WARNING]
14    /// > This operation is non deterministic, but will at best effort yield changes.
15    /// # Arguments
16    /// * `key` - The key to watch
17    /// # Returns
18    /// * A watch stream and the first value of the key
19    /// # Errors
20    /// * If the watch stream could not be created
21    pub fn get_watch(&mut self, key: &str) -> Result<(WatchStream, ScalarValue)> {
22        let mut new_watch_stream = WatchStream::new(self.host.clone(), self.port)?;
23        new_watch_stream.handshake()?;
24        let get_watch = Command::GETWATCH {
25            key: key.to_string(),
26        };
27        let reply = new_watch_stream.execute_scalar_command(get_watch)?;
28        new_watch_stream.fingerprint = Some(key.to_string());
29        Ok((new_watch_stream, reply))
30    }
31}
32
33#[cfg(test)]
34mod tests {
35    use std::{
36        sync::{Arc, Mutex},
37        thread,
38    };
39
40    use super::*;
41    const HOST: &str = "localhost";
42    const PORT: u16 = 7379;
43
44    // BUG: When keys contain underscores, it seems to give inconsistent behaviors
45    #[allow(dead_code)]
46    const BUGGY_KEYS: [&str; 4] = [
47        "watch_key",
48        "watch_key_first_value",
49        "watch_key_first_val_int",
50        "watch_key_iter",
51    ];
52
53    const GOOD_KEYS: [&str; 4] = [
54        "watchkey",
55        "watchkeyfirstvalue",
56        "watchkeyfirstvalint",
57        "watchkeyiter",
58    ];
59
60    const KEYS: [&str; 4] = GOOD_KEYS;
61
62    #[test]
63    fn test_create_watcher() {
64        let mut client = Client::new(HOST.to_string(), PORT).unwrap();
65        let key = KEYS[0];
66        let watch_stream = client.get_watch(key);
67        assert!(watch_stream.is_ok());
68    }
69
70    #[test]
71    fn test_get_watch_first_value_null() {
72        let mut client = Client::new(HOST.to_string(), PORT).unwrap();
73        let key = KEYS[1];
74        let watch_stream = client.get_watch(key).unwrap();
75        let (_, first_value) = watch_stream;
76        assert_eq!(first_value, ScalarValue::VNull);
77    }
78
79    #[test]
80    fn test_get_watch_first_val_int() {
81        let mut client = Client::new(HOST.to_string(), PORT).unwrap();
82        let key = KEYS[2];
83        client.set(key, 1).unwrap();
84        let watch_stream = client.get_watch(key).unwrap();
85        let (_, first_value) = watch_stream;
86        assert_eq!(first_value, ScalarValue::VInt(1));
87    }
88
89    #[test]
90    #[ignore] // BUG: Flaky test
91    fn test_get_watch_iter() {
92        let key = KEYS[3];
93        let mut client = Client::new(HOST.to_string(), PORT).unwrap();
94        client.del(key).unwrap();
95        thread::sleep(std::time::Duration::from_secs(1));
96        let (watch_stream, _) = client.get_watch(key).unwrap();
97        thread::sleep(std::time::Duration::from_secs(1));
98        let empty_value_vec: Vec<ScalarValue> = vec![];
99        let changed = Arc::new(Mutex::new(empty_value_vec));
100        let changed_clone = changed.clone();
101        thread::spawn(move || {
102            let watch_stream = watch_stream;
103            for change in watch_stream {
104                changed.lock().unwrap().push(change.into());
105            }
106        });
107        for i in 0..=5 {
108            client.set(key, i).unwrap();
109        }
110
111        thread::sleep(std::time::Duration::from_secs(1));
112        let changed = changed_clone.lock().unwrap();
113        assert_eq!(
114            *changed,
115            vec![
116                ScalarValue::VNull, // WARN: Sometimes this is omitted, sometimes not
117                ScalarValue::VInt(0),
118                ScalarValue::VInt(1),
119                ScalarValue::VInt(2),
120                ScalarValue::VInt(3),
121                ScalarValue::VInt(4),
122                ScalarValue::VInt(5),
123            ]
124        );
125    }
126}
dicedb_rs/watchrpc.rs

dicedb_rs/
watchrpc.rs
