rocketmq_protocol/protocol/header/namesrv/
broker_request.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 rocketmq_macros::RequestHeaderCodec;
16use std::fmt::Display;
17
18use cheetah_string::CheetahString;
19use serde::Deserialize;
20use serde::Serialize;
21
22/// Maximum Controller heartbeat lease accepted from a Broker.
23///
24/// Brokers continuously refresh this lease. One day preserves a generous
25/// operational window while preventing an untrusted value from pinning
26/// Controller session-fencing capacity indefinitely.
27pub const MAX_BROKER_HEARTBEAT_TIMEOUT_MILLIS: u64 = 24 * 60 * 60 * 1_000;
28
29#[derive(Debug, Clone, Deserialize, Serialize, Default, RequestHeaderCodec)]
30#[serde(rename_all = "camelCase")]
31#[header(
32    type_id = "rocketmq_protocol::protocol::header::namesrv::broker_request::UnRegisterBrokerRequestHeader",
33    java_class = "org.apache.rocketmq.remoting.protocol.header.namesrv.UnRegisterBrokerRequestHeader"
34)]
35pub struct UnRegisterBrokerRequestHeader {
36    #[header(required)]
37    pub broker_name: CheetahString,
38
39    #[header(required)]
40    pub broker_addr: CheetahString,
41
42    #[header(required)]
43    pub cluster_name: CheetahString,
44
45    #[header(required, range = "i64")]
46    pub broker_id: u64,
47}
48
49impl UnRegisterBrokerRequestHeader {
50    pub fn new(
51        broker_name: impl Into<CheetahString>,
52        broker_addr: impl Into<CheetahString>,
53        cluster_name: impl Into<CheetahString>,
54        broker_id: u64,
55    ) -> Self {
56        Self {
57            broker_name: broker_name.into(),
58            broker_addr: broker_addr.into(),
59            cluster_name: cluster_name.into(),
60            broker_id,
61        }
62    }
63}
64
65impl Display for UnRegisterBrokerRequestHeader {
66    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
67        write!(
68            f,
69            "UnRegisterBrokerRequestHeader {{ brokerName: {}, brokerAddr: {}, clusterName: {}, brokerId: {} }}",
70            self.broker_name, self.broker_addr, self.cluster_name, self.broker_id
71        )
72    }
73}
74
75#[derive(Debug, Clone, Deserialize, Serialize, Default, RequestHeaderCodec)]
76#[serde(rename_all = "camelCase")]
77#[header(
78    type_id = "rocketmq_protocol::protocol::header::namesrv::broker_request::BrokerHeartbeatRequestHeader",
79    java_class = "org.apache.rocketmq.remoting.protocol.header.namesrv.BrokerHeartbeatRequestHeader"
80)]
81pub struct BrokerHeartbeatRequestHeader {
82    #[header(required)]
83    pub cluster_name: CheetahString,
84
85    #[header(required)]
86    pub broker_addr: CheetahString,
87
88    #[header(required)]
89    pub broker_name: CheetahString,
90    pub broker_id: Option<i64>,
91    pub epoch: Option<i32>,
92    pub max_offset: Option<i64>,
93    pub confirm_offset: Option<i64>,
94    /// Rust-native Controller extension; Java peers and NameServers ignore it.
95    pub store_ready: Option<bool>,
96    pub heartbeat_timeout_mills: Option<i64>,
97    pub election_priority: Option<i32>,
98}
99
100#[derive(Debug, Clone, Deserialize, Serialize, Default, RequestHeaderCodec)]
101#[serde(rename_all = "camelCase")]
102#[header(
103    type_id = "rocketmq_protocol::protocol::header::namesrv::broker_request::GetBrokerMemberGroupRequestHeader",
104    java_class = "org.apache.rocketmq.remoting.protocol.header.GetBrokerMemberGroupRequestHeader"
105)]
106pub struct GetBrokerMemberGroupRequestHeader {
107    #[header(required)]
108    pub cluster_name: CheetahString,
109    #[header(required)]
110    pub broker_name: CheetahString,
111}
112
113impl GetBrokerMemberGroupRequestHeader {
114    pub fn new(cluster_name: impl Into<CheetahString>, broker_name: impl Into<CheetahString>) -> Self {
115        Self {
116            cluster_name: cluster_name.into(),
117            broker_name: broker_name.into(),
118        }
119    }
120}
121
122#[cfg(test)]
123mod tests {
124    use super::*;
125
126    #[test]
127    fn unregister_broker_request_header_display() {
128        let header = UnRegisterBrokerRequestHeader::new("name", "addr", "cluster", 1);
129        let display = format!("{}", header);
130        let expected =
131            "UnRegisterBrokerRequestHeader { brokerName: name, brokerAddr: addr, clusterName: cluster, brokerId: 1 }";
132        assert_eq!(display, expected);
133    }
134
135    #[test]
136    fn get_broker_member_group_request_uses_java_wire_names() {
137        let header = GetBrokerMemberGroupRequestHeader::new("cluster-a", "broker-a");
138
139        assert_eq!(
140            serde_json::to_string(&header).unwrap(),
141            r#"{"clusterName":"cluster-a","brokerName":"broker-a"}"#
142        );
143    }
144}
rocketmq_protocol/protocol/header/namesrv/broker_request.rs

rocketmq_protocol/protocol/header/namesrv/
broker_request.rs
