dicedb-rs 0.1.2

Rust SDK for DiceDb.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

use std::io;

use uuid::Uuid;

use crate::{
    commands::{Command, CommandExecutor, ExecutionMode, Value, WatchValue},
    stream::{Stream, StreamError, WatchValueReceiver},
};

#[derive(Debug)]
pub enum WatchStreamError {
    IoError(io::Error),
    UnexpectedResponse(Value),
    StreamError(StreamError),
}

impl From<io::Error> for WatchStreamError {
    fn from(error: io::Error) -> Self {
        WatchStreamError::IoError(error)
    }
}

impl From<StreamError> for WatchStreamError {
    fn from(error: StreamError) -> Self {
        WatchStreamError::StreamError(error)
    }
}

pub struct WatchStream {
    host: String,
    port: u16,
    pub(crate) fingerprint: Option<String>,
    pub(crate) id: String,
    pub(crate) stream: std::net::TcpStream,
}

impl WatchStream {
    pub(crate) fn new(host: String, port: u16) -> Result<Self, WatchStreamError> {
        let stream = std::net::TcpStream::connect(format!("{}:{}", host, port))?;
        let id = Uuid::new_v4().to_string();
        let fingerprint = None;
        Ok(WatchStream {
            stream,
            id,
            fingerprint,
            host,
            port,
        })
    }
}

impl Drop for WatchStream {
    fn drop(&mut self) {
        match &self.fingerprint {
            Some(f) => _ = self.execute_command(Command::UNWATCH { key: f.to_string() }),
            None => {}
        }
    }
}

impl Iterator for WatchStream {
    type Item = WatchValue;

    fn next(&mut self) -> Option<Self::Item> {
        let value = self.recieve_watchvalue();
        match value {
            Ok(val) => Some(val),
            Err(_) => None,
        }
    }
}

impl Stream for WatchStream {
    fn host(&self) -> &str {
        self.host.as_str()
    }

    fn port(&self) -> u16 {
        self.port
    }

    fn set_stream(&mut self, stream: std::net::TcpStream) {
        self.stream = stream;
    }

    fn tcp_stream(&mut self) -> &std::net::TcpStream {
        &self.stream
    }

    fn handshake(&mut self) -> Result<(), StreamError> {
        let handshake = Command::HANDSHAKE {
            client_id: self.id.clone(),
            execution_mode: ExecutionMode::Watch,
        };
        let reply = self.execute_command(handshake)?;
        match reply {
            Value::VStr(v) if v == "OK" => Ok(()),
            value => Err(StreamError::IoError(io::Error::new(
                io::ErrorKind::Other,
                format!("Handshake error: {:?}", value),
            ))),
        }
    }
}