rocketmq_remoting/protocol/heartbeat/
consumer_data.rs

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

rocketmq_remoting/protocol/heartbeat/
consumer_data.rs
