mtop_client/
core.rs

1use std::borrow::Borrow;
2use std::cmp::Ordering;
3use std::collections::{BTreeSet, HashMap};
4use std::error;
5use std::fmt;
6use std::io;
7use std::str::FromStr;
8use std::time::Duration;
9use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader, BufWriter};
10
11#[derive(Debug, Default, PartialEq, Clone)]
12pub struct Stats {
13    // Server info
14    pub pid: i64,
15    pub uptime: u64,
16    pub server_time: i64,
17    pub threads: u64,
18    pub version: String,
19
20    // CPU
21    pub rusage_user: f64,
22    pub rusage_system: f64,
23
24    // Connections
25    pub max_connections: u64,
26    pub curr_connections: u64,
27    pub total_connections: u64,
28    pub rejected_connections: u64,
29
30    // Commands
31    pub cmd_get: u64,
32    pub cmd_set: u64,
33    pub cmd_flush: u64,
34    pub cmd_touch: u64,
35    pub cmd_meta: u64,
36
37    // Gets
38    pub get_hits: u64,
39    pub get_misses: u64,
40    pub get_expired: u64,
41    pub get_flushed: u64,
42
43    // Sets
44    pub store_too_large: u64,
45    pub store_no_memory: u64,
46
47    // Deletes
48    pub delete_hits: u64,
49    pub delete_misses: u64,
50
51    // Incr/Decr
52    pub incr_hits: u64,
53    pub incr_misses: u64,
54    pub decr_hits: u64,
55    pub decr_misses: u64,
56
57    // Touches
58    pub touch_hits: u64,
59    pub touch_misses: u64,
60
61    // Bytes
62    pub bytes_read: u64,
63    pub bytes_written: u64,
64    pub bytes: u64,
65    pub max_bytes: u64,
66
67    // Items
68    pub curr_items: u64,
69    pub total_items: u64,
70    pub evictions: u64,
71}
72
73impl TryFrom<&HashMap<String, String>> for Stats {
74    type Error = MtopError;
75
76    fn try_from(value: &HashMap<String, String>) -> Result<Self, Self::Error> {
77        Ok(Stats {
78            pid: parse_field("pid", value)?,
79            uptime: parse_field("uptime", value)?,
80            server_time: parse_field("time", value)?,
81            version: parse_field("version", value)?,
82            threads: parse_field("threads", value)?,
83
84            rusage_user: parse_field("rusage_user", value)?,
85            rusage_system: parse_field("rusage_system", value)?,
86
87            max_connections: parse_field("max_connections", value)?,
88            curr_connections: parse_field("curr_connections", value)?,
89            total_connections: parse_field("total_connections", value)?,
90            rejected_connections: parse_field("rejected_connections", value)?,
91
92            cmd_get: parse_field("cmd_get", value)?,
93            cmd_set: parse_field("cmd_set", value)?,
94            cmd_flush: parse_field("cmd_flush", value)?,
95            cmd_touch: parse_field("cmd_touch", value)?,
96            cmd_meta: parse_field("cmd_meta", value)?,
97
98            get_hits: parse_field("get_hits", value)?,
99            get_misses: parse_field("get_misses", value)?,
100            get_expired: parse_field("get_expired", value)?,
101            get_flushed: parse_field("get_flushed", value)?,
102
103            store_too_large: parse_field("store_too_large", value)?,
104            store_no_memory: parse_field("store_no_memory", value)?,
105
106            delete_hits: parse_field("delete_hits", value)?,
107            delete_misses: parse_field("delete_misses", value)?,
108
109            incr_hits: parse_field("incr_hits", value)?,
110            incr_misses: parse_field("incr_misses", value)?,
111
112            decr_hits: parse_field("decr_hits", value)?,
113            decr_misses: parse_field("decr_misses", value)?,
114
115            touch_hits: parse_field("touch_hits", value)?,
116            touch_misses: parse_field("touch_misses", value)?,
117
118            bytes_read: parse_field("bytes_read", value)?,
119            bytes_written: parse_field("bytes_written", value)?,
120            bytes: parse_field("bytes", value)?,
121            max_bytes: parse_field("limit_maxbytes", value)?,
122
123            curr_items: parse_field("curr_items", value)?,
124            total_items: parse_field("total_items", value)?,
125            evictions: parse_field("evictions", value)?,
126        })
127    }
128}
129
130#[derive(Debug, Default, PartialEq, Eq, Clone, Hash)]
131pub struct Slab {
132    pub id: u64,
133    pub chunk_size: u64,
134    pub chunks_per_page: u64,
135    pub total_pages: u64,
136    pub total_chunks: u64,
137    pub used_chunks: u64,
138    pub free_chunks: u64,
139    pub get_hits: u64,
140    pub cmd_set: u64,
141    pub delete_hits: u64,
142    pub incr_hits: u64,
143    pub decr_hits: u64,
144    pub cas_hits: u64,
145    pub cas_badval: u64,
146    pub touch_hits: u64,
147}
148
149impl PartialOrd for Slab {
150    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
151        Some(self.cmp(other))
152    }
153}
154
155impl Ord for Slab {
156    fn cmp(&self, other: &Self) -> Ordering {
157        self.id.cmp(&other.id)
158    }
159}
160
161#[derive(Debug, Default, PartialEq, Eq, Clone)]
162#[repr(transparent)]
163pub struct Slabs(Vec<Slab>);
164
165impl Slabs {
166    pub fn iter(&self) -> impl ExactSizeIterator<Item = &Slab> {
167        self.0.iter()
168    }
169
170    pub fn len(&self) -> usize {
171        self.0.len()
172    }
173
174    pub fn is_empty(&self) -> bool {
175        self.0.is_empty()
176    }
177}
178
179impl IntoIterator for Slabs {
180    type Item = Slab;
181    type IntoIter = std::vec::IntoIter<Slab>;
182
183    fn into_iter(self) -> Self::IntoIter {
184        self.0.into_iter()
185    }
186}
187
188impl TryFrom<&HashMap<String, String>> for Slabs {
189    type Error = MtopError;
190
191    fn try_from(value: &HashMap<String, String>) -> Result<Self, Self::Error> {
192        // Parse the slab IDs from each of the raw stats. We have to do this because
193        // Memcached isn't guaranteed to use a particular slab ID if there are no items
194        // to store in that size class. Otherwise, we could just loop from one to
195        // $active_slabs + 1.
196        let mut ids = BTreeSet::new();
197        for k in value.keys() {
198            let key_id: Option<u64> = k.split_once(':').map(|(raw, _rest)| raw).and_then(|raw| raw.parse().ok());
199
200            if let Some(id) = key_id {
201                ids.insert(id);
202            }
203        }
204
205        let mut slabs = Vec::with_capacity(ids.len());
206
207        for id in ids {
208            slabs.push(Slab {
209                id,
210                chunk_size: parse_field(&format!("{}:chunk_size", id), value)?,
211                chunks_per_page: parse_field(&format!("{}:chunks_per_page", id), value)?,
212                total_pages: parse_field(&format!("{}:total_pages", id), value)?,
213                total_chunks: parse_field(&format!("{}:total_chunks", id), value)?,
214                used_chunks: parse_field(&format!("{}:used_chunks", id), value)?,
215                free_chunks: parse_field(&format!("{}:free_chunks", id), value)?,
216                get_hits: parse_field(&format!("{}:get_hits", id), value)?,
217                cmd_set: parse_field(&format!("{}:cmd_set", id), value)?,
218                delete_hits: parse_field(&format!("{}:delete_hits", id), value)?,
219                incr_hits: parse_field(&format!("{}:incr_hits", id), value)?,
220                decr_hits: parse_field(&format!("{}:decr_hits", id), value)?,
221                cas_hits: parse_field(&format!("{}:cas_hits", id), value)?,
222                cas_badval: parse_field(&format!("{}:cas_badval", id), value)?,
223                touch_hits: parse_field(&format!("{}:touch_hits", id), value)?,
224            });
225        }
226
227        Ok(Self(slabs))
228    }
229}
230
231#[derive(Debug, Default, PartialEq, Eq, Clone, Hash)]
232pub struct SlabItem {
233    pub id: u64,
234    pub number: u64,
235    pub number_hot: u64,
236    pub number_warm: u64,
237    pub number_cold: u64,
238    pub age_hot: u64,
239    pub age_warm: u64,
240    pub age: u64,
241    pub mem_requested: u64,
242    pub evicted: u64,
243    pub evicted_nonzero: u64,
244    pub evicted_time: u64,
245    pub out_of_memory: u64,
246    pub tail_repairs: u64,
247    pub reclaimed: u64,
248    pub expired_unfetched: u64,
249    pub evicted_unfetched: u64,
250    pub evicted_active: u64,
251    pub crawler_reclaimed: u64,
252    pub crawler_items_checked: u64,
253    pub lrutail_reflocked: u64,
254    pub moves_to_cold: u64,
255    pub moves_to_warm: u64,
256    pub moves_within_lru: u64,
257    pub direct_reclaims: u64,
258    pub hits_to_hot: u64,
259    pub hits_to_warm: u64,
260    pub hits_to_cold: u64,
261    pub hits_to_temp: u64,
262}
263
264impl PartialOrd for SlabItem {
265    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
266        Some(self.cmp(other))
267    }
268}
269
270impl Ord for SlabItem {
271    fn cmp(&self, other: &Self) -> Ordering {
272        self.id.cmp(&other.id)
273    }
274}
275
276#[derive(Debug, Default, PartialEq, Eq, Clone, Hash)]
277#[repr(transparent)]
278pub struct SlabItems(Vec<SlabItem>);
279
280impl SlabItems {
281    pub fn iter(&self) -> impl ExactSizeIterator<Item = &SlabItem> {
282        self.0.iter()
283    }
284
285    pub fn len(&self) -> usize {
286        self.0.len()
287    }
288
289    pub fn is_empty(&self) -> bool {
290        self.0.is_empty()
291    }
292}
293
294impl IntoIterator for SlabItems {
295    type Item = SlabItem;
296    type IntoIter = std::vec::IntoIter<SlabItem>;
297
298    fn into_iter(self) -> Self::IntoIter {
299        self.0.into_iter()
300    }
301}
302
303impl TryFrom<&HashMap<String, String>> for SlabItems {
304    type Error = MtopError;
305
306    fn try_from(value: &HashMap<String, String>) -> Result<Self, Self::Error> {
307        // Parse the slab IDs from each of the raw stats. We have to do this because
308        // Memcached isn't guaranteed to use a particular slab ID if there are no items
309        // to store in that size class. Otherwise, we could just loop from one to
310        // $active_slabs + 1.
311        let mut ids = BTreeSet::new();
312        for k in value.keys() {
313            let key_id: Option<u64> = k
314                .trim_start_matches("items:")
315                .split_once(':')
316                .map(|(raw, _rest)| raw)
317                .and_then(|raw| raw.parse().ok());
318
319            if let Some(id) = key_id {
320                ids.insert(id);
321            }
322        }
323
324        let mut items = Vec::with_capacity(ids.len());
325
326        for id in ids {
327            items.push(SlabItem {
328                id,
329                number: parse_field(&format!("items:{}:number", id), value)?,
330                number_hot: parse_field(&format!("items:{}:number_hot", id), value)?,
331                number_warm: parse_field(&format!("items:{}:number_warm", id), value)?,
332                number_cold: parse_field(&format!("items:{}:number_cold", id), value)?,
333                age_hot: parse_field(&format!("items:{}:age_hot", id), value)?,
334                age_warm: parse_field(&format!("items:{}:age_warm", id), value)?,
335                age: parse_field(&format!("items:{}:age", id), value)?,
336                mem_requested: parse_field(&format!("items:{}:mem_requested", id), value)?,
337                evicted: parse_field(&format!("items:{}:evicted", id), value)?,
338                evicted_nonzero: parse_field(&format!("items:{}:evicted_nonzero", id), value)?,
339                evicted_time: parse_field(&format!("items:{}:evicted_time", id), value)?,
340                out_of_memory: parse_field(&format!("items:{}:outofmemory", id), value)?,
341                tail_repairs: parse_field(&format!("items:{}:tailrepairs", id), value)?,
342                reclaimed: parse_field(&format!("items:{}:reclaimed", id), value)?,
343                expired_unfetched: parse_field(&format!("items:{}:expired_unfetched", id), value)?,
344                evicted_unfetched: parse_field(&format!("items:{}:evicted_unfetched", id), value)?,
345                evicted_active: parse_field(&format!("items:{}:evicted_active", id), value)?,
346                crawler_reclaimed: parse_field(&format!("items:{}:crawler_reclaimed", id), value)?,
347                crawler_items_checked: parse_field(&format!("items:{}:crawler_items_checked", id), value)?,
348                lrutail_reflocked: parse_field(&format!("items:{}:lrutail_reflocked", id), value)?,
349                moves_to_cold: parse_field(&format!("items:{}:moves_to_cold", id), value)?,
350                moves_to_warm: parse_field(&format!("items:{}:moves_to_warm", id), value)?,
351                moves_within_lru: parse_field(&format!("items:{}:moves_within_lru", id), value)?,
352                direct_reclaims: parse_field(&format!("items:{}:direct_reclaims", id), value)?,
353                hits_to_hot: parse_field(&format!("items:{}:hits_to_hot", id), value)?,
354                hits_to_warm: parse_field(&format!("items:{}:hits_to_warm", id), value)?,
355                hits_to_cold: parse_field(&format!("items:{}:hits_to_cold", id), value)?,
356                hits_to_temp: parse_field(&format!("items:{}:hits_to_temp", id), value)?,
357            });
358        }
359
360        Ok(Self(items))
361    }
362}
363
364#[derive(Debug, Default, PartialEq, Eq, Clone, Hash)]
365pub struct Meta {
366    pub key: String,
367    pub expires: i64,
368    pub size: u64,
369}
370
371impl Meta {
372    // Meta information is returned from the server as multiple key-value pairs per
373    // line. We only care about a subset of those keys. Define them here to avoid doing
374    // extra work when parsing the server response.
375    const KEYS: &'static [&'static str] = &["key", "exp", "size"];
376}
377
378impl TryFrom<&HashMap<String, String>> for Meta {
379    type Error = MtopError;
380
381    fn try_from(value: &HashMap<String, String>) -> Result<Self, Self::Error> {
382        Ok(Meta {
383            key: parse_field("key", value)?,
384            expires: parse_field("exp", value)?,
385            size: parse_field("size", value)?,
386        })
387    }
388}
389
390impl PartialOrd for Meta {
391    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
392        Some(self.cmp(other))
393    }
394}
395
396impl Ord for Meta {
397    fn cmp(&self, other: &Self) -> Ordering {
398        self.key.cmp(&other.key)
399    }
400}
401
402#[derive(Debug, Default, PartialEq, Eq, Clone, Hash)]
403pub struct Value {
404    pub key: String,
405    pub cas: u64,
406    pub flags: u64,
407    pub data: Vec<u8>,
408}
409
410impl PartialOrd for Value {
411    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
412        Some(self.cmp(other))
413    }
414}
415
416impl Ord for Value {
417    fn cmp(&self, other: &Self) -> Ordering {
418        self.key.cmp(&other.key)
419    }
420}
421
422fn parse_field<T>(key: &str, map: &HashMap<String, String>) -> Result<T, MtopError>
423where
424    T: FromStr,
425    <T as FromStr>::Err: fmt::Display + Send + Sync + error::Error + 'static,
426{
427    map.get(key)
428        .ok_or_else(|| MtopError::runtime(format!("field {} missing", key)))
429        .and_then(|v| {
430            v.parse()
431                .map_err(|e| MtopError::runtime_cause(format!("field {} value '{}'", key, v), e))
432        })
433}
434
435fn parse_value<T>(val: &str, line: &str) -> Result<T, MtopError>
436where
437    T: FromStr + fmt::Display,
438    <T as FromStr>::Err: fmt::Display + Send + Sync + error::Error + 'static,
439{
440    val.parse()
441        .map_err(|e| MtopError::runtime_cause(format!("parsing {} from '{}'", val, line), e))
442}
443
444#[derive(Debug, PartialOrd, PartialEq, Copy, Clone)]
445pub enum ErrorKind {
446    Runtime,
447    IO,
448    Protocol,
449    Configuration,
450}
451
452impl fmt::Display for ErrorKind {
453    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
454        match self {
455            Self::Runtime => write!(f, "runtime error"),
456            Self::IO => write!(f, "io error"),
457            Self::Protocol => write!(f, "protocol error"),
458            Self::Configuration => write!(f, "configuration error"),
459        }
460    }
461}
462
463#[derive(Debug)]
464enum ErrorRepr {
465    Message(String),
466    Cause(Box<dyn error::Error + Send + Sync + 'static>),
467    MessageCause(String, Box<dyn error::Error + Send + Sync + 'static>),
468}
469
470#[derive(Debug)]
471pub struct MtopError {
472    kind: ErrorKind,
473    repr: ErrorRepr,
474}
475
476impl MtopError {
477    pub fn runtime<S>(msg: S) -> MtopError
478    where
479        S: Into<String>,
480    {
481        MtopError {
482            kind: ErrorKind::Runtime,
483            repr: ErrorRepr::Message(msg.into()),
484        }
485    }
486
487    pub fn runtime_cause<S, E>(msg: S, e: E) -> MtopError
488    where
489        S: Into<String>,
490        E: error::Error + Send + Sync + 'static,
491    {
492        MtopError {
493            kind: ErrorKind::Runtime,
494            repr: ErrorRepr::MessageCause(msg.into(), Box::new(e)),
495        }
496    }
497
498    pub fn configuration<S>(msg: S) -> MtopError
499    where
500        S: Into<String>,
501    {
502        MtopError {
503            kind: ErrorKind::Configuration,
504            repr: ErrorRepr::Message(msg.into()),
505        }
506    }
507
508    pub fn configuration_cause<S, E>(msg: S, e: E) -> MtopError
509    where
510        S: Into<String>,
511        E: error::Error + Send + Sync + 'static,
512    {
513        MtopError {
514            kind: ErrorKind::Configuration,
515            repr: ErrorRepr::MessageCause(msg.into(), Box::new(e)),
516        }
517    }
518
519    pub fn timeout<D>(t: Duration, operation: D) -> MtopError
520    where
521        D: fmt::Display,
522    {
523        MtopError {
524            kind: ErrorKind::IO,
525            repr: ErrorRepr::Message(format!("operation {} timed out after {:?}", operation, t)),
526        }
527    }
528
529    pub fn kind(&self) -> ErrorKind {
530        self.kind
531    }
532}
533
534impl fmt::Display for MtopError {
535    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
536        match &self.repr {
537            ErrorRepr::Message(msg) => write!(f, "{}: {}", self.kind, msg),
538            ErrorRepr::Cause(e) => write!(f, "{}: {}", self.kind, e),
539            ErrorRepr::MessageCause(msg, e) => write!(f, "{}: {}: {}", self.kind, msg, e),
540        }
541    }
542}
543
544impl error::Error for MtopError {
545    fn source(&self) -> Option<&(dyn error::Error + 'static)> {
546        match &self.repr {
547            ErrorRepr::Message(_) => None,
548            ErrorRepr::Cause(e) | ErrorRepr::MessageCause(_, e) => Some(e.as_ref()),
549        }
550    }
551}
552
553impl From<(String, io::Error)> for MtopError {
554    fn from((s, e): (String, io::Error)) -> Self {
555        MtopError {
556            kind: ErrorKind::IO,
557            repr: ErrorRepr::MessageCause(s, Box::new(e)),
558        }
559    }
560}
561
562impl From<io::Error> for MtopError {
563    fn from(e: io::Error) -> Self {
564        MtopError {
565            kind: ErrorKind::IO,
566            repr: ErrorRepr::Cause(Box::new(e)),
567        }
568    }
569}
570
571impl From<ProtocolError> for MtopError {
572    fn from(e: ProtocolError) -> Self {
573        MtopError {
574            kind: ErrorKind::Protocol,
575            repr: ErrorRepr::Cause(Box::new(e)),
576        }
577    }
578}
579
580#[derive(Debug, PartialEq, Eq, Copy, Clone, Hash)]
581pub enum ProtocolErrorKind {
582    BadClass,
583    Busy,
584    Client,
585    NotFound,
586    NotStored,
587    Server,
588    Syntax,
589}
590
591impl fmt::Display for ProtocolErrorKind {
592    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
593        match self {
594            Self::BadClass => "BADCLASS".fmt(f),
595            Self::Busy => "BUSY".fmt(f),
596            Self::Client => "CLIENT_ERROR".fmt(f),
597            Self::NotFound => "NOT_FOUND".fmt(f),
598            Self::NotStored => "NOT_STORED".fmt(f),
599            Self::Server => "SERVER_ERROR".fmt(f),
600            Self::Syntax => "ERROR".fmt(f),
601        }
602    }
603}
604
605#[derive(Debug)]
606pub struct ProtocolError {
607    kind: ProtocolErrorKind,
608    message: Option<String>,
609}
610
611impl ProtocolError {
612    pub fn kind(&self) -> ProtocolErrorKind {
613        self.kind
614    }
615}
616
617impl fmt::Display for ProtocolError {
618    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
619        if let Some(msg) = &self.message {
620            write!(f, "{} {}", self.kind, msg)
621        } else {
622            write!(f, "{}", self.kind)
623        }
624    }
625}
626
627impl error::Error for ProtocolError {}
628
629#[derive(Debug, Eq, PartialEq, Clone)]
630enum Command<'a> {
631    Add(&'a Key, u64, u32, &'a [u8]),
632    CrawlerMetadump,
633    Decr(&'a Key, u64),
634    Delete(&'a Key),
635    FlushAll(u64),
636    Gets(&'a [Key]),
637    Incr(&'a Key, u64),
638    MetaNoop,
639    Replace(&'a Key, u64, u32, &'a [u8]),
640    Stats,
641    StatsItems,
642    StatsSlabs,
643    Set(&'a Key, u64, u32, &'a [u8]),
644    Touch(&'a Key, u32),
645}
646
647impl<'a> From<Command<'a>> for Vec<u8> {
648    fn from(value: Command<'a>) -> Self {
649        match value {
650            Command::Add(key, flags, ttl, data) => storage_command("add", key, flags, ttl, data),
651            Command::CrawlerMetadump => "lru_crawler metadump hash\r\n".to_owned().into_bytes(),
652            Command::Decr(key, delta) => format!("decr {} {}\r\n", key, delta).into_bytes(),
653            Command::Delete(key) => format!("delete {}\r\n", key).into_bytes(),
654            Command::FlushAll(wait) => format!("flush_all {}\r\n", wait).into_bytes(),
655            Command::Gets(keys) => format!("gets {}\r\n", keys.join(" ")).into_bytes(),
656            Command::Incr(key, delta) => format!("incr {} {}\r\n", key, delta).into_bytes(),
657            Command::MetaNoop => "mn\r\n".to_owned().into_bytes(),
658            Command::Replace(key, flags, ttl, data) => storage_command("replace", key, flags, ttl, data),
659            Command::Stats => "stats\r\n".to_owned().into_bytes(),
660            Command::StatsItems => "stats items\r\n".to_owned().into_bytes(),
661            Command::StatsSlabs => "stats slabs\r\n".to_owned().into_bytes(),
662            Command::Set(key, flags, ttl, data) => storage_command("set", key, flags, ttl, data),
663            Command::Touch(key, ttl) => format!("touch {} {}\r\n", key, ttl).into_bytes(),
664        }
665    }
666}
667
668fn storage_command(verb: &str, key: &Key, flags: u64, ttl: u32, data: &[u8]) -> Vec<u8> {
669    let mut bytes = Vec::with_capacity(key.len() + data.len() + 32);
670    io::Write::write_all(
671        &mut bytes,
672        format!("{} {} {} {} {}\r\n", verb, key, flags, ttl, data.len()).as_bytes(),
673    )
674    .unwrap();
675    io::Write::write_all(&mut bytes, data).unwrap();
676    io::Write::write_all(&mut bytes, "\r\n".as_bytes()).unwrap();
677    bytes
678}
679
680pub struct Memcached {
681    read: BufReader<Box<dyn AsyncRead + Send + Sync + Unpin>>,
682    write: BufWriter<Box<dyn AsyncWrite + Send + Sync + Unpin>>,
683
684    buf_pairs: HashMap<String, String>,
685    buf_line: String,
686}
687
688impl Memcached {
689    const MAX_PAYLOAD_SIZE: u64 = 1024 * 1024 * 1024;
690
691    pub fn new<R, W>(read: R, write: W) -> Self
692    where
693        R: AsyncRead + Send + Sync + Unpin + 'static,
694        W: AsyncWrite + Send + Sync + Unpin + 'static,
695    {
696        Memcached {
697            read: BufReader::new(Box::new(read)),
698            write: BufWriter::new(Box::new(write)),
699            buf_pairs: HashMap::new(),
700            buf_line: String::new(),
701        }
702    }
703
704    /// Get a `Stats` object with the current values of the interesting stats for the server.
705    pub async fn stats(&mut self) -> Result<Stats, MtopError> {
706        self.send(Command::Stats).await?;
707        self.read_stats_response().await
708    }
709
710    /// Get a `Slabs` object with information about each set of `Slab`s maintained by
711    /// the Memcached server. You can think of each `Slab` as a class of objects that
712    /// are stored together in memory. Note that `Slab` IDs may not be contiguous based
713    /// on the size of items actually stored by the server.
714    pub async fn slabs(&mut self) -> Result<Slabs, MtopError> {
715        self.send(Command::StatsSlabs).await?;
716        self.read_stats_response().await
717    }
718
719    /// Get a `SlabsItems` object with information about the `SlabItem` items stored in
720    /// each slab class maintained by the Memcached server. The ID of each `SlabItem`
721    /// corresponds to a `Slab` maintained by the server. Note that `SlabItem` IDs may
722    /// not be contiguous based on the size of items actually stored by the server.
723    pub async fn items(&mut self) -> Result<SlabItems, MtopError> {
724        self.send(Command::StatsItems).await?;
725        self.read_stats_response().await
726    }
727
728    /// Get a `Meta` object for every item in the cache which includes its key and expiration
729    /// time as a UNIX timestamp. Expiration time will be `-1` if the item was set with an
730    /// infinite TTL.
731    pub async fn metas(&mut self) -> Result<Vec<Meta>, MtopError> {
732        self.send(Command::CrawlerMetadump).await?;
733
734        let mut out = Vec::new();
735        self.buf_line.clear();
736        self.buf_pairs.clear();
737
738        while self.read.read_line(&mut self.buf_line).await? != 0 {
739            let line = self.buf_line.trim_end();
740            if line == "END" {
741                break;
742            }
743
744            // Check for an error first because the `metadump` command doesn't
745            // have any sort of prefix for each result line like `STAT` or `VALUE`
746            // so it's hard to know if it's valid without looking for an error.
747            if let Some(err) = Self::parse_error(line) {
748                return Err(MtopError::from(err));
749            }
750
751            let item = Self::parse_crawler_meta(line, Meta::KEYS, &mut self.buf_pairs)?;
752            out.push(item);
753
754            self.buf_line.clear();
755            self.buf_pairs.clear();
756        }
757
758        Ok(out)
759    }
760
761    /// Send a simple command to verify our connection to the server is working.
762    pub async fn ping(&mut self) -> Result<(), MtopError> {
763        self.send(Command::MetaNoop).await?;
764        self.read_simple_response("MN").await
765    }
766
767    /// Flush all entries in the cache, optionally after a delay. When a delay is used, the
768    /// server will flush entries after a delay but the call will still return immediately.
769    pub async fn flush_all(&mut self, wait: Duration) -> Result<(), MtopError> {
770        self.send(Command::FlushAll(wait.as_secs())).await?;
771        self.read_simple_response("OK").await
772    }
773
774    /// Get a map of the requested keys and their corresponding `Value` in the cache
775    /// including the key, flags, and data.
776    pub async fn get(&mut self, keys: &[Key]) -> Result<HashMap<String, Value>, MtopError> {
777        self.send(Command::Gets(keys)).await?;
778
779        let mut out = HashMap::with_capacity(keys.len());
780        self.buf_line.clear();
781
782        while self.read.read_line(&mut self.buf_line).await? != 0 {
783            let line = self.buf_line.trim_end();
784            if line == "END" {
785                break;
786            }
787
788            let value = Self::parse_gets_value(line, &mut self.read).await?;
789            out.insert(value.key.clone(), value);
790            self.buf_line.clear();
791        }
792
793        Ok(out)
794    }
795
796    /// Increment the value of a key by the given delta if the value is numeric returning
797    /// the new value. Returns an error if the value is _not_ numeric.
798    pub async fn incr(&mut self, key: &Key, delta: u64) -> Result<u64, MtopError> {
799        self.send(Command::Incr(key, delta)).await?;
800        self.buf_line.clear();
801
802        if self.read.read_line(&mut self.buf_line).await? != 0 {
803            Self::parse_numeric_response(self.buf_line.trim_end())
804        } else {
805            Err(MtopError::runtime("unexpected empty response"))
806        }
807    }
808
809    /// Decrement the value of a key by the given delta if the value is numeric returning
810    /// the new value with a minimum of 0. Returns an error if the value is _not_ numeric.
811    pub async fn decr(&mut self, key: &Key, delta: u64) -> Result<u64, MtopError> {
812        self.send(Command::Decr(key, delta)).await?;
813        self.buf_line.clear();
814
815        if self.read.read_line(&mut self.buf_line).await? != 0 {
816            Self::parse_numeric_response(self.buf_line.trim_end())
817        } else {
818            Err(MtopError::runtime("unexpected empty response"))
819        }
820    }
821
822    /// Store the provided item in the cache, regardless of whether it already exists.
823    pub async fn set<V>(&mut self, key: &Key, flags: u64, ttl: u32, data: V) -> Result<(), MtopError>
824    where
825        V: AsRef<[u8]>,
826    {
827        self.send(Command::Set(key, flags, ttl, data.as_ref())).await?;
828        self.read_simple_response("STORED").await
829    }
830
831    /// Store the provided item in the cache only if it does not already exist.
832    pub async fn add<V>(&mut self, key: &Key, flags: u64, ttl: u32, data: V) -> Result<(), MtopError>
833    where
834        V: AsRef<[u8]>,
835    {
836        self.send(Command::Add(key, flags, ttl, data.as_ref())).await?;
837        self.read_simple_response("STORED").await
838    }
839
840    /// Store the provided item in the cache only if it already exists.
841    pub async fn replace<V>(&mut self, key: &Key, flags: u64, ttl: u32, data: V) -> Result<(), MtopError>
842    where
843        V: AsRef<[u8]>,
844    {
845        self.send(Command::Replace(key, flags, ttl, data.as_ref())).await?;
846        self.read_simple_response("STORED").await
847    }
848
849    /// Update the TTL of an item in the cache if it exists, return an error otherwise.
850    pub async fn touch(&mut self, key: &Key, ttl: u32) -> Result<(), MtopError> {
851        self.send(Command::Touch(key, ttl)).await?;
852        self.read_simple_response("TOUCHED").await
853    }
854
855    /// Delete an item in the cache if it exists, return an error otherwise.
856    pub async fn delete(&mut self, key: &Key) -> Result<(), MtopError> {
857        self.send(Command::Delete(key)).await?;
858        self.read_simple_response("DELETED").await
859    }
860
861    async fn read_stats_response<'this, 's, S>(&'this mut self) -> Result<S, MtopError>
862    where
863        'this: 's,
864        S: TryFrom<&'s HashMap<String, String>, Error = MtopError>,
865    {
866        self.buf_pairs.clear();
867        self.buf_line.clear();
868
869        while self.read.read_line(&mut self.buf_line).await? != 0 {
870            let line = self.buf_line.trim_end();
871            if line == "END" {
872                break;
873            }
874
875            let (key, val) = Self::parse_stat_line(line)?;
876            self.buf_pairs.insert(key.to_owned(), val.to_owned());
877            self.buf_line.clear();
878        }
879
880        S::try_from(&self.buf_pairs)
881    }
882
883    fn parse_stat_line(line: &str) -> Result<(&str, &str), MtopError> {
884        let mut parts = line.splitn(3, ' ');
885        match (parts.next(), parts.next(), parts.next()) {
886            (Some("STAT"), Some(key), Some(val)) => Ok((key, val)),
887            _ => {
888                if let Some(err) = Self::parse_error(line) {
889                    Err(MtopError::from(err))
890                } else {
891                    Err(MtopError::runtime(format!("unable to parse '{}'", line)))
892                }
893            }
894        }
895    }
896
897    #[inline]
898    fn parse_crawler_meta(line: &str, keys: &[&str], raw: &mut HashMap<String, String>) -> Result<Meta, MtopError> {
899        assert!(raw.is_empty(), "scratch HashMap should be cleared before being used");
900
901        for p in line.split(' ') {
902            let (key, val) = p
903                .split_once('=')
904                .ok_or_else(|| MtopError::runtime(format!("unexpected metadump format '{}'", line)))?;
905
906            // Avoid spending time decoding values or allocating for data we don't care about.
907            // Use a slice here since it's faster than a HashSet when the number of entries is
908            // small and the number of keys we're searching through is always small.
909            if !keys.contains(&key) {
910                continue;
911            }
912
913            let decoded = crate::codec::url_decode(val)?;
914            raw.insert(key.to_owned(), decoded);
915        }
916
917        Meta::try_from(&*raw)
918    }
919
920    async fn parse_gets_value<R>(line: &str, reader: &mut BufReader<R>) -> Result<Value, MtopError>
921    where
922        R: AsyncRead + Send + Sync + Unpin + 'static,
923    {
924        let mut parts = line.splitn(5, ' ');
925
926        match (parts.next(), parts.next(), parts.next(), parts.next(), parts.next()) {
927            (Some("VALUE"), Some(k), Some(flags), Some(len), Some(cas)) => {
928                let flags: u64 = parse_value(flags, line)?;
929                let len: u64 = parse_value(len, line)?;
930                let cas: u64 = parse_value(cas, line)?;
931
932                // The max size of an object in Memcached is represented with a `u64` which
933                // means it's basically infinite. In practice the default max size of an object
934                // in a Memcached server is 1MB but can be configured higher. Place a limit on
935                // the size that we'll accept here to avoid a denial of service from bad lengths.
936                if len > Self::MAX_PAYLOAD_SIZE {
937                    return Err(MtopError::runtime(format!(
938                        "server response of length {} exceeds client max of {}",
939                        len,
940                        Self::MAX_PAYLOAD_SIZE
941                    )));
942                }
943
944                let data = {
945                    // Two extra bytes to read the trailing \r\n but then truncate them.
946                    let mut data = Vec::with_capacity(usize::try_from(len).unwrap() + 2);
947                    reader.take(len + 2).read_to_end(&mut data).await?;
948                    data.truncate(usize::try_from(len).unwrap());
949                    data
950                };
951
952                Ok(Value {
953                    key: k.to_owned(),
954                    flags,
955                    cas,
956                    data,
957                })
958            }
959            _ => {
960                // Response doesn't look like a `VALUE` line, see if the server has
961                // responded with an error that we can parse. Otherwise, consider this
962                // an internal error.
963                if let Some(err) = Self::parse_error(line) {
964                    Err(MtopError::from(err))
965                } else {
966                    Err(MtopError::runtime(format!("unable to parse '{}'", line)))
967                }
968            }
969        }
970    }
971
972    fn parse_numeric_response(line: &str) -> Result<u64, MtopError> {
973        if let Some(err) = Self::parse_error(line) {
974            Err(MtopError::from(err))
975        } else {
976            line.parse()
977                .map_err(|_e| MtopError::runtime(format!("unable to parse '{}'", line)))
978        }
979    }
980
981    async fn read_simple_response(&mut self, expected: &str) -> Result<(), MtopError> {
982        self.buf_line.clear();
983
984        if self.read.read_line(&mut self.buf_line).await? != 0 {
985            let line = self.buf_line.trim_end();
986            if line == expected {
987                Ok(())
988            } else if let Some(err) = Self::parse_error(line) {
989                Err(MtopError::from(err))
990            } else {
991                Err(MtopError::runtime(format!("unable to parse '{}'", line)))
992            }
993        } else {
994            Err(MtopError::runtime("unexpected empty response"))
995        }
996    }
997
998    fn parse_error(line: &str) -> Option<ProtocolError> {
999        let mut values = line.splitn(2, ' ');
1000        let (kind, message) = match (values.next(), values.next()) {
1001            (Some("BADCLASS"), Some(msg)) => (ProtocolErrorKind::BadClass, Some(msg.to_owned())),
1002            (Some("BUSY"), Some(msg)) => (ProtocolErrorKind::Busy, Some(msg.to_owned())),
1003            (Some("CLIENT_ERROR"), Some(msg)) => (ProtocolErrorKind::Client, Some(msg.to_owned())),
1004            (Some("ERROR"), None) => (ProtocolErrorKind::Syntax, None),
1005            (Some("ERROR"), Some(msg)) => (ProtocolErrorKind::Syntax, Some(msg.to_owned())),
1006            (Some("NOT_FOUND"), None) => (ProtocolErrorKind::NotFound, None),
1007            (Some("NOT_STORED"), None) => (ProtocolErrorKind::NotStored, None),
1008            (Some("SERVER_ERROR"), Some(msg)) => (ProtocolErrorKind::Server, Some(msg.to_owned())),
1009
1010            _ => return None,
1011        };
1012
1013        Some(ProtocolError { kind, message })
1014    }
1015
1016    async fn send(&mut self, cmd: Command<'_>) -> Result<(), MtopError> {
1017        let cmd_bytes: Vec<u8> = cmd.into();
1018        self.write.write_all(&cmd_bytes).await?;
1019        Ok(self.write.flush().await?)
1020    }
1021}
1022
1023impl fmt::Debug for Memcached {
1024    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1025        f.debug_struct("Memcached")
1026            .field("read", &"...")
1027            .field("write", &"...")
1028            .finish()
1029    }
1030}
1031
1032#[derive(Debug, Clone, Eq, PartialEq, Hash, Ord, PartialOrd)]
1033#[repr(transparent)]
1034pub struct Key(String);
1035
1036impl Key {
1037    const MAX_LENGTH: usize = 250;
1038
1039    pub fn parse_all<I, T>(vals: I) -> Result<Vec<Self>, MtopError>
1040    where
1041        I: IntoIterator<Item = T>,
1042        T: TryInto<Key, Error = MtopError>,
1043    {
1044        let iter = vals.into_iter();
1045        let (sz, _) = iter.size_hint();
1046        let mut out = Vec::with_capacity(sz);
1047
1048        for val in iter {
1049            out.push(val.try_into()?);
1050        }
1051
1052        Ok(out)
1053    }
1054
1055    pub fn len(&self) -> usize {
1056        self.0.len()
1057    }
1058
1059    pub fn is_empty(&self) -> bool {
1060        self.0.is_empty()
1061    }
1062
1063    fn parse<T>(val: T) -> Result<Self, MtopError>
1064    where
1065        T: Into<String>,
1066    {
1067        let val = val.into();
1068        if Self::is_legal_val(&val) {
1069            Ok(Key(val))
1070        } else {
1071            Err(MtopError::runtime(format!("invalid key {}", val)))
1072        }
1073    }
1074
1075    fn is_legal_val(val: &str) -> bool {
1076        if val.len() > Self::MAX_LENGTH {
1077            return false;
1078        }
1079
1080        for c in val.chars() {
1081            if !c.is_ascii() || c.is_ascii_whitespace() || c.is_ascii_control() {
1082                return false;
1083            }
1084        }
1085
1086        true
1087    }
1088}
1089
1090impl fmt::Display for Key {
1091    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1092        self.0.fmt(f)
1093    }
1094}
1095
1096impl AsRef<str> for Key {
1097    fn as_ref(&self) -> &str {
1098        &self.0
1099    }
1100}
1101
1102impl Borrow<str> for Key {
1103    fn borrow(&self) -> &str {
1104        &self.0
1105    }
1106}
1107
1108impl TryFrom<String> for Key {
1109    type Error = MtopError;
1110
1111    fn try_from(value: String) -> Result<Self, Self::Error> {
1112        Key::parse(value)
1113    }
1114}
1115
1116impl TryFrom<&String> for Key {
1117    type Error = MtopError;
1118
1119    fn try_from(value: &String) -> Result<Self, Self::Error> {
1120        Key::parse(value)
1121    }
1122}
1123
1124impl TryFrom<&str> for Key {
1125    type Error = MtopError;
1126
1127    fn try_from(value: &str) -> Result<Self, Self::Error> {
1128        Key::parse(value)
1129    }
1130}
1131
1132#[cfg(test)]
1133mod test {
1134    use super::{ErrorKind, Key, Memcached, Meta, Slab, SlabItem, SlabItems};
1135    use std::io::{Cursor, Error};
1136    use std::pin::Pin;
1137    use std::task::{Context, Poll};
1138    use std::time::Duration;
1139    use tokio::io::AsyncWrite;
1140    use tokio::sync::mpsc::{self, UnboundedSender};
1141
1142    /////////
1143    // key //
1144    /////////
1145
1146    #[test]
1147    fn test_key_parse_length() {
1148        let val = "abc".repeat(Key::MAX_LENGTH);
1149        let res = Key::parse(val);
1150        assert!(res.is_err());
1151    }
1152
1153    #[test]
1154    fn test_key_parse_non_ascii() {
1155        let val = "🤦";
1156        let res = Key::parse(val);
1157        assert!(res.is_err());
1158    }
1159
1160    #[test]
1161    fn test_key_parse_whitespace() {
1162        let val = "some thing";
1163        let res = Key::parse(val);
1164        assert!(res.is_err());
1165    }
1166
1167    #[test]
1168    fn test_key_parse_control_char() {
1169        let val = "\x7F";
1170        let res = Key::parse(val);
1171        assert!(res.is_err());
1172    }
1173
1174    #[test]
1175    fn test_key_parse_success() {
1176        let val = "a-reasonable-key";
1177        let key = Key::parse(val).unwrap();
1178
1179        assert_eq!(Key("a-reasonable-key".to_owned()), key);
1180    }
1181
1182    #[test]
1183    fn test_key_parse_all_string() {
1184        let vals = vec![String::from("foo"), String::from("bar")];
1185        let keys = Key::parse_all(vals).unwrap();
1186
1187        assert_eq!(vec![Key("foo".to_owned()), Key("bar".to_owned())], keys,);
1188    }
1189
1190    #[test]
1191    fn test_key_parse_all_string_borrowed() {
1192        let vals = vec![String::from("foo"), String::from("bar")];
1193        let keys = Key::parse_all(&vals).unwrap();
1194
1195        assert_eq!(vec![Key("foo".to_owned()), Key("bar".to_owned())], keys,);
1196    }
1197
1198    #[test]
1199    fn test_key_parse_all_str() {
1200        let vals = ["foo", "bar"];
1201        let keys = Key::parse_all(vals).unwrap();
1202
1203        assert_eq!(vec![Key("foo".to_owned()), Key("bar".to_owned())], keys,);
1204    }
1205
1206    struct WriteAdapter {
1207        tx: UnboundedSender<Vec<u8>>,
1208    }
1209
1210    impl AsyncWrite for WriteAdapter {
1211        fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll<Result<usize, Error>> {
1212            self.tx.send(buf.to_owned()).unwrap();
1213            Poll::Ready(Ok(buf.len()))
1214        }
1215
1216        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
1217            Poll::Ready(Ok(()))
1218        }
1219
1220        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
1221            Poll::Ready(Ok(()))
1222        }
1223    }
1224
1225    /// Create a new receiver channel and `Memcached` instance to read the provided server
1226    /// response. Anything written by the client is able to be read from the receiver channel.
1227    /// NOTE that it is important that the receiver not be dropped by the caller since this will
1228    /// cause writes to the channel to fail from within the client.
1229    macro_rules! client {
1230        () => ({
1231            let (tx, rx) = mpsc::unbounded_channel();
1232            let reads = Vec::new();
1233            (rx, Memcached::new(Cursor::new(reads), WriteAdapter { tx }))
1234        });
1235        ($($line:expr),+ $(,)?) => ({
1236            let (tx, rx) = mpsc::unbounded_channel();
1237            let mut reads = Vec::new();
1238            $(reads.extend_from_slice($line.as_bytes());)+
1239            (rx, Memcached::new(Cursor::new(reads), WriteAdapter { tx }))
1240        })
1241    }
1242
1243    /////////
1244    // get //
1245    /////////
1246
1247    #[tokio::test]
1248    async fn test_memcached_get_no_key() {
1249        let (_rx, mut client) = client!();
1250        let vals: Vec<String> = vec![];
1251        let keys = Key::parse_all(vals).unwrap();
1252        let res = client.get(&keys).await.unwrap();
1253
1254        assert!(res.is_empty());
1255    }
1256
1257    #[tokio::test]
1258    async fn test_memcached_get_error() {
1259        let (_rx, mut client) = client!("SERVER_ERROR backend failure\r\n");
1260        let keys = Key::parse_all(vec!["foo", "baz"]).unwrap();
1261        let res = client.get(&keys).await;
1262
1263        assert!(res.is_err());
1264        let err = res.unwrap_err();
1265        assert_eq!(ErrorKind::Protocol, err.kind());
1266    }
1267
1268    #[tokio::test]
1269    async fn test_memcached_get_miss() {
1270        let (_rx, mut client) = client!("END\r\n");
1271        let keys = Key::parse_all(vec!["foo", "baz"]).unwrap();
1272        let res = client.get(&keys).await.unwrap();
1273
1274        assert!(res.is_empty());
1275    }
1276
1277    #[tokio::test]
1278    async fn test_memcached_get_hit() {
1279        let (_rx, mut client) = client!(
1280            "VALUE foo 32 3 1\r\n",
1281            "bar\r\n",
1282            "VALUE baz 64 3 2\r\n",
1283            "qux\r\n",
1284            "END\r\n",
1285        );
1286        let keys = Key::parse_all(vec!["foo", "baz"]).unwrap();
1287        let res = client.get(&keys).await.unwrap();
1288
1289        let val1 = res.get("foo").unwrap();
1290        assert_eq!("foo", val1.key);
1291        assert_eq!("bar".as_bytes(), val1.data);
1292        assert_eq!(32, val1.flags);
1293        assert_eq!(1, val1.cas);
1294
1295        let val2 = res.get("baz").unwrap();
1296        assert_eq!("baz", val2.key);
1297        assert_eq!("qux".as_bytes(), val2.data);
1298        assert_eq!(64, val2.flags);
1299        assert_eq!(2, val2.cas);
1300    }
1301
1302    //////////
1303    // incr //
1304    //////////
1305
1306    #[tokio::test]
1307    async fn test_memcached_incr_bad_val() {
1308        let (mut rx, mut client) = client!("CLIENT_ERROR cannot increment or decrement non-numeric value\r\n");
1309        let key = Key::parse("test").unwrap();
1310        let res = client.incr(&key, 2).await;
1311
1312        assert!(res.is_err());
1313        let err = res.unwrap_err();
1314        assert_eq!(ErrorKind::Protocol, err.kind());
1315
1316        let bytes = rx.recv().await.unwrap();
1317        let command = String::from_utf8(bytes).unwrap();
1318        assert_eq!("incr test 2\r\n", command);
1319    }
1320
1321    #[tokio::test]
1322    async fn test_memcached_incr_success() {
1323        let (mut rx, mut client) = client!("3\r\n");
1324        let key = Key::parse("test").unwrap();
1325        let res = client.incr(&key, 2).await.unwrap();
1326
1327        assert_eq!(3, res);
1328        let bytes = rx.recv().await.unwrap();
1329        let command = String::from_utf8(bytes).unwrap();
1330        assert_eq!("incr test 2\r\n", command);
1331    }
1332
1333    //////////
1334    // decr //
1335    //////////
1336
1337    #[tokio::test]
1338    async fn test_memcached_decr_bad_val() {
1339        let (mut rx, mut client) = client!("CLIENT_ERROR cannot increment or decrement non-numeric value\r\n");
1340        let key = Key::parse("test").unwrap();
1341        let res = client.decr(&key, 1).await;
1342
1343        assert!(res.is_err());
1344        let err = res.unwrap_err();
1345        assert_eq!(ErrorKind::Protocol, err.kind());
1346
1347        let bytes = rx.recv().await.unwrap();
1348        let command = String::from_utf8(bytes).unwrap();
1349        assert_eq!("decr test 1\r\n", command);
1350    }
1351
1352    #[tokio::test]
1353    async fn test_memcached_decr_success() {
1354        let (mut rx, mut client) = client!("3\r\n");
1355        let key = Key::parse("test").unwrap();
1356        let res = client.decr(&key, 1).await.unwrap();
1357
1358        assert_eq!(3, res);
1359        let bytes = rx.recv().await.unwrap();
1360        let command = String::from_utf8(bytes).unwrap();
1361        assert_eq!("decr test 1\r\n", command);
1362    }
1363
1364    //////////
1365    // ping //
1366    //////////
1367
1368    #[tokio::test]
1369    async fn test_memcached_ping_bad_val() {
1370        let (mut rx, mut client) = client!("220 localhost ESMTP Postfix\r\n");
1371        let res = client.ping().await;
1372
1373        assert!(res.is_err());
1374        let err = res.unwrap_err();
1375        assert_eq!(ErrorKind::Runtime, err.kind());
1376
1377        let bytes = rx.recv().await.unwrap();
1378        let command = String::from_utf8(bytes).unwrap();
1379        assert_eq!("mn\r\n", command);
1380    }
1381
1382    #[tokio::test]
1383    async fn test_memcached_ping_success() {
1384        let (mut rx, mut client) = client!("MN\r\n");
1385        client.ping().await.unwrap();
1386
1387        let bytes = rx.recv().await.unwrap();
1388        let command = String::from_utf8(bytes).unwrap();
1389        assert_eq!("mn\r\n", command);
1390    }
1391
1392    macro_rules! test_store_command_success {
1393        ($method:ident, $verb:expr) => {
1394            let (mut rx, mut client) = client!("STORED\r\n");
1395            let res = client.$method(&Key::parse("test").unwrap(), 0, 300, "val".as_bytes()).await;
1396
1397            assert!(res.is_ok());
1398            let bytes = rx.recv().await.unwrap();
1399            let command = String::from_utf8(bytes).unwrap();
1400            assert_eq!(concat!($verb, " test 0 300 3\r\nval\r\n"), command);
1401        };
1402    }
1403
1404    macro_rules! test_store_command_error {
1405        ($method:ident, $verb:expr) => {
1406            let (mut rx, mut client) = client!("NOT_STORED\r\n");
1407            let res = client.$method(&Key::parse("test").unwrap(), 0, 300, "val".as_bytes()).await;
1408
1409            assert!(res.is_err());
1410            let err = res.unwrap_err();
1411            assert_eq!(ErrorKind::Protocol, err.kind());
1412
1413            let bytes = rx.recv().await.unwrap();
1414            let command = String::from_utf8(bytes).unwrap();
1415            assert_eq!(concat!($verb, " test 0 300 3\r\nval\r\n"), command);
1416        };
1417    }
1418
1419    /////////
1420    // set //
1421    /////////
1422
1423    #[tokio::test]
1424    async fn test_memcached_set_success() {
1425        test_store_command_success!(set, "set");
1426    }
1427
1428    #[tokio::test]
1429    async fn test_memcached_set_error() {
1430        test_store_command_error!(set, "set");
1431    }
1432
1433    /////////
1434    // add //
1435    /////////
1436
1437    #[tokio::test]
1438    async fn test_memcached_add_success() {
1439        test_store_command_success!(add, "add");
1440    }
1441
1442    #[tokio::test]
1443    async fn test_memcached_add_error() {
1444        test_store_command_error!(add, "add");
1445    }
1446
1447    /////////////
1448    // replace //
1449    /////////////
1450
1451    #[tokio::test]
1452    async fn test_memcached_replace_success() {
1453        test_store_command_success!(replace, "replace");
1454    }
1455
1456    #[tokio::test]
1457    async fn test_memcached_replace_error() {
1458        test_store_command_error!(replace, "replace");
1459    }
1460
1461    ///////////
1462    // touch //
1463    ///////////
1464
1465    #[tokio::test]
1466    async fn test_memcached_touch_success() {
1467        let (mut rx, mut client) = client!("TOUCHED\r\n");
1468        let key = Key::parse("test").unwrap();
1469        let res = client.touch(&key, 300).await;
1470
1471        assert!(res.is_ok());
1472        let bytes = rx.recv().await.unwrap();
1473        let command = String::from_utf8(bytes).unwrap();
1474        assert_eq!("touch test 300\r\n", command);
1475    }
1476
1477    #[tokio::test]
1478    async fn test_memcached_touch_error() {
1479        let (mut rx, mut client) = client!("NOT_FOUND\r\n");
1480        let key = Key::parse("test").unwrap();
1481        let res = client.touch(&key, 300).await;
1482
1483        assert!(res.is_err());
1484        let err = res.unwrap_err();
1485        assert_eq!(ErrorKind::Protocol, err.kind());
1486
1487        let bytes = rx.recv().await.unwrap();
1488        let command = String::from_utf8(bytes).unwrap();
1489        assert_eq!("touch test 300\r\n", command);
1490    }
1491
1492    ////////////
1493    // delete //
1494    ////////////
1495
1496    #[tokio::test]
1497    async fn test_memcached_delete_success() {
1498        let (mut rx, mut client) = client!("DELETED\r\n");
1499        let key = Key::parse("test").unwrap();
1500        let res = client.delete(&key).await;
1501
1502        assert!(res.is_ok());
1503        let bytes = rx.recv().await.unwrap();
1504        let command = String::from_utf8(bytes).unwrap();
1505        assert_eq!("delete test\r\n", command);
1506    }
1507
1508    #[tokio::test]
1509    async fn test_memcached_delete_error() {
1510        let (mut rx, mut client) = client!("NOT_FOUND\r\n");
1511        let key = Key::parse("test").unwrap();
1512        let res = client.delete(&key).await;
1513
1514        assert!(res.is_err());
1515        let err = res.unwrap_err();
1516        assert_eq!(ErrorKind::Protocol, err.kind());
1517
1518        let bytes = rx.recv().await.unwrap();
1519        let command = String::from_utf8(bytes).unwrap();
1520        assert_eq!("delete test\r\n", command);
1521    }
1522
1523    ///////////////
1524    // flush_all //
1525    ///////////////
1526
1527    #[tokio::test]
1528    async fn test_memcached_flush_all_with_wait_success() {
1529        let (mut rx, mut client) = client!("OK\r\n");
1530        let res = client.flush_all(Duration::from_secs(25)).await;
1531
1532        assert!(res.is_ok());
1533        let bytes = rx.recv().await.unwrap();
1534        let command = String::from_utf8(bytes).unwrap();
1535        assert_eq!("flush_all 25\r\n", command);
1536    }
1537
1538    #[tokio::test]
1539    async fn test_memcached_flush_all_with_wait_error() {
1540        let (mut rx, mut client) = client!("ERROR\r\n");
1541        let res = client.flush_all(Duration::from_secs(25)).await;
1542
1543        assert!(res.is_err());
1544        let err = res.unwrap_err();
1545        assert_eq!(ErrorKind::Protocol, err.kind());
1546
1547        let bytes = rx.recv().await.unwrap();
1548        let command = String::from_utf8(bytes).unwrap();
1549        assert_eq!("flush_all 25\r\n", command);
1550    }
1551
1552    #[tokio::test]
1553    async fn test_memcached_flush_all_no_wait_success() {
1554        let (mut rx, mut client) = client!("OK\r\n");
1555        let res = client.flush_all(Duration::ZERO).await;
1556
1557        assert!(res.is_ok());
1558        let bytes = rx.recv().await.unwrap();
1559        let command = String::from_utf8(bytes).unwrap();
1560        assert_eq!("flush_all 0\r\n", command);
1561    }
1562
1563    #[tokio::test]
1564    async fn test_memcached_flush_all_no_wait_error() {
1565        let (mut rx, mut client) = client!("ERROR\r\n");
1566        let res = client.flush_all(Duration::ZERO).await;
1567
1568        assert!(res.is_err());
1569        let err = res.unwrap_err();
1570        assert_eq!(ErrorKind::Protocol, err.kind());
1571
1572        let bytes = rx.recv().await.unwrap();
1573        let command = String::from_utf8(bytes).unwrap();
1574        assert_eq!("flush_all 0\r\n", command);
1575    }
1576
1577    ///////////
1578    // stats //
1579    ///////////
1580
1581    #[tokio::test]
1582    async fn test_memcached_stats_empty() {
1583        let (_rx, mut client) = client!("END\r\n");
1584        let res = client.stats().await;
1585
1586        assert!(res.is_err());
1587        let err = res.unwrap_err();
1588        assert_eq!(ErrorKind::Runtime, err.kind());
1589    }
1590
1591    #[tokio::test]
1592    async fn test_memcached_stats_error() {
1593        let (_rx, mut client) = client!("SERVER_ERROR backend failure\r\n");
1594        let res = client.stats().await;
1595
1596        assert!(res.is_err());
1597        let err = res.unwrap_err();
1598        assert_eq!(ErrorKind::Protocol, err.kind());
1599    }
1600
1601    #[tokio::test]
1602    async fn test_memcached_stats_success() {
1603        let (_rx, mut client) = client!(
1604            "STAT pid 1525\r\n",
1605            "STAT uptime 271984\r\n",
1606            "STAT time 1687212809\r\n",
1607            "STAT version 1.6.14\r\n",
1608            "STAT libevent 2.1.12-stable\r\n",
1609            "STAT pointer_size 64\r\n",
1610            "STAT rusage_user 17.544323\r\n",
1611            "STAT rusage_system 11.830461\r\n",
1612            "STAT max_connections 1024\r\n",
1613            "STAT curr_connections 1\r\n",
1614            "STAT total_connections 3\r\n",
1615            "STAT rejected_connections 0\r\n",
1616            "STAT connection_structures 2\r\n",
1617            "STAT response_obj_oom 0\r\n",
1618            "STAT response_obj_count 1\r\n",
1619            "STAT response_obj_bytes 32768\r\n",
1620            "STAT read_buf_count 4\r\n",
1621            "STAT read_buf_bytes 65536\r\n",
1622            "STAT read_buf_bytes_free 16384\r\n",
1623            "STAT read_buf_oom 0\r\n",
1624            "STAT reserved_fds 20\r\n",
1625            "STAT cmd_get 1\r\n",
1626            "STAT cmd_set 0\r\n",
1627            "STAT cmd_flush 0\r\n",
1628            "STAT cmd_touch 0\r\n",
1629            "STAT cmd_meta 0\r\n",
1630            "STAT get_hits 0\r\n",
1631            "STAT get_misses 1\r\n",
1632            "STAT get_expired 0\r\n",
1633            "STAT get_flushed 0\r\n",
1634            "STAT delete_misses 0\r\n",
1635            "STAT delete_hits 0\r\n",
1636            "STAT incr_misses 0\r\n",
1637            "STAT incr_hits 0\r\n",
1638            "STAT decr_misses 0\r\n",
1639            "STAT decr_hits 0\r\n",
1640            "STAT cas_misses 0\r\n",
1641            "STAT cas_hits 0\r\n",
1642            "STAT cas_badval 0\r\n",
1643            "STAT touch_hits 0\r\n",
1644            "STAT touch_misses 0\r\n",
1645            "STAT store_too_large 0\r\n",
1646            "STAT store_no_memory 0\r\n",
1647            "STAT auth_cmds 0\r\n",
1648            "STAT auth_errors 0\r\n",
1649            "STAT bytes_read 16\r\n",
1650            "STAT bytes_written 7\r\n",
1651            "STAT limit_maxbytes 67108864\r\n",
1652            "STAT accepting_conns 1\r\n",
1653            "STAT listen_disabled_num 0\r\n",
1654            "STAT time_in_listen_disabled_us 0\r\n",
1655            "STAT threads 4\r\n",
1656            "STAT conn_yields 0\r\n",
1657            "STAT hash_power_level 16\r\n",
1658            "STAT hash_bytes 524288\r\n",
1659            "STAT hash_is_expanding 0\r\n",
1660            "STAT slab_reassign_rescues 0\r\n",
1661            "STAT slab_reassign_chunk_rescues 0\r\n",
1662            "STAT slab_reassign_evictions_nomem 0\r\n",
1663            "STAT slab_reassign_inline_reclaim 0\r\n",
1664            "STAT slab_reassign_busy_items 0\r\n",
1665            "STAT slab_reassign_busy_deletes 0\r\n",
1666            "STAT slab_reassign_running 0\r\n",
1667            "STAT slabs_moved 0\r\n",
1668            "STAT lru_crawler_running 0\r\n",
1669            "STAT lru_crawler_starts 105\r\n",
1670            "STAT lru_maintainer_juggles 271976\r\n",
1671            "STAT malloc_fails 0\r\n",
1672            "STAT log_worker_dropped 0\r\n",
1673            "STAT log_worker_written 0\r\n",
1674            "STAT log_watcher_skipped 0\r\n",
1675            "STAT log_watcher_sent 0\r\n",
1676            "STAT log_watchers 0\r\n",
1677            "STAT unexpected_napi_ids 0\r\n",
1678            "STAT round_robin_fallback 0\r\n",
1679            "STAT bytes 0\r\n",
1680            "STAT curr_items 0\r\n",
1681            "STAT total_items 0\r\n",
1682            "STAT slab_global_page_pool 0\r\n",
1683            "STAT expired_unfetched 0\r\n",
1684            "STAT evicted_unfetched 0\r\n",
1685            "STAT evicted_active 0\r\n",
1686            "STAT evictions 0\r\n",
1687            "STAT reclaimed 0\r\n",
1688            "STAT crawler_reclaimed 0\r\n",
1689            "STAT crawler_items_checked 0\r\n",
1690            "STAT lrutail_reflocked 0\r\n",
1691            "STAT moves_to_cold 0\r\n",
1692            "STAT moves_to_warm 0\r\n",
1693            "STAT moves_within_lru 0\r\n",
1694            "STAT direct_reclaims 0\r\n",
1695            "STAT lru_bumps_dropped 0\r\n",
1696            "END\r\n",
1697        );
1698        let res = client.stats().await.unwrap();
1699
1700        assert_eq!(0, res.cmd_set);
1701        assert_eq!(1, res.cmd_get);
1702        assert_eq!(1, res.get_misses);
1703        assert_eq!(0, res.get_hits);
1704    }
1705
1706    ///////////
1707    // slabs //
1708    ///////////
1709
1710    #[tokio::test]
1711    async fn test_memcached_slabs_empty() {
1712        let (_rx, mut client) = client!("STAT active_slabs 0\r\n", "STAT total_malloced 0\r\n", "END\r\n");
1713        let res = client.slabs().await.unwrap();
1714
1715        assert!(res.0.is_empty());
1716    }
1717
1718    #[tokio::test]
1719    async fn test_memcached_slabs_error() {
1720        let (_rx, mut client) = client!("ERROR Too many open connections\r\n");
1721        let res = client.slabs().await;
1722
1723        assert!(res.is_err());
1724        let err = res.unwrap_err();
1725        assert_eq!(ErrorKind::Protocol, err.kind());
1726    }
1727
1728    #[tokio::test]
1729    async fn test_memcached_slabs_success() {
1730        let (_rx, mut client) = client!(
1731            "STAT 6:chunk_size 304\r\n",
1732            "STAT 6:chunks_per_page 3449\r\n",
1733            "STAT 6:total_pages 1\r\n",
1734            "STAT 6:total_chunks 3449\r\n",
1735            "STAT 6:used_chunks 1\r\n",
1736            "STAT 6:free_chunks 3448\r\n",
1737            "STAT 6:free_chunks_end 0\r\n",
1738            "STAT 6:get_hits 951\r\n",
1739            "STAT 6:cmd_set 100\r\n",
1740            "STAT 6:delete_hits 0\r\n",
1741            "STAT 6:incr_hits 0\r\n",
1742            "STAT 6:decr_hits 0\r\n",
1743            "STAT 6:cas_hits 0\r\n",
1744            "STAT 6:cas_badval 0\r\n",
1745            "STAT 6:touch_hits 0\r\n",
1746            "STAT 7:chunk_size 384\r\n",
1747            "STAT 7:chunks_per_page 2730\r\n",
1748            "STAT 7:total_pages 1\r\n",
1749            "STAT 7:total_chunks 2730\r\n",
1750            "STAT 7:used_chunks 5\r\n",
1751            "STAT 7:free_chunks 2725\r\n",
1752            "STAT 7:free_chunks_end 0\r\n",
1753            "STAT 7:get_hits 4792\r\n",
1754            "STAT 7:cmd_set 520\r\n",
1755            "STAT 7:delete_hits 0\r\n",
1756            "STAT 7:incr_hits 0\r\n",
1757            "STAT 7:decr_hits 0\r\n",
1758            "STAT 7:cas_hits 0\r\n",
1759            "STAT 7:cas_badval 0\r\n",
1760            "STAT 7:touch_hits 0\r\n",
1761            "STAT active_slabs 2\r\n",
1762            "STAT total_malloced 30408704\r\n",
1763            "END\r\n",
1764        );
1765        let res = client.slabs().await.unwrap();
1766
1767        let expected = vec![
1768            Slab {
1769                id: 6,
1770                chunk_size: 304,
1771                chunks_per_page: 3449,
1772                total_pages: 1,
1773                total_chunks: 3449,
1774                used_chunks: 1,
1775                free_chunks: 3448,
1776                get_hits: 951,
1777                cmd_set: 100,
1778                delete_hits: 0,
1779                incr_hits: 0,
1780                decr_hits: 0,
1781                cas_hits: 0,
1782                cas_badval: 0,
1783                touch_hits: 0,
1784            },
1785            Slab {
1786                id: 7,
1787                chunk_size: 384,
1788                chunks_per_page: 2730,
1789                total_pages: 1,
1790                total_chunks: 2730,
1791                used_chunks: 5,
1792                free_chunks: 2725,
1793                get_hits: 4792,
1794                cmd_set: 520,
1795                delete_hits: 0,
1796                incr_hits: 0,
1797                decr_hits: 0,
1798                cas_hits: 0,
1799                cas_badval: 0,
1800                touch_hits: 0,
1801            },
1802        ];
1803
1804        assert_eq!(expected, res.0);
1805    }
1806
1807    ///////////
1808    // items //
1809    ///////////
1810
1811    #[tokio::test]
1812    async fn test_memcached_items_empty() {
1813        let (_rx, mut client) = client!();
1814        let res = client.items().await.unwrap();
1815
1816        assert!(res.is_empty());
1817    }
1818
1819    #[tokio::test]
1820    async fn test_memcached_items_error() {
1821        let (_rx, mut client) = client!("ERROR Too many open connections\r\n");
1822        let res = client.items().await;
1823
1824        assert!(res.is_err());
1825        let err = res.unwrap_err();
1826        assert_eq!(ErrorKind::Protocol, err.kind());
1827    }
1828
1829    #[tokio::test]
1830    async fn test_memcached_items_success() {
1831        let (_rx, mut client) = client!(
1832            "STAT items:39:number 3\r\n",
1833            "STAT items:39:number_hot 0\r\n",
1834            "STAT items:39:number_warm 1\r\n",
1835            "STAT items:39:number_cold 2\r\n",
1836            "STAT items:39:age_hot 0\r\n",
1837            "STAT items:39:age_warm 7\r\n",
1838            "STAT items:39:age 8\r\n",
1839            "STAT items:39:mem_requested 1535788\r\n",
1840            "STAT items:39:evicted 1646\r\n",
1841            "STAT items:39:evicted_nonzero 1646\r\n",
1842            "STAT items:39:evicted_time 0\r\n",
1843            "STAT items:39:outofmemory 9\r\n",
1844            "STAT items:39:tailrepairs 0\r\n",
1845            "STAT items:39:reclaimed 13\r\n",
1846            "STAT items:39:expired_unfetched 4\r\n",
1847            "STAT items:39:evicted_unfetched 202\r\n",
1848            "STAT items:39:evicted_active 6\r\n",
1849            "STAT items:39:crawler_reclaimed 0\r\n",
1850            "STAT items:39:crawler_items_checked 40\r\n",
1851            "STAT items:39:lrutail_reflocked 17365\r\n",
1852            "STAT items:39:moves_to_cold 8703\r\n",
1853            "STAT items:39:moves_to_warm 7285\r\n",
1854            "STAT items:39:moves_within_lru 3651\r\n",
1855            "STAT items:39:direct_reclaims 1949\r\n",
1856            "STAT items:39:hits_to_hot 894\r\n",
1857            "STAT items:39:hits_to_warm 4079\r\n",
1858            "STAT items:39:hits_to_cold 8043\r\n",
1859            "STAT items:39:hits_to_temp 0\r\n",
1860            "END\r\n",
1861        );
1862        let res = client.items().await.unwrap();
1863
1864        let expected = SlabItems(vec![SlabItem {
1865            id: 39,
1866            number: 3,
1867            number_hot: 0,
1868            number_warm: 1,
1869            number_cold: 2,
1870            age_hot: 0,
1871            age_warm: 7,
1872            age: 8,
1873            mem_requested: 1_535_788,
1874            evicted: 1646,
1875            evicted_nonzero: 1646,
1876            evicted_time: 0,
1877            out_of_memory: 9,
1878            tail_repairs: 0,
1879            reclaimed: 13,
1880            expired_unfetched: 4,
1881            evicted_unfetched: 202,
1882            evicted_active: 6,
1883            crawler_reclaimed: 0,
1884            crawler_items_checked: 40,
1885            lrutail_reflocked: 17365,
1886            moves_to_cold: 8703,
1887            moves_to_warm: 7285,
1888            moves_within_lru: 3651,
1889            direct_reclaims: 1949,
1890            hits_to_hot: 894,
1891            hits_to_warm: 4079,
1892            hits_to_cold: 8043,
1893            hits_to_temp: 0,
1894        }]);
1895
1896        assert_eq!(expected, res);
1897    }
1898
1899    //////////
1900    // meta //
1901    //////////
1902
1903    #[tokio::test]
1904    async fn test_memcached_metas_empty() {
1905        let (_rx, mut client) = client!();
1906        let res = client.metas().await.unwrap();
1907
1908        assert!(res.is_empty());
1909    }
1910
1911    #[tokio::test]
1912    async fn test_memcached_metas_error() {
1913        let (_rx, mut client) = client!("BUSY crawler is busy\r\n",);
1914        let res = client.metas().await;
1915
1916        assert!(res.is_err());
1917        let err = res.unwrap_err();
1918        assert_eq!(ErrorKind::Protocol, err.kind());
1919    }
1920
1921    #[tokio::test]
1922    async fn test_memcached_metas_success() {
1923        let (_rx, mut client) = client!(
1924            "key=memcached%2Fmurmur3_hash.c exp=1687216956 la=1687216656 cas=259502 fetch=yes cls=17 size=2912\r\n",
1925            "key=memcached%2Fmd5.h exp=1687216956 la=1687216656 cas=259731 fetch=yes cls=17 size=3593\r\n",
1926            "END\r\n",
1927        );
1928        let res = client.metas().await.unwrap();
1929
1930        let expected = vec![
1931            Meta {
1932                key: "memcached/murmur3_hash.c".to_string(),
1933                expires: 1_687_216_956,
1934                size: 2912,
1935            },
1936            Meta {
1937                key: "memcached/md5.h".to_string(),
1938                expires: 1_687_216_956,
1939                size: 3593,
1940            },
1941        ];
1942
1943        assert_eq!(expected, res);
1944    }
1945}
mtop_client/core.rs

mtop_client/
core.rs
