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
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
use crate::resp::{ArgLayout, ArgSerializer, ArgsLayout};
use bytes::{Bytes, BytesMut};
use serde::{Serialize, ser::Error as _, ser::SerializeSeq};
/// 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: ArgsLayout,
/// First serialization error deferred from [`arg`](Self::arg), re-emitted by
/// this type's `Serialize` impl so that — when a `CommandArgsMut` is embedded
/// as an option field of a command — the failure propagates up through the
/// outer command builder rather than panicking.
pub(crate) pending_error: Option<crate::Error>,
}
impl Default for CommandArgsMut {
fn default() -> Self {
Self {
buffer: BytesMut::with_capacity(1024),
args_layout: Default::default(),
pending_error: None,
}
}
}
impl CommandArgsMut {
#[inline(always)]
pub fn arg(mut self, arg: impl Serialize) -> Self {
let result = {
let mut serializer = ArgSerializer::new(&mut self.buffer, &mut self.args_layout);
arg.serialize(&mut serializer)
};
if let Err(e) = result
&& self.pending_error.is_none()
{
self.pending_error = Some(e);
}
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)
}
}
// Propagate a deferred build error so an embedded `CommandArgsMut`
// surfaces it through the outer command builder.
if let Some(error) = &self.pending_error {
return Err(S::Error::custom(error.to_string()));
}
let mut seq = serializer.serialize_seq(Some(self.len()))?;
// An `ArgLayout` range is recorded as the argument's bytes are appended
// to this same buffer, so `get` always yields. Asking for it rather than
// indexing turns a builder bug into the error this `Serialize` can
// already return, instead of a panic in the caller's thread.
for arg_layout in &self.args_layout {
let arg_bytes = self
.buffer
.get(arg_layout.range())
.ok_or_else(|| S::Error::custom("argument layout is out of its buffer"))?;
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: ArgsLayout,
}
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()
}
}