rustis 0.23.0

Redis async driver for Rust
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127

use crate::{
    ClientError, ErrorKind, Result,
    client::PubSubMessage,
    resp::{RespBuf, RespFrameParser, RespResponse, RespTapeMut},
};
use bytes::Bytes;

/// Builds the `RespResponse` a push frame decodes to, the way the network
/// handler hands one to the stream.
fn response(buf: &[u8]) -> Result<RespResponse> {
    let mut tape = RespTapeMut::default();
    let (frame, len) = RespFrameParser::new(buf, &mut tape).parse()?;
    Ok(RespResponse::new(
        RespBuf::from(Bytes::copy_from_slice(&buf[..len])),
        frame,
    ))
}

/// A push frame of `kind` carrying `parts` as bulk strings.
fn push(kind: &str, parts: &[&[u8]]) -> Vec<u8> {
    let mut buf = format!(">{}\r\n${}\r\n{kind}\r\n", parts.len() + 1, kind.len()).into_bytes();
    for part in parts {
        buf.extend_from_slice(format!("${}\r\n", part.len()).as_bytes());
        buf.extend_from_slice(part);
        buf.extend_from_slice(b"\r\n");
    }
    buf
}

fn convert(buf: &[u8]) -> Result<PubSubMessage> {
    PubSubMessage::try_from(&response(buf)?)
}

#[test]
fn message_carries_no_pattern() -> Result<()> {
    let message = convert(&push("message", &[b"mychannel", b"mymessage"]))?;

    assert_eq!(b"", message.pattern());
    assert_eq!(b"mychannel", message.channel());
    assert_eq!(b"mymessage", message.payload());

    Ok(())
}

#[test]
fn smessage_carries_no_pattern() -> Result<()> {
    let message = convert(&push("smessage", &[b"mychannel", b"mymessage"]))?;

    assert_eq!(b"", message.pattern());
    assert_eq!(b"mychannel", message.channel());
    assert_eq!(b"mymessage", message.payload());

    Ok(())
}

/// The three segments share one buffer, so the boundaries are what can go
/// wrong: a `pmessage` is the only shape where all three are non-empty.
#[test]
fn pmessage_keeps_its_three_segments_apart() -> Result<()> {
    let message = convert(&push(
        "pmessage",
        &[b"mychannel*", b"mychannel11", b"mymessage"],
    ))?;

    assert_eq!(b"mychannel*", message.pattern());
    assert_eq!(b"mychannel11", message.channel());
    assert_eq!(b"mymessage", message.payload());

    Ok(())
}

/// Empty segments are legal — `PUBLISH mychannel ""` is a valid publish — and
/// they are the degenerate case for offset arithmetic.
#[test]
fn empty_segments_stay_empty_and_distinct() -> Result<()> {
    let message = convert(&push("message", &[b"mychannel", b""]))?;
    assert_eq!(b"", message.pattern());
    assert_eq!(b"mychannel", message.channel());
    assert_eq!(b"", message.payload());

    let message = convert(&push("pmessage", &[b"*", b"", b""]))?;
    assert_eq!(b"*", message.pattern());
    assert_eq!(b"", message.channel());
    assert_eq!(b"", message.payload());

    Ok(())
}

/// The block is sized from the segments, so its length is the one thing every
/// payload size has to agree on.
#[test]
fn a_payload_of_any_size_reads_back_whole() -> Result<()> {
    for size in [1usize, 63, 64, 65, 1024, 64 * 1024] {
        let payload = vec![b'x'; size];
        let message = convert(&push("message", &[b"mychannel", &payload]))?;

        assert_eq!(b"", message.pattern());
        assert_eq!(b"mychannel", message.channel());
        assert_eq!(size, message.payload().len());
        assert_eq!(payload, message.payload());
    }

    Ok(())
}

/// A push the stream cannot be handed — a subscription confirmation, or a frame
/// that is not a push at all — is refused rather than delivered as a message.
#[test]
fn a_push_that_is_not_a_message_is_refused() -> Result<()> {
    for buf in [
        push("subscribe", &[b"mychannel", b"1"]),
        push("unsubscribe", &[b"mychannel", b"0"]),
        push("unknown", &[b"mychannel", b"mymessage"]),
        b"*3\r\n$7\r\nmessage\r\n$9\r\nmychannel\r\n$9\r\nmymessage\r\n".to_vec(),
    ] {
        let error = convert(&buf).unwrap_err();
        assert!(
            matches!(
                error.kind(),
                ErrorKind::Client(ClientError::UnexpectedPubSubMessage)
            ),
            "expected UnexpectedPubSubMessage, got {error:?}"
        );
    }

    Ok(())
}