async_memcached/proto/
meta_protocol.rs

1use crate::{AsMemcachedValue, Client, Error, ErrorKind, Status};
2
3use crate::parser::{
4    parse_meta_arithmetic_response, parse_meta_delete_response, parse_meta_get_response,
5    parse_meta_set_response,
6};
7use crate::parser::{MetaResponse, MetaValue};
8
9use std::future::Future;
10
11use tokio::io::AsyncWriteExt;
12
13type MetaResponseParser = fn(&[u8]) -> Result<Option<(usize, MetaResponse)>, ErrorKind>;
14
15async fn drain_quiet_noop_response(
16    client: &mut Client,
17    parser: MetaResponseParser,
18) -> Result<(), Error> {
19    match client.drive_receive(parser).await? {
20        MetaResponse::Status(Status::NoOp) => Ok(()),
21        MetaResponse::Status(status) => Err(Status::Error(ErrorKind::Protocol(Some(format!(
22            "Expected quiet-mode no-op response, got status {status}"
23        ))))
24        .into()),
25        MetaResponse::Data(_) => Err(Status::Error(ErrorKind::Protocol(Some(
26            "Expected quiet-mode no-op response, got data response".to_string(),
27        )))
28        .into()),
29    }
30}
31
32/// Trait defining Meta protocol-specific methods for the Client.
33pub trait MetaProtocol {
34    /// Gets the given key with additional metadata.
35    ///
36    /// If the key is found, `Some(MetaValue)` is returned, describing the metadata and data of the key.
37    ///
38    /// Otherwise, `None` is returned.
39    //
40    // Command format:
41    // mg <key> <meta_flags>*\r\n
42    //
43    // - <key> is the key string, with a maximum length of 250 bytes.
44    //
45    // - <meta_flags> is an optional slice of string references for meta flags.
46    // Meta flags may have associated tokens after the initial character, e.g. "O123" for opaque.
47    // Using the "q" flag for quiet mode will append a no-op command to the request ("mn\r\n") so that the client
48    // can proceed properly in the event of a cache miss.
49    fn meta_get<K: AsRef<[u8]>>(
50        &mut self,
51        key: K,
52        is_quiet: bool,
53        opaque: Option<&[u8]>,
54        meta_flags: Option<&[&str]>,
55    ) -> impl Future<Output = Result<Option<MetaValue>, Error>>;
56
57    /// Sets the given key with additional metadata.
58    ///
59    /// If the value is set successfully, `Some(MetaValue)` is returned, otherwise [`Error`] is returned.
60    /// NOTE: That the data in this MetaValue is sparsely populated, containing only requested data by meta_flags
61    /// The meta set command is a generic command for storing data to memcached. Based on the flags supplied,
62    /// it can replace all storage commands (see token M) as well as adds new options.
63    //
64    // Command format:
65    // ms <key> <datalen> <meta_flags>*\r\n<data_block>\r\n
66    //
67    // - <key> is the key string, with a maximum length of 250 bytes.
68    // - <datalen> is the length of the payload data.
69    //
70    // - <meta_flags> is an optional slice of string references for meta flags.
71    // Meta flags may have associated tokens after the initial character, e.g. "O123" for opaque.
72    //
73    // - <data_block> is the payload data to be stored, with a maximum size of ~1MB.
74    fn meta_set<K, V>(
75        &mut self,
76        key: K,
77        value: V,
78        is_quiet: bool,
79        opaque: Option<&[u8]>,
80        meta_flags: Option<&[&str]>,
81    ) -> impl Future<Output = Result<Option<MetaValue>, Error>>
82    where
83        K: AsRef<[u8]>,
84        V: AsMemcachedValue;
85
86    /// Deletes the given key with additional metadata.
87    ///
88    /// If the key is found, it will be deleted, invalidated or tombstoned depending on the meta flags provided.
89    /// If data is requested back via meta flags then a `MetaValue` is returned, otherwise `None`.
90    //
91    // Command format:
92    // md <key> <meta_flags>*\r\n
93    //
94    // - <key> is the key string, with a maximum length of 250 bytes.
95    //
96    // - <meta_flags> is an optional slice of string references for meta flags.
97    // Meta flags may have associated tokens after the initial character, e.g. "O123" for opaque.
98    fn meta_delete<K: AsRef<[u8]>>(
99        &mut self,
100        key: K,
101        is_quiet: bool,
102        opaque: Option<&[u8]>,
103        meta_flags: Option<&[&str]>,
104    ) -> impl Future<Output = Result<Option<MetaValue>, Error>>;
105
106    /// Performs an increment (arithmetic) operation on the given key.
107    ///
108    /// If the key is found, the increment operation is performed.
109    /// If data is requested back via meta flags then a `MetaValue` is returned, otherwise `None`.
110    ///
111    /// Command format:
112    ///   ma <key> <meta_flags>*\r\n
113    ///
114    /// - <key> is the key string, with a maximum length of 250 bytes.
115    ///
116    /// - <opaque> is an optional slice of string references with a maximum length of 32 bytes.
117    ///
118    /// - <delta> is an optional u64 value for the decrement delta.
119    ///   The default behaviour is to decrement with a delta of 1.
120    ///
121    /// - <is_quiet> is a boolean value indicating whether to use quiet mode.
122    ///   quiet mode will append a no-op command to the request ("mn\r\n") so that the client
123    ///   can proceed properly in the event of a cache miss.
124    ///
125    /// - <meta_flags> is an optional slice of string references for additional meta flags.
126    ///   Meta flags may have associated tokens after the initial character, e.g "N123"
127    ///   Do not include "M", "D", "O" or "q" flags as additional meta flags, they will be ignored.
128    ///   Instead, use the specified parameters.
129    fn meta_increment<K: AsRef<[u8]>>(
130        &mut self,
131        key: K,
132        is_quiet: bool,
133        opaque: Option<&[u8]>,
134        delta: Option<u64>,
135        meta_flags: Option<&[&str]>,
136    ) -> impl Future<Output = Result<Option<MetaValue>, Error>>;
137
138    /// Performs a decrement (arithmetic) operation on the given key.
139    ///
140    /// If the key is found, the decrement operation is performed.
141    /// If data is requested back via meta flags then a `MetaValue` is returned, otherwise `None`.
142    ///
143    /// Command format:
144    ///   ma <key> MD <meta_flags>*\r\n
145    ///
146    /// - <key> is the key string, with a maximum length of 250 bytes.
147    ///
148    /// - <opaque> is an optional slice of string references with a maximum length of 32 bytes.
149    ///
150    /// - <delta> is an optional u64 value for the decrement delta.
151    ///   The default behaviour is to decrement with a delta of 1.
152    ///
153    /// - <is_quiet> is a boolean value indicating whether to use quiet mode.
154    ///   quiet mode will append a no-op command to the request ("mn\r\n") so that the client
155    ///   can proceed properly in the event of a cache miss.
156    ///
157    /// - <meta_flags> is an optional slice of string references for additional meta flags.
158    ///   Meta flags may have associated tokens after the initial character, e.g "N123"
159    ///   Do not include "M", "D", "O" or "q" flags as additional meta flags, they will be ignored.
160    ///   Instead, use the specified parameters.
161    fn meta_decrement<K: AsRef<[u8]>>(
162        &mut self,
163        key: K,
164        is_quiet: bool,
165        opaque: Option<&[u8]>,
166        delta: Option<u64>,
167        meta_flags: Option<&[&str]>,
168    ) -> impl Future<Output = Result<Option<MetaValue>, Error>>;
169}
170
171impl MetaProtocol for Client {
172    async fn meta_get<K: AsRef<[u8]>>(
173        &mut self,
174        key: K,
175        is_quiet: bool,
176        opaque: Option<&[u8]>,
177        meta_flags: Option<&[&str]>,
178    ) -> Result<Option<MetaValue>, Error> {
179        let kr = Self::validate_key_length(key.as_ref())?;
180
181        if let Some(opaque) = &opaque {
182            Self::validate_opaque_length(opaque)?;
183        }
184
185        self.conn.write_all(b"mg ").await?;
186        self.conn.write_all(kr).await?;
187
188        Self::check_and_write_opaque(self, opaque).await?;
189
190        Self::check_and_write_meta_flags(self, meta_flags, opaque).await?;
191
192        Self::check_and_write_quiet_mode(self, is_quiet).await?;
193
194        self.conn.flush().await?;
195
196        let response = self.drive_receive(parse_meta_get_response).await?;
197        if is_quiet && !matches!(response, MetaResponse::Status(Status::NoOp)) {
198            drain_quiet_noop_response(self, parse_meta_get_response).await?;
199        }
200
201        match response {
202            MetaResponse::Status(Status::NotFound) => Ok(None),
203            MetaResponse::Status(Status::NoOp) => Ok(None),
204            MetaResponse::Status(s) => Err(s.into()),
205            MetaResponse::Data(d) => d
206                .map(|mut items| {
207                    let item = items.remove(0);
208                    Ok(item)
209                })
210                .transpose(),
211        }
212    }
213
214    async fn meta_set<K, V>(
215        &mut self,
216        key: K,
217        value: V,
218        is_quiet: bool,
219        opaque: Option<&[u8]>,
220        meta_flags: Option<&[&str]>,
221    ) -> Result<Option<MetaValue>, Error>
222    where
223        K: AsRef<[u8]>,
224        V: AsMemcachedValue,
225    {
226        let kr = Self::validate_key_length(key.as_ref())?;
227
228        if let Some(opaque) = &opaque {
229            Self::validate_opaque_length(opaque)?;
230        }
231
232        let vr = value.as_bytes();
233
234        self.conn.write_all(b"ms ").await?;
235        self.conn.write_all(kr).await?;
236
237        let vlen = vr.len().to_string();
238        self.conn.write_all(b" ").await?;
239        self.conn.write_all(vlen.as_ref()).await?;
240
241        Self::check_and_write_opaque(self, opaque).await?;
242
243        Self::check_and_write_meta_flags(self, meta_flags, opaque).await?;
244
245        if is_quiet {
246            self.conn.write_all(b" q").await?;
247        }
248
249        self.conn.write_all(b"\r\n").await?;
250        self.conn.write_all(vr.as_ref()).await?;
251        self.conn.write_all(b"\r\n").await?;
252
253        if is_quiet {
254            self.conn.write_all(b"mn\r\n").await?;
255        }
256
257        self.conn.flush().await?;
258
259        let response = self.drive_receive(parse_meta_set_response).await?;
260        if is_quiet && !matches!(response, MetaResponse::Status(Status::NoOp)) {
261            drain_quiet_noop_response(self, parse_meta_set_response).await?;
262        }
263
264        match response {
265            MetaResponse::Status(Status::Stored) => Ok(None),
266            MetaResponse::Status(Status::NoOp) => Ok(None),
267            MetaResponse::Status(s) => Err(s.into()),
268            MetaResponse::Data(d) => d
269                .map(|mut items| {
270                    let item = items.remove(0);
271                    Ok(item)
272                })
273                .transpose(),
274        }
275    }
276
277    async fn meta_delete<K: AsRef<[u8]>>(
278        &mut self,
279        key: K,
280        is_quiet: bool,
281        opaque: Option<&[u8]>,
282        meta_flags: Option<&[&str]>,
283    ) -> Result<Option<MetaValue>, Error> {
284        let kr = Self::validate_key_length(key.as_ref())?;
285
286        if let Some(opaque) = &opaque {
287            Self::validate_opaque_length(opaque)?;
288        }
289
290        self.conn.write_all(b"md ").await?;
291        self.conn.write_all(kr).await?;
292
293        Self::check_and_write_opaque(self, opaque).await?;
294
295        Self::check_and_write_meta_flags(self, meta_flags, opaque).await?;
296
297        Self::check_and_write_quiet_mode(self, is_quiet).await?;
298
299        self.conn.flush().await?;
300
301        let response = self.drive_receive(parse_meta_delete_response).await?;
302        if is_quiet && !matches!(response, MetaResponse::Status(Status::NoOp)) {
303            drain_quiet_noop_response(self, parse_meta_delete_response).await?;
304        }
305
306        match response {
307            MetaResponse::Status(Status::Deleted) => Ok(None),
308            MetaResponse::Status(Status::Exists) => Err(Error::Protocol(Status::Exists)),
309            MetaResponse::Status(Status::NoOp) => Ok(None),
310            MetaResponse::Status(s) => Err(s.into()),
311            MetaResponse::Data(d) => d
312                .map(|mut items| {
313                    let item = items.remove(0);
314                    Ok(item)
315                })
316                .transpose(),
317        }
318    }
319
320    async fn meta_increment<K: AsRef<[u8]>>(
321        &mut self,
322        key: K,
323        is_quiet: bool,
324        opaque: Option<&[u8]>,
325        delta: Option<u64>,
326        meta_flags: Option<&[&str]>,
327    ) -> Result<Option<MetaValue>, Error> {
328        let kr = Self::validate_key_length(key.as_ref())?;
329
330        if let Some(opaque) = &opaque {
331            Self::validate_opaque_length(opaque)?;
332        }
333
334        self.conn.write_all(b"ma ").await?;
335        self.conn.write_all(kr).await?;
336
337        Self::check_and_write_opaque(self, opaque).await?;
338
339        // skip writing "MI" because it's default behaviour and we can save the bytes.
340        if let Some(delta) = delta {
341            if delta != 1 {
342                self.conn.write_all(b" D").await?;
343                self.conn.write_all(delta.to_string().as_bytes()).await?;
344            }
345        }
346
347        if let Some(meta_flags) = meta_flags {
348            for flag in meta_flags {
349                // ignore M flag because it's specific to the method called, ignore q and require param to be used
350                // prefer explicit D and O params over meta flags
351                if flag.starts_with('M')
352                    || flag.starts_with('q')
353                    || (flag.starts_with('D') && delta.is_some())
354                    || (flag.starts_with('O') && opaque.is_some())
355                {
356                    continue;
357                } else {
358                    self.conn.write_all(b" ").await?;
359                    self.conn.write_all(flag.as_bytes()).await?;
360                }
361            }
362        }
363
364        Self::check_and_write_quiet_mode(self, is_quiet).await?;
365
366        self.conn.flush().await?;
367
368        let response = self.drive_receive(parse_meta_arithmetic_response).await?;
369        if is_quiet && !matches!(response, MetaResponse::Status(Status::NoOp)) {
370            drain_quiet_noop_response(self, parse_meta_arithmetic_response).await?;
371        }
372
373        match response {
374            MetaResponse::Status(Status::Stored) => Ok(None),
375            MetaResponse::Status(Status::NoOp) => Ok(None),
376            MetaResponse::Status(s) => Err(s.into()),
377            MetaResponse::Data(d) => d
378                .map(|mut items| {
379                    let item = items.remove(0);
380                    Ok(item)
381                })
382                .transpose(),
383        }
384    }
385
386    async fn meta_decrement<K: AsRef<[u8]>>(
387        &mut self,
388        key: K,
389        is_quiet: bool,
390        opaque: Option<&[u8]>,
391        delta: Option<u64>,
392        meta_flags: Option<&[&str]>,
393    ) -> Result<Option<MetaValue>, Error> {
394        let kr = Self::validate_key_length(key.as_ref())?;
395
396        if let Some(opaque) = &opaque {
397            Self::validate_opaque_length(opaque)?;
398        }
399
400        self.conn.write_all(b"ma ").await?;
401        self.conn.write_all(kr).await?;
402        self.conn.write_all(b" MD").await?;
403
404        Self::check_and_write_opaque(self, opaque).await?;
405
406        if let Some(delta) = delta {
407            if delta != 1 {
408                self.conn.write_all(b" D").await?;
409                self.conn.write_all(delta.to_string().as_bytes()).await?;
410            }
411        }
412
413        if let Some(meta_flags) = meta_flags {
414            for flag in meta_flags {
415                // ignore M flag because it's specific to the method called, ignore q and require param to be used
416                // prefer explicit D and O params over meta flags
417                if flag.starts_with('M')
418                    || flag.starts_with('q')
419                    || (flag.starts_with('D') && delta.is_some())
420                    || (flag.starts_with('O') && opaque.is_some())
421                {
422                    continue;
423                } else {
424                    self.conn.write_all(b" ").await?;
425                    self.conn.write_all(flag.as_bytes()).await?;
426                }
427            }
428        }
429
430        Self::check_and_write_quiet_mode(self, is_quiet).await?;
431
432        self.conn.flush().await?;
433
434        let response = self.drive_receive(parse_meta_arithmetic_response).await?;
435        if is_quiet && !matches!(response, MetaResponse::Status(Status::NoOp)) {
436            drain_quiet_noop_response(self, parse_meta_arithmetic_response).await?;
437        }
438
439        match response {
440            MetaResponse::Status(Status::Stored) => Ok(None),
441            MetaResponse::Status(Status::NoOp) => Ok(None),
442            MetaResponse::Status(s) => Err(s.into()),
443            MetaResponse::Data(d) => d
444                .map(|mut items| {
445                    let item = items.remove(0);
446                    Ok(item)
447                })
448                .transpose(),
449        }
450    }
451}
async_memcached/proto/meta_protocol.rs

async_memcached/proto/
meta_protocol.rs
