rocketmq_protocol/protocol/header/controller/
clean_broker_data_request_header.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 cheetah_string::CheetahString;
16use rocketmq_macros::RequestHeaderCodec;
17use serde::Deserialize;
18use serde::Serialize;
19
20fn default_invoke_time() -> u64 {
21    rocketmq_model::time::current_millis()
22}
23
24#[derive(Clone, Debug, Serialize, Deserialize, RequestHeaderCodec)]
25#[header(
26    type_id = "rocketmq_protocol::protocol::header::controller::clean_broker_data_request_header::CleanBrokerDataRequestHeader",
27    java_class = "org.apache.rocketmq.remoting.protocol.header.controller.admin.CleanControllerBrokerDataRequestHeader",
28    fast
29)]
30#[serde(rename_all = "camelCase")]
31pub struct CleanBrokerDataRequestHeader {
32    pub cluster_name: Option<CheetahString>,
33    #[header(required)]
34    pub broker_name: CheetahString,
35    pub broker_controller_ids_to_clean: Option<CheetahString>,
36    #[serde(rename = "isCleanLivingBroker", alias = "cleanLivingBroker")]
37    #[header(
38        key = "isCleanLivingBroker",
39        alias = "cleanLivingBroker",
40        alias_conflict = "prefer_canonical",
41        default,
42        default_semantic = "literal:false"
43    )]
44    pub clean_living_broker: bool,
45    #[serde(default = "default_invoke_time")]
46    #[header(
47        default_with = "default_invoke_time",
48        default_semantic = "dynamic:current_time_millis",
49        range = "i64"
50    )]
51    pub invoke_time: u64,
52}
53
54pub type CleanControllerBrokerDataRequestHeader = CleanBrokerDataRequestHeader;
55
56impl Default for CleanBrokerDataRequestHeader {
57    fn default() -> Self {
58        Self {
59            cluster_name: None,
60            broker_name: CheetahString::new(),
61            broker_controller_ids_to_clean: None,
62            clean_living_broker: false,
63            invoke_time: default_invoke_time(),
64        }
65    }
66}
67
68#[cfg(test)]
69mod tests {
70    use std::collections::HashMap;
71
72    use super::*;
73    use crate::protocol::command_custom_header::CommandCustomHeader;
74    use crate::protocol::command_custom_header::FromMap;
75
76    #[test]
77    fn clean_broker_data_request_header_serializes_correctly() {
78        let header = CleanBrokerDataRequestHeader {
79            cluster_name: Some(CheetahString::from_static_str("test_cluster")),
80            broker_name: CheetahString::from_static_str("test_broker"),
81            broker_controller_ids_to_clean: Some(CheetahString::from_static_str("1;2")),
82            clean_living_broker: true,
83            invoke_time: 1234567890,
84        };
85
86        let map = header.to_map().unwrap();
87        assert_eq!(
88            map.get(&CheetahString::from_static_str("clusterName")).unwrap(),
89            "test_cluster"
90        );
91        assert_eq!(
92            map.get(&CheetahString::from_static_str("brokerName")).unwrap(),
93            "test_broker"
94        );
95        assert_eq!(
96            map.get(&CheetahString::from_static_str("brokerControllerIdsToClean"))
97                .unwrap(),
98            "1;2"
99        );
100        assert_eq!(
101            map.get(&CheetahString::from_static_str("isCleanLivingBroker")).unwrap(),
102            "true"
103        );
104    }
105
106    #[test]
107    fn clean_broker_data_request_header_deserializes_correctly() {
108        let mut map = HashMap::new();
109        map.insert(
110            CheetahString::from_static_str("clusterName"),
111            CheetahString::from_static_str("test_cluster"),
112        );
113        map.insert(
114            CheetahString::from_static_str("brokerName"),
115            CheetahString::from_static_str("test_broker"),
116        );
117        map.insert(
118            CheetahString::from_static_str("brokerControllerIdsToClean"),
119            CheetahString::from_static_str("1;2"),
120        );
121        map.insert(
122            CheetahString::from_static_str("isCleanLivingBroker"),
123            CheetahString::from_static_str("false"),
124        );
125        map.insert(
126            CheetahString::from_static_str("invokeTime"),
127            CheetahString::from_static_str("1234567890"),
128        );
129
130        let header = <CleanBrokerDataRequestHeader as FromMap>::from(&map).unwrap();
131        assert_eq!(
132            header.cluster_name,
133            Some(CheetahString::from_static_str("test_cluster"))
134        );
135        assert_eq!(header.broker_name, CheetahString::from_static_str("test_broker"));
136        assert_eq!(
137            header.broker_controller_ids_to_clean,
138            Some(CheetahString::from_static_str("1;2"))
139        );
140        assert!(!header.clean_living_broker);
141        assert_eq!(header.invoke_time, 1234567890);
142    }
143
144    #[test]
145    fn missing_invoke_time_uses_the_java_dynamic_default() {
146        let before = rocketmq_model::time::current_millis();
147        let map = HashMap::from([(
148            CheetahString::from_static_str("brokerName"),
149            CheetahString::from_static_str("broker-a"),
150        )]);
151
152        let header = <CleanBrokerDataRequestHeader as FromMap>::from(&map).unwrap();
153        let after = rocketmq_model::time::current_millis();
154
155        assert!((before..=after).contains(&header.invoke_time));
156        assert!(!header.clean_living_broker);
157    }
158
159    #[test]
160    fn clean_broker_data_request_header_default() {
161        let before = rocketmq_model::time::current_millis();
162        let header = CleanBrokerDataRequestHeader::default();
163        let after = rocketmq_model::time::current_millis();
164
165        assert!(header.cluster_name.is_none());
166        assert_eq!(header.broker_name, "");
167        assert!(header.broker_controller_ids_to_clean.is_none());
168        assert!(!header.clean_living_broker);
169        assert!((before..=after).contains(&header.invoke_time));
170    }
171
172    #[test]
173    fn canonical_clean_living_key_wins_the_legacy_alias() {
174        let map = HashMap::from([
175            (
176                CheetahString::from_static_str("brokerName"),
177                CheetahString::from_static_str("broker-a"),
178            ),
179            (
180                CheetahString::from_static_str("isCleanLivingBroker"),
181                CheetahString::from_static_str("true"),
182            ),
183            (
184                CheetahString::from_static_str("cleanLivingBroker"),
185                CheetahString::from_static_str("false"),
186            ),
187        ]);
188
189        let header = <CleanBrokerDataRequestHeader as FromMap>::from(&map).unwrap();
190
191        assert!(header.clean_living_broker);
192    }
193}
rocketmq_protocol/protocol/header/controller/clean_broker_data_request_header.rs

rocketmq_protocol/protocol/header/controller/
clean_broker_data_request_header.rs
