rustis 0.19.3

Redis async driver for Rust
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179

use crate::resp::{ArgLayout, ArgSerializer};
use bytes::{Bytes, BytesMut};
use serde::{Serialize, ser::SerializeSeq};
use smallvec::SmallVec;

/// A specialized buffer for Redis command arguments.
///
/// This structure acts as a "RESP Writer". It holds the raw bytes of the arguments
/// and maintains a layout index to allow random access to arguments before the
/// command is finalized.
pub struct CommandArgsMut {
    /// The raw buffer containing the serialized arguments (in RESP format).
    pub(crate) buffer: BytesMut,
    /// An ephemeral index of argument positions (Start Offset, Length).
    ///
    /// This allows the `Client` to extract keys (for Cluster sharding) or
    /// channel names (for Pub/Sub) in O(1) time without re-parsing the buffer.
    /// This index is dropped when the command is sent to the network layer.
    pub(crate) args_layout: SmallVec<[ArgLayout; 10]>,
}

impl Default for CommandArgsMut {
    fn default() -> Self {
        Self {
            buffer: BytesMut::with_capacity(1024),
            args_layout: Default::default(),
        }
    }
}

impl CommandArgsMut {
    #[inline(always)]
    pub fn arg(mut self, arg: impl Serialize) -> Self {
        let mut serializer = ArgSerializer::new(&mut self.buffer, &mut self.args_layout);
        arg.serialize(&mut serializer)
            .expect("Arg serialization failed");
        self
    }

    /// Returns the number of arguments currently written.
    #[inline]
    pub fn len(&self) -> usize {
        self.args_layout.len()
    }

    /// Returns `true` if there is no argument
    #[inline]
    pub fn is_empty(&self) -> bool {
        self.args_layout.is_empty()
    }

    #[inline]
    pub fn freeze(self) -> CommandArgs {
        CommandArgs {
            buffer: self.buffer.freeze(),
            args_layout: self.args_layout,
        }
    }
}

impl Serialize for CommandArgsMut {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        struct RawBytes<'a>(&'a [u8]);

        impl<'a> Serialize for RawBytes<'a> {
            fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
                serializer.serialize_bytes(self.0)
            }
        }

        let mut seq = serializer.serialize_seq(Some(self.len()))?;

        for arg_layout in &self.args_layout {
            let arg_bytes = &self.buffer[arg_layout.range()];
            seq.serialize_element(&RawBytes(arg_bytes))?;
        }

        seq.end()
    }
}

/// A specialized buffer for Redis command arguments.
///
/// This structure acts as a "RESP Writer". It holds the raw bytes of the arguments
/// and maintains a layout index to allow random access to arguments before the
/// command is finalized.
#[derive(Default)]
pub struct CommandArgs {
    /// The raw buffer containing the serialized arguments (in RESP format).
    pub(crate) buffer: Bytes,
    /// An ephemeral index of argument positions (Start Offset, Length).
    ///
    /// This allows the `Client` to extract keys (for Cluster sharding) or
    /// channel names (for Pub/Sub) in O(1) time without re-parsing the buffer.
    /// This index is dropped when the command is sent to the network layer.
    pub(crate) args_layout: SmallVec<[ArgLayout; 10]>,
}

impl CommandArgs {
    /// Returns the number of arguments currently written.
    #[inline]
    pub fn len(&self) -> usize {
        self.args_layout.len()
    }

    /// Returns `true` if there is no argument
    #[inline]
    pub fn is_empty(&self) -> bool {
        self.args_layout.is_empty()
    }

    pub fn iter(&self) -> CommandArgsIterator<'_> {
        self.into_iter()
    }
}

impl Serialize for CommandArgs {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        let mut seq = serializer.serialize_seq(Some(self.len()))?;

        for arg in self {
            seq.serialize_element(&arg)?;
        }

        seq.end()
    }
}

impl<'a> IntoIterator for &'a CommandArgs {
    type Item = Bytes;
    type IntoIter = CommandArgsIterator<'a>;

    #[inline]
    fn into_iter(self) -> Self::IntoIter {
        CommandArgsIterator {
            buffer: self.buffer.clone(),
            layout_iter: self.args_layout.iter(),
        }
    }
}

/// [`CommandArgs`] iterator
pub struct CommandArgsIterator<'a> {
    pub(crate) buffer: Bytes,
    pub(crate) layout_iter: std::slice::Iter<'a, ArgLayout>,
}

impl<'a> Iterator for CommandArgsIterator<'a> {
    type Item = Bytes;

    #[inline]
    fn next(&mut self) -> Option<Self::Item> {
        let arg_layout = self.layout_iter.next()?;
        Some(self.buffer.slice(arg_layout.range()))
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        self.layout_iter.size_hint()
    }
}

impl<'a> DoubleEndedIterator for CommandArgsIterator<'a> {
    fn next_back(&mut self) -> Option<Self::Item> {
        let arg_layout = self.layout_iter.next_back()?;
        Some(self.buffer.slice(arg_layout.range()))
    }
}

impl<'a> ExactSizeIterator for CommandArgsIterator<'a> {
    fn len(&self) -> usize {
        self.layout_iter.len()
    }
}