rocketmq_protocol/protocol/header/namesrv/
register_broker_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
20/// Represents the header for a broker registration request.
21#[derive(Debug, Clone, Deserialize, Serialize, Default, RequestHeaderCodec)]
22#[header(
23    type_id = "rocketmq_protocol::protocol::header::namesrv::register_broker_header::RegisterBrokerRequestHeader",
24    java_class = "org.apache.rocketmq.remoting.protocol.header.namesrv.RegisterBrokerRequestHeader"
25)]
26pub struct RegisterBrokerRequestHeader {
27    /// The name of the broker.
28    #[serde(rename = "brokerName")]
29    #[header(required)]
30    pub broker_name: CheetahString,
31
32    /// The address of the broker.
33    #[serde(rename = "brokerAddr")]
34    #[header(required)]
35    pub broker_addr: CheetahString,
36
37    /// The name of the cluster to which the broker belongs.
38    #[serde(rename = "clusterName")]
39    #[header(required)]
40    pub cluster_name: CheetahString,
41
42    /// The address of the highly available (HA) remoting_server associated with the broker.
43    #[serde(rename = "haServerAddr")]
44    #[header(required)]
45    pub ha_server_addr: CheetahString,
46
47    /// The unique identifier for the broker.
48    #[serde(rename = "brokerId")]
49    #[header(required, range = "i64")]
50    pub broker_id: u64,
51
52    /// The optional heartbeat timeout in milliseconds.
53    #[serde(rename = "heartbeatTimeoutMillis")]
54    pub heartbeat_timeout_millis: Option<i64>,
55
56    /// The optional flag indicating whether acting as the master is enabled.
57    #[serde(rename = "enableActingMaster")]
58    pub enable_acting_master: Option<bool>,
59
60    /// Indicates whether the data is compressed.
61    #[header(default, default_semantic = "literal:false")]
62    pub compressed: bool,
63
64    /// The CRC32 checksum for the message body.
65    #[serde(rename = "bodyCrc32")]
66    #[header(default, default_semantic = "literal:0", range = "i32")]
67    pub body_crc32: u32,
68}
69
70impl RegisterBrokerRequestHeader {
71    /// Creates a new instance of `RegisterBrokerRequestHeader`.
72    ///
73    /// # Arguments
74    ///
75    /// * `broker_name` - The name of the broker.
76    /// * `broker_addr` - The address of the broker.
77    /// * `cluster_name` - The name of the cluster.
78    /// * `ha_server_addr` - The address of the HA remoting_server.
79    /// * `broker_id` - The unique identifier for the broker.
80    /// * `heartbeat_timeout_millis` - The optional heartbeat timeout in milliseconds.
81    /// * `enable_acting_master` - The optional flag indicating whether acting as the master is
82    ///   enabled.
83    /// * `compressed` - Indicates whether the data is compressed.
84    /// * `body_crc32` - The CRC32 checksum for the message body.
85    ///
86    /// # Returns
87    ///
88    /// A new `RegisterBrokerRequestHeader` instance.
89    pub fn new(
90        broker_name: CheetahString,
91        broker_addr: CheetahString,
92        cluster_name: CheetahString,
93        ha_server_addr: CheetahString,
94        broker_id: u64,
95        heartbeat_timeout_millis: Option<i64>,
96        enable_acting_master: Option<bool>,
97        compressed: bool,
98        body_crc32: u32,
99    ) -> Self {
100        RegisterBrokerRequestHeader {
101            broker_name,
102            broker_addr,
103            cluster_name,
104            ha_server_addr,
105            broker_id,
106            heartbeat_timeout_millis,
107            enable_acting_master,
108            compressed,
109            body_crc32,
110        }
111    }
112}
113
114#[derive(Debug, Clone, Deserialize, Serialize, Default, RequestHeaderCodec)]
115#[serde(rename_all = "camelCase")]
116#[header(
117    type_id = "rocketmq_protocol::protocol::header::namesrv::register_broker_header::RegisterBrokerResponseHeader",
118    java_class = "org.apache.rocketmq.remoting.protocol.header.namesrv.RegisterBrokerResponseHeader"
119)]
120pub struct RegisterBrokerResponseHeader {
121    pub ha_server_addr: Option<CheetahString>,
122    pub master_addr: Option<CheetahString>,
123}
124
125impl RegisterBrokerResponseHeader {
126    pub fn new(ha_server_addr: Option<CheetahString>, master_addr: Option<CheetahString>) -> Self {
127        RegisterBrokerResponseHeader {
128            ha_server_addr,
129            master_addr,
130        }
131    }
132}
133
134#[cfg(test)]
135mod tests {
136    use cheetah_string::CheetahString;
137
138    use super::*;
139
140    #[test]
141    fn register_broker_request_header_serialization() {
142        let header = RegisterBrokerRequestHeader::new(
143            CheetahString::from("broker1"),
144            CheetahString::from("127.0.0.1"),
145            CheetahString::from("cluster1"),
146            CheetahString::from("127.0.0.2"),
147            1,
148            Some(3000),
149            Some(true),
150            true,
151            12345,
152        );
153        let serialized = serde_json::to_string(&header).unwrap();
154        assert_eq!(
155            serialized,
156            r#"{"brokerName":"broker1","brokerAddr":"127.0.0.1","clusterName":"cluster1","haServerAddr":"127.0.0.2","brokerId":1,"heartbeatTimeoutMillis":3000,"enableActingMaster":true,"compressed":true,"bodyCrc32":12345}"#
157        );
158    }
159
160    #[test]
161    fn register_broker_request_header_deserialization() {
162        let json = r#"{"brokerName":"broker1","brokerAddr":"127.0.0.1","clusterName":"cluster1","haServerAddr":"127.0.0.2","brokerId":1,"heartbeatTimeoutMillis":3000,"enableActingMaster":true,"compressed":true,"bodyCrc32":12345}"#;
163        let deserialized: RegisterBrokerRequestHeader = serde_json::from_str(json).unwrap();
164        assert_eq!(deserialized.broker_name, CheetahString::from("broker1"));
165        assert_eq!(deserialized.broker_addr, CheetahString::from("127.0.0.1"));
166        assert_eq!(deserialized.cluster_name, CheetahString::from("cluster1"));
167        assert_eq!(deserialized.ha_server_addr, CheetahString::from("127.0.0.2"));
168        assert_eq!(deserialized.broker_id, 1);
169        assert_eq!(deserialized.heartbeat_timeout_millis, Some(3000));
170        assert_eq!(deserialized.enable_acting_master, Some(true));
171        assert!(deserialized.compressed);
172        assert_eq!(deserialized.body_crc32, 12345);
173    }
174
175    #[test]
176    fn register_broker_request_header_deserialization_missing_fields() {
177        let json = r#"{"brokerName":"broker1","brokerAddr":"127.0.0.1","clusterName":"cluster1","haServerAddr":"127.0.0.2","brokerId":1,"compressed":true,"bodyCrc32":12345}"#;
178        let deserialized: RegisterBrokerRequestHeader = serde_json::from_str(json).unwrap();
179        assert_eq!(deserialized.broker_name, CheetahString::from("broker1"));
180        assert_eq!(deserialized.broker_addr, CheetahString::from("127.0.0.1"));
181        assert_eq!(deserialized.cluster_name, CheetahString::from("cluster1"));
182        assert_eq!(deserialized.ha_server_addr, CheetahString::from("127.0.0.2"));
183        assert_eq!(deserialized.broker_id, 1);
184        assert_eq!(deserialized.heartbeat_timeout_millis, None);
185        assert_eq!(deserialized.enable_acting_master, None);
186        assert!(deserialized.compressed);
187        assert_eq!(deserialized.body_crc32, 12345);
188    }
189
190    #[test]
191    fn register_broker_response_header_serialization() {
192        let header = RegisterBrokerResponseHeader::new(
193            Some(CheetahString::from("127.0.0.2")),
194            Some(CheetahString::from("127.0.0.3")),
195        );
196        let serialized = serde_json::to_string(&header).unwrap();
197        assert_eq!(serialized, r#"{"haServerAddr":"127.0.0.2","masterAddr":"127.0.0.3"}"#);
198    }
199
200    #[test]
201    fn register_broker_response_header_deserialization() {
202        let json = r#"{"haServerAddr":"127.0.0.2","masterAddr":"127.0.0.3"}"#;
203        let deserialized: RegisterBrokerResponseHeader = serde_json::from_str(json).unwrap();
204        assert_eq!(deserialized.ha_server_addr, Some(CheetahString::from("127.0.0.2")));
205        assert_eq!(deserialized.master_addr, Some(CheetahString::from("127.0.0.3")));
206    }
207
208    #[test]
209    fn register_broker_response_header_deserialization_missing_fields() {
210        let json = r#"{"haServerAddr":"127.0.0.2"}"#;
211        let deserialized: RegisterBrokerResponseHeader = serde_json::from_str(json).unwrap();
212        assert_eq!(deserialized.ha_server_addr, Some(CheetahString::from("127.0.0.2")));
213        assert_eq!(deserialized.master_addr, None);
214    }
215}
rocketmq_protocol/protocol/header/namesrv/register_broker_header.rs

rocketmq_protocol/protocol/header/namesrv/
register_broker_header.rs
