dicedb_rs/
watchstream.rs

1//! # WatchStream Module
2//! The watchstream module contains the WatchStream struct and its implementation.
3use std::io;
4
5use uuid::Uuid;
6
7use crate::{
8    commands::{Command, CommandExecutor, ExecutionMode, ScalarValue, WatchValue},
9    errors::{StreamError, WatchStreamError},
10    stream::{Stream, WatchValueReceiver},
11};
12
13/// WatchStream is a stream that is used to watch for changes in a key.
14/// It is build from the [`Client`](crate::client::Client) using the
15/// [`get_watch`](crate::client::Client::get_watch) method.
16///
17/// The stream implements the [`Iterator`] trait
18/// and will yield [`WatchValue`] values.
19///
20/// Therefore to use the stream, you can use it in a for loop like this:
21///
22/// ```rust
23/// use dicedb_rs::client::Client;
24/// fn main() -> Result<(), dicedb_rs::errors::ClientError> {
25///     let mut client = Client::new("localhost".to_string(), 7379)?;
26///     let (watch_stream, first_value) = client.get_watch("key").unwrap();
27///     eprintln!("First value: {:?}", first_value);
28///     // watch stream is an iterator:
29///     // for value in watch_stream {
30///        // println!("Value: {:?}", value);
31///        // Do something with the value
32///        // ...
33///    // }
34/// Ok(())
35/// }
36/// ```
37#[derive(Debug)]
38pub struct WatchStream {
39    host: String,
40    port: u16,
41    pub(crate) fingerprint: Option<String>,
42    pub(crate) id: String,
43    pub(crate) stream: std::net::TcpStream,
44}
45
46impl WatchStream {
47    pub(crate) fn new(host: String, port: u16) -> Result<Self, WatchStreamError> {
48        let stream = std::net::TcpStream::connect(format!("{}:{}", host, port))?;
49        let id = Uuid::new_v4().to_string();
50        let fingerprint = None;
51        Ok(WatchStream {
52            stream,
53            id,
54            fingerprint,
55            host,
56            port,
57        })
58    }
59}
60
61impl Drop for WatchStream {
62    fn drop(&mut self) {
63        match &self.fingerprint {
64            Some(f) => _ = self.execute_scalar_command(Command::UNWATCH { key: f.to_string() }),
65            None => {}
66        }
67    }
68}
69
70impl Iterator for WatchStream {
71    type Item = WatchValue;
72
73    fn next(&mut self) -> Option<Self::Item> {
74        let value = self.recieve_watchvalue();
75        match value {
76            Ok(val) => Some(val),
77            Err(_) => None,
78        }
79    }
80}
81
82impl Stream for WatchStream {
83    fn host(&self) -> &str {
84        self.host.as_str()
85    }
86
87    fn port(&self) -> u16 {
88        self.port
89    }
90
91    fn set_stream(&mut self, stream: std::net::TcpStream) {
92        self.stream = stream;
93    }
94
95    fn tcp_stream(&mut self) -> &std::net::TcpStream {
96        &self.stream
97    }
98
99    fn handshake(&mut self) -> Result<(), StreamError> {
100        let handshake = Command::HANDSHAKE {
101            client_id: self.id.clone(),
102            execution_mode: ExecutionMode::Watch,
103        };
104        let reply = self.execute_scalar_command(handshake)?;
105        match reply {
106            ScalarValue::VStr(v) if v == "OK" => Ok(()),
107            value => Err(StreamError::IoError(io::Error::new(
108                io::ErrorKind::Other,
109                format!("Handshake error: {:?}", value),
110            ))),
111        }
112    }
113}
dicedb_rs/watchstream.rs

dicedb_rs/
watchstream.rs
