feox_server/pubsub/
message.rs

1use super::ThreadId;
2use bytes::Bytes;
3
4#[derive(Debug, Clone)]
5pub enum BroadcastMsg {
6    Publish {
7        channel: Vec<u8>,
8        message: Bytes,
9        exclude_thread: Option<ThreadId>,
10    },
11    PatternPublish {
12        channel: Vec<u8>,
13        message: Bytes,
14        exclude_thread: Option<ThreadId>,
15    },
16}
17
18#[derive(Debug, Clone)]
19pub enum PubSubMessage {
20    Message {
21        channel: Vec<u8>,
22        payload: Bytes,
23    },
24    PatternMessage {
25        pattern: Vec<u8>,
26        channel: Vec<u8>,
27        payload: Bytes,
28    },
29    Subscribe {
30        channel: Vec<u8>,
31        count: usize,
32    },
33    Unsubscribe {
34        channel: Option<Vec<u8>>,
35        count: usize,
36    },
37    PSubscribe {
38        pattern: Vec<u8>,
39        count: usize,
40    },
41    PUnsubscribe {
42        pattern: Option<Vec<u8>>,
43        count: usize,
44    },
45}
46
47impl PubSubMessage {
48    pub fn to_resp(&self) -> crate::protocol::resp::RespValue {
49        use crate::protocol::resp::RespValue;
50
51        match self {
52            PubSubMessage::Message { channel, payload } => RespValue::Array(Some(vec![
53                RespValue::BulkString(Some(Bytes::from_static(b"message"))),
54                RespValue::BulkString(Some(Bytes::from(channel.clone()))),
55                RespValue::BulkString(Some(payload.clone())),
56            ])),
57            PubSubMessage::PatternMessage {
58                pattern,
59                channel,
60                payload,
61            } => RespValue::Array(Some(vec![
62                RespValue::BulkString(Some(Bytes::from_static(b"pmessage"))),
63                RespValue::BulkString(Some(Bytes::from(pattern.clone()))),
64                RespValue::BulkString(Some(Bytes::from(channel.clone()))),
65                RespValue::BulkString(Some(payload.clone())),
66            ])),
67            PubSubMessage::Subscribe { channel, count } => RespValue::Array(Some(vec![
68                RespValue::BulkString(Some(Bytes::from_static(b"subscribe"))),
69                RespValue::BulkString(Some(Bytes::from(channel.clone()))),
70                RespValue::Integer(*count as i64),
71            ])),
72            PubSubMessage::Unsubscribe { channel, count } => RespValue::Array(Some(vec![
73                RespValue::BulkString(Some(Bytes::from_static(b"unsubscribe"))),
74                RespValue::BulkString(channel.as_ref().map(|c| Bytes::from(c.clone()))),
75                RespValue::Integer(*count as i64),
76            ])),
77            PubSubMessage::PSubscribe { pattern, count } => RespValue::Array(Some(vec![
78                RespValue::BulkString(Some(Bytes::from_static(b"psubscribe"))),
79                RespValue::BulkString(Some(Bytes::from(pattern.clone()))),
80                RespValue::Integer(*count as i64),
81            ])),
82            PubSubMessage::PUnsubscribe { pattern, count } => RespValue::Array(Some(vec![
83                RespValue::BulkString(Some(Bytes::from_static(b"punsubscribe"))),
84                RespValue::BulkString(pattern.as_ref().map(|p| Bytes::from(p.clone()))),
85                RespValue::Integer(*count as i64),
86            ])),
87        }
88    }
89}
feox_server/pubsub/message.rs

feox_server/pubsub/
message.rs
