ntex_redis/
codec.rs

1//! Redis protocol codec
2use std::{cmp, collections::HashMap, convert::TryFrom, hash::BuildHasher, hash::Hash, str};
3
4use ntex::codec::{Decoder, Encoder};
5use ntex::util::{Buf, BufMut, ByteString, Bytes, BytesMut};
6
7use super::errors::Error;
8
9/// Codec to read/write redis values
10pub struct Codec;
11
12impl Encoder for Codec {
13    type Item = Request;
14    type Error = Error;
15
16    fn encode(&self, msg: Request, buf: &mut BytesMut) -> Result<(), Self::Error> {
17        match msg {
18            Request::Array(ary) => {
19                write_header(b'*', ary.len() as i64, buf, 0);
20                for v in ary {
21                    self.encode(v, buf)?;
22                }
23            }
24            Request::BulkString(bstr) => {
25                let len = bstr.0.len();
26                write_header(b'$', len as i64, buf, len + 2);
27                buf.extend_from_slice(&bstr.0[..]);
28                write_rn(buf);
29            }
30            Request::BulkStatic(bstr) => {
31                let len = bstr.len();
32                write_header(b'$', len as i64, buf, len + 2);
33                buf.extend_from_slice(bstr);
34                write_rn(buf);
35            }
36            Request::BulkInteger(i) => {
37                let mut buffer = itoa::Buffer::new();
38                let rendered = buffer.format(i);
39                write_header(b'$', rendered.len() as i64, buf, rendered.len() + 2);
40                buf.extend_from_slice(rendered.as_bytes());
41                write_rn(buf);
42            }
43            Request::String(ref string) => {
44                write_string(b'+', string, buf);
45            }
46            Request::Integer(val) => {
47                // Simple integer are just the header
48                write_header(b':', val, buf, 0);
49            }
50        }
51        Ok(())
52    }
53}
54
55impl Decoder for Codec {
56    type Item = Response;
57    type Error = Error;
58
59    fn decode(&self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
60        match decode(buf, 0)? {
61            Some((pos, item)) => {
62                buf.advance(pos);
63                Ok(Some(item))
64            }
65            None => Ok(None),
66        }
67    }
68}
69
70#[derive(Debug, Clone, Eq, PartialEq, Hash)]
71/// A bulk string.
72///
73/// In Redis terminology a string is a byte-array, so this is stored as a
74/// vector of `u8`s to allow clients to interpret the bytes as appropriate.
75pub struct BulkString(Bytes);
76
77impl BulkString {
78    /// Create request from static str
79    pub fn from_static(data: &'static str) -> Self {
80        BulkString(Bytes::from_static(data.as_ref()))
81    }
82
83    /// Create request from static str
84    pub fn from_bstatic(data: &'static [u8]) -> Self {
85        BulkString(Bytes::from_static(data))
86    }
87}
88
89impl From<ByteString> for BulkString {
90    fn from(val: ByteString) -> BulkString {
91        BulkString(val.into_bytes())
92    }
93}
94
95impl From<String> for BulkString {
96    fn from(val: String) -> BulkString {
97        BulkString(Bytes::from(val))
98    }
99}
100
101impl<'a> From<&'a String> for BulkString {
102    fn from(val: &'a String) -> BulkString {
103        BulkString(Bytes::copy_from_slice(val.as_ref()))
104    }
105}
106
107impl<'a> From<&'a str> for BulkString {
108    fn from(val: &'a str) -> BulkString {
109        BulkString(Bytes::copy_from_slice(val.as_bytes()))
110    }
111}
112
113impl<'a> From<&&'a str> for BulkString {
114    fn from(val: &&'a str) -> BulkString {
115        BulkString(Bytes::copy_from_slice(val.as_bytes()))
116    }
117}
118
119impl From<Bytes> for BulkString {
120    fn from(val: Bytes) -> BulkString {
121        BulkString(val)
122    }
123}
124
125impl From<BytesMut> for BulkString {
126    fn from(val: BytesMut) -> BulkString {
127        BulkString(val.freeze())
128    }
129}
130
131impl<'a> From<&'a Bytes> for BulkString {
132    fn from(val: &'a Bytes) -> BulkString {
133        BulkString(val.clone())
134    }
135}
136
137impl<'a> From<&'a ByteString> for BulkString {
138    fn from(val: &'a ByteString) -> BulkString {
139        BulkString(val.clone().into_bytes())
140    }
141}
142
143impl<'a> From<&'a [u8]> for BulkString {
144    fn from(val: &'a [u8]) -> BulkString {
145        BulkString(Bytes::copy_from_slice(val))
146    }
147}
148
149impl From<Vec<u8>> for BulkString {
150    fn from(val: Vec<u8>) -> BulkString {
151        BulkString(Bytes::from(val))
152    }
153}
154
155/// A single RESP value, this owns the data that is to-be written to Redis.
156///
157/// It is cloneable to allow multiple copies to be delivered in certain circumstances, e.g. multiple
158/// subscribers to the same topic.
159#[derive(Debug, Clone, Eq, PartialEq, Hash)]
160pub enum Request {
161    /// Zero, one or more other `Reqeests`s.
162    Array(Vec<Request>),
163
164    /// A bulk string. In Redis terminology a string is a byte-array, so this is stored as a
165    /// vector of `u8`s to allow clients to interpret the bytes as appropriate.
166    BulkString(BulkString),
167
168    /// A bulk string. In Redis terminology a string is a byte-array, so this is stored as a
169    /// vector of `u8`s to allow clients to interpret the bytes as appropriate.
170    BulkStatic(&'static [u8]),
171
172    /// Convert integer to string representation.
173    BulkInteger(i64),
174
175    /// A valid utf-8 string
176    String(ByteString),
177
178    /// Redis documentation defines an integer as being a signed 64-bit integer:
179    /// https://redis.io/topics/protocol#resp-integers
180    Integer(i64),
181}
182
183impl Request {
184    /// Create request from static str
185    pub fn from_static(data: &'static str) -> Self {
186        Request::BulkStatic(data.as_ref())
187    }
188
189    /// Create request from static str
190    pub fn from_bstatic(data: &'static [u8]) -> Self {
191        Request::BulkStatic(data)
192    }
193
194    #[allow(clippy::should_implement_trait)]
195    /// Convenience function for building dynamic Redis commands with variable numbers of
196    /// arguments, e.g. RPUSH
197    ///
198    /// Self get converted to array if it is not an array.
199    pub fn add<T>(mut self, other: T) -> Self
200    where
201        Request: From<T>,
202    {
203        match self {
204            Request::Array(ref mut vals) => {
205                vals.push(other.into());
206                self
207            }
208            _ => Request::Array(vec![self, other.into()]),
209        }
210    }
211
212    /// Convenience function for building dynamic Redis commands with variable numbers of
213    /// arguments, e.g. RPUSH
214    ///
215    /// Self get converted to array if it is not an array.
216    pub fn extend<T>(mut self, other: impl IntoIterator<Item = T>) -> Self
217    where
218        Request: From<T>,
219    {
220        match self {
221            Request::Array(ref mut vals) => {
222                vals.extend(other.into_iter().map(|t| t.into()));
223                self
224            }
225            _ => {
226                let mut vals = vec![self];
227                vals.extend(other.into_iter().map(|t| t.into()));
228                Request::Array(vals)
229            }
230        }
231    }
232}
233
234impl<T> From<T> for Request
235where
236    BulkString: From<T>,
237{
238    fn from(val: T) -> Request {
239        Request::BulkString(val.into())
240    }
241}
242
243impl From<i8> for Request {
244    fn from(val: i8) -> Request {
245        Request::Integer(val as i64)
246    }
247}
248
249impl From<i16> for Request {
250    fn from(val: i16) -> Request {
251        Request::Integer(val as i64)
252    }
253}
254
255impl From<i32> for Request {
256    fn from(val: i32) -> Request {
257        Request::Integer(val as i64)
258    }
259}
260
261impl From<i64> for Request {
262    fn from(val: i64) -> Request {
263        Request::Integer(val)
264    }
265}
266
267impl From<u8> for Request {
268    fn from(val: u8) -> Request {
269        Request::Integer(val as i64)
270    }
271}
272
273impl From<u16> for Request {
274    fn from(val: u16) -> Request {
275        Request::Integer(val as i64)
276    }
277}
278
279impl From<u32> for Request {
280    fn from(val: u32) -> Request {
281        Request::Integer(val as i64)
282    }
283}
284
285impl From<usize> for Request {
286    fn from(val: usize) -> Request {
287        Request::Integer(val as i64)
288    }
289}
290
291/// A single RESP value, this owns the data that is read from Redis.
292#[derive(Debug, Clone, Eq, PartialEq, Hash)]
293pub enum Response {
294    Nil,
295
296    /// Zero, one or more other `Response`s.
297    Array(Vec<Response>),
298
299    /// A bulk string. In Redis terminology a string is a byte-array, so this is stored as a
300    /// vector of `u8`s to allow clients to interpret the bytes as appropriate.
301    Bytes(Bytes),
302
303    /// A valid utf-8 string
304    String(ByteString),
305
306    /// An error from the Redis server
307    Error(ByteString),
308
309    /// Redis documentation defines an integer as being a signed 64-bit integer:
310    /// https://redis.io/topics/protocol#resp-integers
311    Integer(i64),
312}
313
314impl Response {
315    /// Extract redis server error to Result
316    pub fn into_result(self) -> Result<Response, ByteString> {
317        match self {
318            Response::Error(val) => Err(val),
319            val => Ok(val),
320        }
321    }
322}
323
324impl TryFrom<Response> for Bytes {
325    type Error = (&'static str, Response);
326
327    fn try_from(val: Response) -> Result<Self, Self::Error> {
328        if let Response::Bytes(bytes) = val {
329            Ok(bytes)
330        } else {
331            Err(("Not a bytes object", val))
332        }
333    }
334}
335
336impl TryFrom<Response> for ByteString {
337    type Error = (&'static str, Response);
338
339    fn try_from(val: Response) -> Result<Self, Self::Error> {
340        match val {
341            Response::String(val) => Ok(val),
342            Response::Bytes(val) => {
343                if let Ok(val) = ByteString::try_from(val) {
344                    Ok(val)
345                } else {
346                    Err(("Cannot convert into a string", Response::Nil))
347                }
348            }
349            _ => Err(("Cannot convert into a string", val)),
350        }
351    }
352}
353
354impl TryFrom<Response> for i64 {
355    type Error = (&'static str, Response);
356
357    fn try_from(val: Response) -> Result<Self, Self::Error> {
358        if let Response::Integer(i) = val {
359            Ok(i)
360        } else {
361            Err(("Cannot be converted into an i64", val))
362        }
363    }
364}
365
366impl TryFrom<Response> for bool {
367    type Error = (&'static str, Response);
368
369    fn try_from(val: Response) -> Result<bool, Self::Error> {
370        i64::try_from(val).and_then(|x| match x {
371            0 => Ok(false),
372            1 => Ok(true),
373            _ => Err((
374                "i64 value cannot be represented as bool",
375                Response::Integer(x),
376            )),
377        })
378    }
379}
380
381impl<T> TryFrom<Response> for Vec<T>
382where
383    T: TryFrom<Response, Error = (&'static str, Response)>,
384{
385    type Error = (&'static str, Response);
386
387    fn try_from(val: Response) -> Result<Vec<T>, Self::Error> {
388        if let Response::Array(ary) = val {
389            let mut ar = Vec::with_capacity(ary.len());
390            for value in ary {
391                ar.push(T::try_from(value)?);
392            }
393            Ok(ar)
394        } else {
395            Err(("Cannot be converted into a vector", val))
396        }
397    }
398}
399
400impl TryFrom<Response> for () {
401    type Error = (&'static str, Response);
402
403    fn try_from(val: Response) -> Result<(), Self::Error> {
404        if let Response::String(string) = val {
405            match string.as_ref() {
406                "OK" => Ok(()),
407                _ => Err(("Unexpected value within String", Response::String(string))),
408            }
409        } else {
410            Err(("Unexpected value", val))
411        }
412    }
413}
414
415impl<A, B> TryFrom<Response> for (A, B)
416where
417    A: TryFrom<Response, Error = (&'static str, Response)>,
418    B: TryFrom<Response, Error = (&'static str, Response)>,
419{
420    type Error = (&'static str, Response);
421
422    fn try_from(val: Response) -> Result<(A, B), Self::Error> {
423        match val {
424            Response::Array(ary) => {
425                if ary.len() == 2 {
426                    let mut ary_iter = ary.into_iter();
427                    Ok((
428                        A::try_from(ary_iter.next().expect("No value"))?,
429                        B::try_from(ary_iter.next().expect("No value"))?,
430                    ))
431                } else {
432                    Err(("Array needs to be 2 elements", Response::Array(ary)))
433                }
434            }
435            _ => Err(("Unexpected value", val)),
436        }
437    }
438}
439
440impl<A, B, C> TryFrom<Response> for (A, B, C)
441where
442    A: TryFrom<Response, Error = (&'static str, Response)>,
443    B: TryFrom<Response, Error = (&'static str, Response)>,
444    C: TryFrom<Response, Error = (&'static str, Response)>,
445{
446    type Error = (&'static str, Response);
447
448    fn try_from(val: Response) -> Result<(A, B, C), Self::Error> {
449        match val {
450            Response::Array(ary) => {
451                if ary.len() == 3 {
452                    let mut ary_iter = ary.into_iter();
453                    Ok((
454                        A::try_from(ary_iter.next().expect("No value"))?,
455                        B::try_from(ary_iter.next().expect("No value"))?,
456                        C::try_from(ary_iter.next().expect("No value"))?,
457                    ))
458                } else {
459                    Err(("Array needs to be 3 elements", Response::Array(ary)))
460                }
461            }
462            _ => Err(("Unexpected value", val)),
463        }
464    }
465}
466
467impl<K, T, S> TryFrom<Response> for HashMap<K, T, S>
468where
469    K: TryFrom<Response, Error = (&'static str, Response)> + Hash + Eq,
470    T: TryFrom<Response, Error = (&'static str, Response)>,
471    S: BuildHasher + Default,
472{
473    type Error = (&'static str, Response);
474
475    fn try_from(val: Response) -> Result<HashMap<K, T, S>, Self::Error> {
476        match val {
477            Response::Array(ary) => {
478                let mut map = HashMap::with_capacity_and_hasher(ary.len() / 2, S::default());
479                let mut items = ary.into_iter();
480
481                while let Some(k) = items.next() {
482                    let key = K::try_from(k)?;
483                    let value = T::try_from(items.next().ok_or((
484                        "Cannot convert an odd number of elements into a hashmap",
485                        Response::Nil,
486                    ))?)?;
487                    map.insert(key, value);
488                }
489
490                Ok(map)
491            }
492            _ => Err(("Cannot be converted into a hashmap", val)),
493        }
494    }
495}
496
497macro_rules! impl_tryfrom_integers {
498    ($($int_ty:ident),* $(,)*) => {
499        $(
500            #[allow(clippy::cast_lossless)]
501            impl TryFrom<Response> for $int_ty {
502                type Error = (&'static str, Response);
503
504                fn try_from(val: Response) -> Result<Self, Self::Error> {
505                    i64::try_from(val).and_then(|x| {
506                        // $int_ty::max_value() as i64 > 0 should be optimized out. It tests if
507                        // the target integer type needs an "upper bounds" check
508                        if x < ($int_ty::min_value() as i64)
509                            || ($int_ty::max_value() as i64 > 0
510                                && x > ($int_ty::max_value() as i64))
511                        {
512                            Err((
513                                concat!(
514                                    "i64 value cannot be represented as {}",
515                                    stringify!($int_ty),
516                                ),
517                                Response::Integer(x),
518                            ))
519                        } else {
520                            Ok(x as $int_ty)
521                        }
522                    })
523                }
524            }
525        )*
526    };
527}
528
529impl_tryfrom_integers!(isize, usize, i32, u32, u64);
530
531fn write_rn(buf: &mut BytesMut) {
532    buf.extend_from_slice(b"\r\n");
533}
534
535fn write_header(symb: u8, len: i64, buf: &mut BytesMut, body_size: usize) {
536    let mut len_buf = itoa::Buffer::new();
537    let rendered = len_buf.format(len);
538    buf.reserve(3 + rendered.len() + body_size);
539    buf.put_u8(symb);
540    buf.extend_from_slice(rendered.as_bytes());
541    write_rn(buf);
542}
543
544fn write_string(symb: u8, string: &str, buf: &mut BytesMut) {
545    let bytes = string.as_bytes();
546    buf.reserve(3 + bytes.len());
547    buf.put_u8(symb);
548    buf.extend_from_slice(bytes);
549    write_rn(buf);
550}
551
552type DecodeResult = Result<Option<(usize, Response)>, Error>;
553
554fn decode(buf: &mut BytesMut, idx: usize) -> DecodeResult {
555    if buf.len() > idx {
556        match buf[idx] {
557            b'$' => decode_bytes(buf, idx + 1),
558            b'*' => decode_array(buf, idx + 1),
559            b':' => decode_integer(buf, idx + 1),
560            b'+' => decode_string(buf, idx + 1),
561            b'-' => decode_error(buf, idx + 1),
562            _ => Err(Error::Parse(format!("Unexpected byte: {}", buf[idx]))),
563        }
564    } else {
565        Ok(None)
566    }
567}
568
569fn decode_length(buf: &mut BytesMut, idx: usize) -> Result<Option<(usize, i64)>, Error> {
570    // length is encoded as a string, terminated by "\r\n"
571    let (pos, int_str) = if let Some(pos) = buf[idx..].windows(2).position(|w| w == b"\r\n") {
572        (idx + pos + 2, &buf[idx..idx + pos])
573    } else {
574        return Ok(None);
575    };
576
577    // int encoded as string
578    match btoi::btoi(int_str) {
579        Ok(int) => Ok(Some((pos, int))),
580        Err(_) => Err(Error::Parse(format!(
581            "Not an integer: {:?}",
582            &int_str[..cmp::min(int_str.len(), 10)]
583        ))),
584    }
585}
586
587fn decode_bytes(buf: &mut BytesMut, idx: usize) -> DecodeResult {
588    match decode_length(buf, idx)? {
589        Some((pos, -1)) => Ok(Some((pos, Response::Nil))),
590        Some((pos, size)) if size >= 0 => {
591            let size = size as usize;
592            let remaining = buf.len() - pos;
593            let required_bytes = size + 2;
594
595            if remaining < required_bytes {
596                return Ok(None);
597            }
598            buf.advance(pos);
599            Ok(Some((2, Response::Bytes(buf.split_to(size).freeze()))))
600        }
601        Some((_, size)) => Err(Error::Parse(format!("Invalid string size: {}", size))),
602        None => Ok(None),
603    }
604}
605
606fn is_array_ready_to_decode(
607    buf: &mut BytesMut,
608    idx: usize,
609    array_size: usize,
610) -> Result<(bool, usize), Error> {
611    let mut items: usize = 0;
612    let mut pos = idx;
613
614    // counting beginning of array items in buffer by `\r\n<type>`
615    loop {
616        let Some(new_pos) = buf[pos..].windows(2).position(|w| w.starts_with(b"\r\n")) else {
617            break;
618        };
619
620        if pos + new_pos + 2 >= buf.len() {
621            break;
622        }
623        pos += new_pos + 2;
624
625        items += match &buf[pos] {
626            // check nested array and calc it as item
627            b'*' => match decode_length(buf, pos) {
628                Ok(Some((_, -1))) => 1,
629                Ok(Some((p, size))) if size >= 0 => {
630                    let (ready, end_of_scan) = is_array_ready_to_decode(buf, p, size as usize)?;
631                    // nested array isn't ready
632                    if !ready {
633                        return Ok((false, end_of_scan));
634                    }
635                    pos = end_of_scan;
636                    1
637                }
638                Ok(Some((_, size))) => {
639                    return Err(Error::Parse(format!("Invalid array size: {}", size)))
640                }
641                _ => 0,
642            },
643            // array item found
644            b'$' | b':' | b'+' | b'-' => 1,
645            _ => 0,
646        };
647
648        if array_size <= items {
649            return Ok((true, pos));
650        }
651    }
652
653    Ok((array_size <= items, pos))
654}
655
656fn decode_array(buf: &mut BytesMut, idx: usize) -> DecodeResult {
657    match decode_length(buf, idx)? {
658        Some((pos, -1)) => Ok(Some((pos, Response::Nil))),
659        Some((pos, size)) if size >= 0 => {
660            let size = size as usize;
661
662            let (is_ready, _) = is_array_ready_to_decode(buf, idx, size)?;
663            if !is_ready {
664                return Ok(None);
665            }
666
667            let mut pos = pos;
668            let mut values = Vec::with_capacity(size);
669            for _ in 0..size {
670                match decode(buf, pos) {
671                    Ok(None) => return Ok(None),
672                    Ok(Some((new_pos, value))) => {
673                        values.push(value);
674                        pos = new_pos;
675                    }
676                    Err(e) => return Err(e),
677                }
678            }
679            Ok(Some((pos, Response::Array(values))))
680        }
681        Some((_, size)) => Err(Error::Parse(format!("Invalid array size: {}", size))),
682        None => Ok(None),
683    }
684}
685
686fn decode_integer(buf: &mut BytesMut, idx: usize) -> DecodeResult {
687    if let Some((pos, int)) = decode_length(buf, idx)? {
688        Ok(Some((pos, Response::Integer(int))))
689    } else {
690        Ok(None)
691    }
692}
693
694/// A simple string is any series of bytes that ends with `\r\n`
695fn decode_string(buf: &mut BytesMut, idx: usize) -> DecodeResult {
696    if let Some((pos, string)) = scan_string(buf, idx)? {
697        Ok(Some((pos, Response::String(string))))
698    } else {
699        Ok(None)
700    }
701}
702
703fn decode_error(buf: &mut BytesMut, idx: usize) -> DecodeResult {
704    if let Some((pos, string)) = scan_string(buf, idx)? {
705        Ok(Some((pos, Response::Error(string))))
706    } else {
707        Ok(None)
708    }
709}
710
711fn scan_string(buf: &mut BytesMut, idx: usize) -> Result<Option<(usize, ByteString)>, Error> {
712    if let Some(pos) = buf[idx..].windows(2).position(|w| w == b"\r\n") {
713        buf.advance(idx);
714        match ByteString::try_from(buf.split_to(pos)) {
715            Ok(s) => Ok(Some((2, s))),
716            Err(_) => Err(Error::Parse(format!(
717                "Not a valid string: {:?}",
718                &buf[idx..idx + cmp::min(pos, 10)]
719            ))),
720        }
721    } else {
722        Ok(None)
723    }
724}
725
726#[cfg(test)]
727mod tests {
728    use std::convert::TryFrom;
729
730    use ntex::codec::{Decoder, Encoder};
731    use ntex::util::{ByteString, Bytes, BytesMut, HashMap};
732
733    use super::*;
734    use crate::array;
735
736    fn obj_to_bytes(obj: Request) -> Bytes {
737        let mut bytes = BytesMut::new();
738        Codec.encode(obj, &mut bytes).unwrap();
739        bytes.freeze()
740    }
741
742    #[test]
743    fn test_array_macro() {
744        let resp_object = array!["SET", "x"];
745        let bytes = obj_to_bytes(resp_object);
746        assert_eq!(bytes, b"*2\r\n$3\r\nSET\r\n$1\r\nx\r\n".as_ref());
747
748        let resp_object = array!["RPUSH", "wyz"].extend(vec!["a", "b"]);
749        let bytes = obj_to_bytes(resp_object);
750        assert_eq!(
751            bytes,
752            b"*4\r\n$5\r\nRPUSH\r\n$3\r\nwyz\r\n$1\r\na\r\n$1\r\nb\r\n".as_ref(),
753        );
754
755        let vals = vec!["a", "b"];
756        let resp_object = array!["RPUSH", "xyz"].extend(&vals);
757        let bytes = obj_to_bytes(resp_object);
758        assert_eq!(
759            bytes,
760            &b"*4\r\n$5\r\nRPUSH\r\n$3\r\nxyz\r\n$1\r\na\r\n$1\r\nb\r\n"[..],
761        );
762    }
763
764    #[test]
765    fn test_bulk_string() {
766        let req_object = Request::BulkString(Bytes::from_static(b"THISISATEST").into());
767        let mut bytes = BytesMut::new();
768        let codec = Codec;
769        codec.encode(req_object.clone(), &mut bytes).unwrap();
770        assert_eq!(b"$11\r\nTHISISATEST\r\n".to_vec(), bytes.to_vec());
771
772        let resp_object = Response::Bytes(Bytes::from_static(b"THISISATEST"));
773        let deserialized = codec.decode(&mut bytes).unwrap().unwrap();
774        assert_eq!(deserialized, resp_object);
775    }
776
777    #[test]
778    fn test_array() {
779        let req_object = Request::Array(vec![b"TEST1".as_ref().into(), b"TEST2".as_ref().into()]);
780        let mut bytes = BytesMut::new();
781        let codec = Codec;
782        codec.encode(req_object.clone(), &mut bytes).unwrap();
783        assert_eq!(
784            b"*2\r\n$5\r\nTEST1\r\n$5\r\nTEST2\r\n".to_vec(),
785            bytes.to_vec()
786        );
787
788        let resp = Response::Array(vec![
789            Response::Bytes(Bytes::from_static(b"TEST1")),
790            Response::Bytes(Bytes::from_static(b"TEST2")),
791        ]);
792        let deserialized = codec.decode(&mut bytes).unwrap().unwrap();
793        assert_eq!(deserialized, resp);
794    }
795
796    #[test]
797    fn test_decode_array() {
798        let codec = Codec;
799
800        let resp = Response::Array(vec![
801            Response::Bytes(Bytes::from_static(b"TEST1")),
802            Response::Bytes(Bytes::from_static(b"TEST2")),
803        ]);
804
805        let mut bytes = BytesMut::copy_from_slice(b"*2\r\n$5\r\nTEST1\r\n$5\r\nTEST2\r\n");
806        let deserialized = codec.decode(&mut bytes).unwrap().unwrap();
807        assert_eq!(deserialized, resp);
808
809        // uncomplete array data
810        let mut bytes = BytesMut::copy_from_slice(b"*2\r\n$5\r\nTEST1\r\n");
811        let result = codec.decode(&mut bytes).unwrap();
812        assert!(result.is_none());
813
814        // receiving remain data
815        bytes.extend_from_slice(b"$5\r\nTEST2\r\n");
816
817        let deserialized = codec.decode(&mut bytes).unwrap().unwrap();
818        assert_eq!(deserialized, resp);
819
820        // $ echo -e "mget key1 key2\r\nquit" | curl -s telnet://localhost:6379 | python -c "import sys; print(repr(sys.stdin.read()))"
821        // '*2\r\n$-1\r\n$-1\r\n+OK\r\n'
822        let mut bytes = BytesMut::copy_from_slice(b"*2\r\n$-1\r\n$-1\r\n");
823        let result = codec.decode(&mut bytes).unwrap();
824
825        assert_eq!(
826            result,
827            Some(Response::Array(vec![Response::Nil, Response::Nil]))
828        );
829
830        // uncomplete nested array data
831        // [[['a']]]
832        // $ echo -e 'eval "return {{{\'a\'}}}" 0\r\nquit' | curl -s telnet://localhost:6379 | python -c "import sys; print(repr(sys.stdin.read()))"
833        // '*1\r\n*1\r\n*1\r\n$1\r\na\r\n+OK\r\n'
834        let mut bytes = BytesMut::copy_from_slice(b"*1\r\n*1\r\n*1\r\n");
835        let result = codec.decode(&mut bytes).unwrap();
836        assert_eq!(result, None);
837
838        // receiving remain parts
839        bytes.extend_from_slice(b"$1\r\na");
840        let result = codec.decode(&mut bytes).unwrap();
841        assert_eq!(result, None);
842
843        bytes.extend_from_slice(b"\r");
844        let result = codec.decode(&mut bytes).unwrap();
845        assert_eq!(result, None);
846
847        bytes.extend_from_slice(b"\n");
848        let result = codec.decode(&mut bytes).unwrap();
849
850        assert_eq!(
851            result,
852            Some(Response::Array(vec![Response::Array(vec![
853                Response::Array(vec![Response::Bytes(Bytes::from_static(b"a"))])
854            ])]))
855        );
856    }
857
858    #[test]
859    fn test_nil_string() {
860        let mut bytes = BytesMut::new();
861        bytes.extend_from_slice(&b"$-1\r\n"[..]);
862
863        let codec = Codec;
864        let deserialized = codec.decode(&mut bytes).unwrap().unwrap();
865        assert_eq!(deserialized, Response::Nil);
866    }
867
868    #[test]
869    fn test_integer_overflow() {
870        let resp_object = Response::Integer(i64::max_value());
871        let res = i32::try_from(resp_object);
872        assert!(res.is_err());
873    }
874
875    #[test]
876    fn test_integer_underflow() {
877        let resp_object = Response::Integer(-2);
878        let res = u64::try_from(resp_object);
879        assert!(res.is_err());
880    }
881
882    #[test]
883    fn test_integer_convesion() {
884        let resp_object = Response::Integer(50);
885        assert_eq!(u32::try_from(resp_object).unwrap(), 50);
886    }
887
888    #[test]
889    fn test_hashmap_conversion() {
890        let mut expected = HashMap::default();
891        expected.insert(
892            ByteString::from("KEY1").into(),
893            ByteString::from("VALUE1").into(),
894        );
895        expected.insert(
896            ByteString::from("KEY2").into(),
897            ByteString::from("VALUE2").into(),
898        );
899
900        let resp_object = Response::Array(vec![
901            Response::String(ByteString::from_static("KEY1")),
902            Response::String(ByteString::from_static("VALUE1")),
903            Response::String(ByteString::from_static("KEY2")),
904            Response::String(ByteString::from_static("VALUE2")),
905        ]);
906        assert_eq!(
907            HashMap::<ByteString, ByteString>::try_from(resp_object).unwrap(),
908            expected
909        );
910    }
911
912    #[test]
913    fn test_hashmap_conversion_fails_with_odd_length_array() {
914        let resp_object = Response::Array(vec![
915            Response::String(ByteString::from_static("KEY1")),
916            Response::String(ByteString::from_static("VALUE1")),
917            Response::String(ByteString::from_static("KEY2")),
918            Response::String(ByteString::from_static("VALUE2")),
919            Response::String(ByteString::from_static("KEY3")),
920        ]);
921        let res = HashMap::<ByteString, ByteString>::try_from(resp_object);
922
923        match res {
924            Err((_, _)) => {}
925            _ => panic!("Should not be able to convert an odd number of elements to a hashmap"),
926        }
927    }
928}
ntex_redis/codec.rs

ntex_redis/
codec.rs
