redis_subscribe/
message.rs

1use thiserror::Error;
2
3use super::parser;
4use crate::Error;
5
6#[derive(Debug)]
7pub enum Message {
8    Subscription {
9        channel: String,
10        subscriptions: i64,
11    },
12    Unsubscription {
13        channel: String,
14        subscriptions: i64,
15    },
16    Message {
17        channel: String,
18        message: String,
19    },
20
21    PatternSubscription {
22        channel: String,
23        subscriptions: i64,
24    },
25    PatternUnsubscription {
26        channel: String,
27        subscriptions: i64,
28    },
29    PatternMessage {
30        pattern: String,
31        channel: String,
32        message: String,
33    },
34    Connected,
35    Disconnected(Error),
36    Error(Error),
37}
38
39#[derive(Error, Debug)]
40pub enum ParserError {
41    #[error("The response has an invalid format.")]
42    MalformedResponse,
43    #[error("The provided channel was invalid.")]
44    InvalidChannel,
45    #[error("The provided amount of subscribers is invalid.")]
46    InvalidSubscriberCount,
47    #[error("The provided pattern is invalid.")]
48    InvalidPattern,
49}
50
51impl Message {
52    /// Parse the response to a message.
53    ///
54    /// # Errors
55    /// Returns an error if the response has unexpected types.
56    pub fn from_response(res: parser::Response) -> crate::Result<Self> {
57        // Make sure the response is a array.
58        let arr = match res {
59            parser::Response::Array(arr) => Ok(arr),
60            _ => Err(ParserError::MalformedResponse),
61        }?;
62
63        // Get the first element of the array.
64        let channel = match arr.get(0) {
65            Some(parser::Response::Bulk(channel)) => Ok(channel.as_str()),
66            _ => Err(ParserError::MalformedResponse),
67        }?;
68
69        // Match on the first element text.
70        match channel.to_lowercase().as_str() {
71            "subscribe" => Self::from_subscribe(&arr),
72            "unsubscribe" => Self::from_unsubscribe(&arr),
73            "message" => Self::from_message(&arr),
74            "pmessage" => Self::from_pmessage(&arr),
75            "psubscribe" => Self::from_psubscribe(&arr),
76            "punsubscribe" => Self::from_punsubscribe(&arr),
77            _ => Err(Error::ParserError(ParserError::MalformedResponse)),
78        }
79    }
80
81    /// parse the subscription message.
82    fn from_subscribe(res: &[parser::Response]) -> crate::Result<Self> {
83        let channel = match res.get(1) {
84            Some(parser::Response::Bulk(channel)) => Ok((*channel).clone()),
85            _ => Err(ParserError::InvalidChannel),
86        }?;
87
88        let subscriptions = match res.get(2) {
89            Some(parser::Response::Integer(subscriptions)) => Ok(*subscriptions),
90            _ => Err(ParserError::InvalidSubscriberCount),
91        }?;
92
93        Ok(Self::Subscription {
94            channel,
95            subscriptions,
96        })
97    }
98
99    fn from_psubscribe(res: &[parser::Response]) -> crate::Result<Self> {
100        let channel = match res.get(1) {
101            Some(parser::Response::Bulk(channel)) => Ok((*channel).clone()),
102            _ => Err(ParserError::InvalidChannel),
103        }?;
104
105        let subscriptions = match res.get(2) {
106            Some(parser::Response::Integer(subscriptions)) => Ok(*subscriptions),
107            _ => Err(ParserError::InvalidSubscriberCount),
108        }?;
109
110        Ok(Self::PatternSubscription {
111            channel,
112            subscriptions,
113        })
114    }
115
116    /// parse the unsubscription message.
117    fn from_unsubscribe(res: &[parser::Response]) -> crate::Result<Self> {
118        let channel = match res.get(1) {
119            Some(parser::Response::Bulk(channel)) => Ok((*channel).clone()),
120            _ => Err(ParserError::InvalidChannel),
121        }?;
122
123        let subscriptions = match res.get(2) {
124            Some(parser::Response::Integer(subscriptions)) => Ok(*subscriptions),
125            _ => Err(ParserError::InvalidSubscriberCount),
126        }?;
127
128        Ok(Self::Unsubscription {
129            channel,
130            subscriptions,
131        })
132    }
133
134    fn from_punsubscribe(res: &[parser::Response]) -> crate::Result<Self> {
135        let channel = match res.get(1) {
136            Some(parser::Response::Bulk(channel)) => Ok((*channel).clone()),
137            _ => Err(ParserError::InvalidChannel),
138        }?;
139
140        let subscriptions = match res.get(2) {
141            Some(parser::Response::Integer(subscriptions)) => Ok(*subscriptions),
142            _ => Err(ParserError::InvalidSubscriberCount),
143        }?;
144
145        Ok(Self::PatternUnsubscription {
146            channel,
147            subscriptions,
148        })
149    }
150
151    /// parse the response to a message.
152    fn from_message(res: &[parser::Response]) -> crate::Result<Self> {
153        let channel = match res.get(1) {
154            Some(parser::Response::Bulk(channel)) => Ok((*channel).clone()),
155            _ => Err(ParserError::InvalidChannel),
156        }?;
157
158        let message = match res.get(2) {
159            Some(parser::Response::Bulk(message)) => Ok((*message).clone()),
160            _ => Err(ParserError::InvalidSubscriberCount),
161        }?;
162
163        Ok(Self::Message { channel, message })
164    }
165
166    /// parse the response to a pattern message
167    fn from_pmessage(res: &[parser::Response]) -> crate::Result<Self> {
168        let pattern = match res.get(1) {
169            Some(parser::Response::Bulk(pattern)) => Ok((*pattern).clone()),
170            _ => Err(ParserError::InvalidPattern),
171        }?;
172
173        let channel = match res.get(2) {
174            Some(parser::Response::Bulk(channel)) => Ok((*channel).clone()),
175            _ => Err(ParserError::InvalidChannel),
176        }?;
177
178        let message = match res.get(3) {
179            Some(parser::Response::Bulk(message)) => Ok((*message).clone()),
180            _ => Err(ParserError::InvalidSubscriberCount),
181        }?;
182
183        Ok(Self::PatternMessage {
184            pattern,
185            channel,
186            message,
187        })
188    }
189}
190
191impl Message {
192    #[must_use]
193    #[inline]
194    pub const fn is_subscription(&self) -> bool {
195        matches!(self, Self::Subscription { .. })
196    }
197
198    #[must_use]
199    #[inline]
200    pub const fn is_pattern_subscription(&self) -> bool {
201        matches!(self, Self::PatternSubscription { .. })
202    }
203
204    #[must_use]
205    #[inline]
206    pub const fn is_unsubscription(&self) -> bool {
207        matches!(self, Self::Unsubscription { .. })
208    }
209
210    #[must_use]
211    #[inline]
212    pub const fn is_pattern_unsubscription(&self) -> bool {
213        matches!(self, Self::PatternUnsubscription { .. })
214    }
215
216    #[must_use]
217    #[inline]
218    pub const fn is_message(&self) -> bool {
219        matches!(self, Self::Message { .. })
220    }
221
222    #[must_use]
223    #[inline]
224    pub const fn is_pattern_message(&self) -> bool {
225        matches!(self, Self::PatternMessage { .. })
226    }
227
228    #[must_use]
229    #[inline]
230    pub const fn is_connected(&self) -> bool {
231        matches!(self, Self::Connected)
232    }
233
234    #[must_use]
235    #[inline]
236    pub const fn is_disconnected(&self) -> bool {
237        matches!(self, Self::Disconnected(_))
238    }
239
240    #[must_use]
241    #[inline]
242    pub const fn is_error(&self) -> bool {
243        matches!(self, Self::Error(_))
244    }
245}
redis_subscribe/message.rs

redis_subscribe/
message.rs
