rocketmq_protocol/protocol/heartbeat/
consumer_data.rs

1// Copyright 2023 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::HashSet;
16use std::hash::Hash;
17
18use cheetah_string::CheetahString;
19use rocketmq_model::common::consumer::consume_from_where::ConsumeFromWhere;
20use serde::Deserialize;
21use serde::Serialize;
22
23use crate::protocol::heartbeat::consume_type::ConsumeType;
24use crate::protocol::heartbeat::message_model::MessageModel;
25use crate::protocol::heartbeat::subscription_data::SubscriptionData;
26
27#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq, Default)]
28#[serde(rename_all = "camelCase")]
29pub struct ConsumerData {
30    pub group_name: CheetahString,
31    pub consume_type: ConsumeType,
32    pub message_model: MessageModel,
33    pub consume_from_where: ConsumeFromWhere,
34    pub subscription_data_set: HashSet<SubscriptionData>,
35    pub unit_mode: bool,
36}
37
38impl Hash for ConsumerData {
39    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
40        self.group_name.hash(state);
41        self.consume_type.hash(state);
42        self.message_model.hash(state);
43        self.consume_from_where.hash(state);
44        let mut subscriptions = self.subscription_data_set.iter().collect::<Vec<_>>();
45        subscriptions.sort_unstable();
46        subscriptions.hash(state);
47        self.unit_mode.hash(state);
48    }
49}
50
51#[cfg(test)]
52mod tests {
53    use std::collections::HashSet;
54
55    use cheetah_string::CheetahString;
56    use rocketmq_model::common::consumer::consume_from_where::ConsumeFromWhere;
57
58    use super::*;
59    use crate::protocol::heartbeat::consume_type::ConsumeType;
60    use crate::protocol::heartbeat::message_model::MessageModel;
61    use crate::protocol::heartbeat::subscription_data::SubscriptionData;
62
63    #[test]
64    fn consumer_data_default_values() {
65        let consumer_data = ConsumerData::default();
66        assert_eq!(consumer_data.group_name, CheetahString::new());
67        assert_eq!(consumer_data.consume_type, ConsumeType::default());
68        assert_eq!(consumer_data.message_model, MessageModel::default());
69        assert_eq!(consumer_data.consume_from_where, ConsumeFromWhere::default());
70        assert!(consumer_data.subscription_data_set.is_empty());
71        assert!(!consumer_data.unit_mode);
72    }
73
74    #[test]
75    fn consumer_data_equality() {
76        let mut subscription_data_set = HashSet::new();
77        subscription_data_set.insert(SubscriptionData::default());
78
79        let consumer_data1 = ConsumerData {
80            group_name: CheetahString::from("group1"),
81            consume_type: ConsumeType::default(),
82            message_model: MessageModel::default(),
83            consume_from_where: ConsumeFromWhere::default(),
84            subscription_data_set: subscription_data_set.clone(),
85            unit_mode: false,
86        };
87
88        let consumer_data2 = ConsumerData {
89            group_name: CheetahString::from("group1"),
90            consume_type: ConsumeType::default(),
91            message_model: MessageModel::default(),
92            consume_from_where: ConsumeFromWhere::default(),
93            subscription_data_set,
94            unit_mode: false,
95        };
96
97        assert_eq!(consumer_data1, consumer_data2);
98    }
99
100    #[test]
101    fn consumer_data_inequality() {
102        let consumer_data1 = ConsumerData {
103            group_name: CheetahString::from("group1"),
104            consume_type: ConsumeType::default(),
105            message_model: MessageModel::default(),
106            consume_from_where: ConsumeFromWhere::default(),
107            subscription_data_set: HashSet::new(),
108            unit_mode: false,
109        };
110
111        let consumer_data2 = ConsumerData {
112            group_name: CheetahString::from("group2"),
113            consume_type: ConsumeType::default(),
114            message_model: MessageModel::default(),
115            consume_from_where: ConsumeFromWhere::default(),
116            subscription_data_set: HashSet::new(),
117            unit_mode: false,
118        };
119
120        assert_ne!(consumer_data1, consumer_data2);
121    }
122
123    #[test]
124    fn consumer_data_hash() {
125        use std::collections::hash_map::DefaultHasher;
126        use std::hash::Hasher;
127
128        let consumer_data = ConsumerData {
129            group_name: CheetahString::from("group1"),
130            consume_type: ConsumeType::default(),
131            message_model: MessageModel::default(),
132            consume_from_where: ConsumeFromWhere::default(),
133            subscription_data_set: ["topic-c", "topic-a", "topic-b"]
134                .into_iter()
135                .map(|topic| SubscriptionData {
136                    topic: topic.into(),
137                    ..Default::default()
138                })
139                .collect(),
140            unit_mode: false,
141        };
142
143        let mut hasher = DefaultHasher::new();
144        consumer_data.hash(&mut hasher);
145        let expected = hasher.finish();
146
147        for _ in 0..32 {
148            let equal_consumer = ConsumerData {
149                subscription_data_set: consumer_data.subscription_data_set.iter().cloned().collect(),
150                ..consumer_data.clone()
151            };
152            let mut hasher = DefaultHasher::new();
153            equal_consumer.hash(&mut hasher);
154
155            assert_eq!(consumer_data, equal_consumer);
156            assert_eq!(expected, hasher.finish());
157        }
158    }
159}
rocketmq_protocol/protocol/heartbeat/consumer_data.rs

rocketmq_protocol/protocol/heartbeat/
consumer_data.rs
