rocketmq_protocol/protocol/route/
topic_route_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::BTreeMap;
16use std::collections::HashMap;
17
18use cheetah_string::CheetahString;
19use serde::Deserialize;
20use serde::Serialize;
21
22use crate::protocol::route::route_data_view::BrokerData;
23use crate::protocol::route::route_data_view::QueueData;
24use crate::protocol::static_topic::topic_queue_mapping_info::TopicQueueMappingInfo;
25use crate::protocol::RemotingDeserializable;
26use crate::protocol::RemotingSerializable;
27
28#[derive(Debug, Serialize, Deserialize, Clone, Default, Eq, PartialEq)]
29pub struct TopicRouteData {
30    #[serde(rename = "orderTopicConf")]
31    pub order_topic_conf: Option<CheetahString>,
32    #[serde(rename = "queueDatas")]
33    pub queue_datas: Vec<QueueData>,
34    #[serde(rename = "brokerDatas")]
35    pub broker_datas: Vec<BrokerData>,
36    #[serde(rename = "filterServerTable")]
37    pub filter_server_table: HashMap<CheetahString, Vec<CheetahString>>,
38    #[serde(rename = "topicQueueMappingInfo", alias = "topicQueueMappingByBroker")]
39    pub topic_queue_mapping_by_broker: Option<HashMap<CheetahString, TopicQueueMappingInfo>>,
40}
41
42impl TopicRouteData {
43    pub fn topic_route_data_changed(&self, old_data: Option<&TopicRouteData>) -> bool {
44        let Some(old_data) = old_data else { return true };
45        let mut now = TopicRouteData::from_existing(self);
46        let mut old = TopicRouteData::from_existing(old_data);
47        now.queue_datas.sort();
48        now.broker_datas.sort();
49        old.queue_datas.sort();
50        old.broker_datas.sort();
51        now != old
52    }
53
54    pub fn new() -> Self {
55        TopicRouteData {
56            order_topic_conf: None,
57            queue_datas: Vec::new(),
58            broker_datas: Vec::new(),
59            filter_server_table: HashMap::new(),
60            topic_queue_mapping_by_broker: None,
61        }
62    }
63
64    pub fn from_existing(topic_route_data: &TopicRouteData) -> Self {
65        TopicRouteData {
66            order_topic_conf: topic_route_data.order_topic_conf.clone(),
67            queue_datas: topic_route_data.queue_datas.clone(),
68            broker_datas: topic_route_data.broker_datas.clone(),
69            filter_server_table: topic_route_data.filter_server_table.clone(),
70            topic_queue_mapping_by_broker: topic_route_data.topic_queue_mapping_by_broker.clone(),
71        }
72    }
73
74    /// Encode route data using the Java `acceptStandardJsonOnly` semantics.
75    ///
76    /// The observable difference from the legacy path is deterministic map key ordering,
77    /// which mirrors Java's `MapSortField` behavior for route responses.
78    pub fn encode_standard_json(&self) -> rocketmq_error::Result<Vec<u8>> {
79        StandardJsonTopicRouteData::from(self).encode()
80    }
81
82    pub fn decode(bytes: &[u8]) -> rocketmq_error::Result<Self> {
83        match <Self as RemotingDeserializable>::decode(bytes) {
84            Ok(route_data) => Ok(route_data),
85            Err(error) => {
86                let Ok(raw_body) = std::str::from_utf8(bytes) else {
87                    return Err(error);
88                };
89                let Some(normalized_body) = quote_unquoted_numeric_object_keys(raw_body) else {
90                    return Err(error);
91                };
92                <Self as RemotingDeserializable>::decode_str(&normalized_body)
93            }
94        }
95    }
96}
97
98fn quote_unquoted_numeric_object_keys(input: &str) -> Option<String> {
99    let bytes = input.as_bytes();
100    let mut output = String::with_capacity(input.len() + 8);
101    let mut last_copied = 0;
102    let mut index = 0;
103    let mut in_string = false;
104    let mut escaped = false;
105    let mut changed = false;
106
107    while index < bytes.len() {
108        let byte = bytes[index];
109        if in_string {
110            if escaped {
111                escaped = false;
112            } else if byte == b'\\' {
113                escaped = true;
114            } else if byte == b'"' {
115                in_string = false;
116            }
117            index += 1;
118            continue;
119        }
120
121        if byte == b'"' {
122            in_string = true;
123            index += 1;
124            continue;
125        }
126
127        if byte == b'{' || byte == b',' {
128            let mut key_start = index + 1;
129            while key_start < bytes.len() && bytes[key_start].is_ascii_whitespace() {
130                key_start += 1;
131            }
132
133            let mut digit_start = key_start;
134            if digit_start < bytes.len() && bytes[digit_start] == b'-' {
135                digit_start += 1;
136            }
137
138            if digit_start < bytes.len() && bytes[digit_start].is_ascii_digit() {
139                let mut digit_end = digit_start + 1;
140                while digit_end < bytes.len() && bytes[digit_end].is_ascii_digit() {
141                    digit_end += 1;
142                }
143
144                let mut colon_index = digit_end;
145                while colon_index < bytes.len() && bytes[colon_index].is_ascii_whitespace() {
146                    colon_index += 1;
147                }
148
149                if colon_index < bytes.len() && bytes[colon_index] == b':' {
150                    output.push_str(&input[last_copied..key_start]);
151                    output.push('"');
152                    output.push_str(&input[key_start..digit_end]);
153                    output.push('"');
154                    last_copied = digit_end;
155                    changed = true;
156                    index = digit_end;
157                    continue;
158                }
159            }
160        }
161
162        index += 1;
163    }
164
165    if changed {
166        output.push_str(&input[last_copied..]);
167        Some(output)
168    } else {
169        None
170    }
171}
172
173#[derive(Serialize)]
174struct StandardJsonTopicRouteData<'a> {
175    #[serde(rename = "orderTopicConf")]
176    order_topic_conf: Option<&'a CheetahString>,
177    #[serde(rename = "queueDatas")]
178    queue_datas: &'a [QueueData],
179    #[serde(rename = "brokerDatas")]
180    broker_datas: Vec<StandardJsonBrokerData<'a>>,
181    #[serde(rename = "filterServerTable")]
182    filter_server_table: BTreeMap<String, &'a Vec<CheetahString>>,
183    #[serde(rename = "topicQueueMappingInfo")]
184    topic_queue_mapping_by_broker: Option<BTreeMap<String, StandardJsonTopicQueueMappingInfo<'a>>>,
185}
186
187#[derive(Serialize)]
188struct StandardJsonBrokerData<'a> {
189    cluster: &'a str,
190    #[serde(rename = "brokerName")]
191    broker_name: &'a CheetahString,
192    #[serde(rename = "brokerAddrs")]
193    broker_addrs: BTreeMap<String, &'a CheetahString>,
194    #[serde(rename = "zoneName")]
195    zone_name: Option<&'a CheetahString>,
196    #[serde(rename = "enableActingMaster")]
197    enable_acting_master: bool,
198}
199
200#[derive(Serialize)]
201struct StandardJsonTopicQueueMappingInfo<'a> {
202    topic: Option<&'a CheetahString>,
203    scope: Option<&'a CheetahString>,
204    #[serde(rename = "totalQueues")]
205    total_queues: i32,
206    bname: Option<&'a CheetahString>,
207    epoch: i64,
208    dirty: bool,
209    #[serde(rename = "currIdMap")]
210    curr_id_map: Option<BTreeMap<String, i32>>,
211}
212
213impl<'a> From<&'a TopicRouteData> for StandardJsonTopicRouteData<'a> {
214    fn from(value: &'a TopicRouteData) -> Self {
215        Self {
216            order_topic_conf: value.order_topic_conf.as_ref(),
217            queue_datas: &value.queue_datas,
218            broker_datas: value.broker_datas.iter().map(StandardJsonBrokerData::from).collect(),
219            filter_server_table: value
220                .filter_server_table
221                .iter()
222                .map(|(broker_addr, filter_servers)| (broker_addr.to_string(), filter_servers))
223                .collect(),
224            topic_queue_mapping_by_broker: value.topic_queue_mapping_by_broker.as_ref().map(|mapping| {
225                mapping
226                    .iter()
227                    .map(|(broker_name, info)| (broker_name.to_string(), StandardJsonTopicQueueMappingInfo::from(info)))
228                    .collect()
229            }),
230        }
231    }
232}
233
234impl<'a> From<&'a BrokerData> for StandardJsonBrokerData<'a> {
235    fn from(value: &'a BrokerData) -> Self {
236        Self {
237            cluster: value.cluster(),
238            broker_name: value.broker_name(),
239            broker_addrs: value
240                .broker_addrs()
241                .iter()
242                .map(|(broker_id, broker_addr)| (broker_id.to_string(), broker_addr))
243                .collect(),
244            zone_name: value.zone_name(),
245            enable_acting_master: value.enable_acting_master(),
246        }
247    }
248}
249
250impl<'a> From<&'a TopicQueueMappingInfo> for StandardJsonTopicQueueMappingInfo<'a> {
251    fn from(value: &'a TopicQueueMappingInfo) -> Self {
252        Self {
253            topic: value.topic.as_ref(),
254            scope: value.scope.as_ref(),
255            total_queues: value.total_queues,
256            bname: value.bname.as_ref(),
257            epoch: value.epoch,
258            dirty: value.dirty,
259            curr_id_map: value.curr_id_map.as_ref().map(|mapping| {
260                mapping
261                    .iter()
262                    .map(|(queue_id, mapped_id)| (queue_id.to_string(), *mapped_id))
263                    .collect()
264            }),
265        }
266    }
267}
268
269#[cfg(test)]
270mod tests {
271    use std::collections::HashMap;
272
273    use serde_json;
274
275    use super::*;
276    use crate::protocol::route::route_data_view::BrokerData;
277
278    #[test]
279    fn serde_preserves_topic_route_data() {
280        let mut filter_server_table = HashMap::new();
281        filter_server_table.insert(CheetahString::from("key"), vec![CheetahString::from("value")]);
282        let mut topic_queue_mapping_by_broker = HashMap::new();
283        topic_queue_mapping_by_broker.insert(CheetahString::from("broker"), TopicQueueMappingInfo::default());
284        let topic_route_data = TopicRouteData {
285            order_topic_conf: Some(CheetahString::from("conf")),
286            queue_datas: vec![QueueData::default()],
287            broker_datas: vec![BrokerData::default()],
288            filter_server_table,
289            topic_queue_mapping_by_broker: Some(topic_queue_mapping_by_broker),
290        };
291        let serialized = serde_json::to_string(&topic_route_data).unwrap();
292        assert!(serialized.contains("\"orderTopicConf\":\"conf\""));
293        assert!(serialized.contains("\"queueDatas\":["));
294        assert!(serialized.contains("\"brokerDatas\":["));
295        assert!(serialized.contains("\"filterServerTable\":{\"key\":[\"value\"]}"));
296        assert!(serialized.contains("\"topicQueueMappingInfo\":{\"broker\":{"));
297
298        let decoded = TopicRouteData::decode(serialized.as_bytes()).unwrap();
299        assert_eq!(decoded, topic_route_data);
300    }
301
302    #[test]
303    fn decode_java_fastjson_topic_route_data() {
304        let java_body = br#"{"brokerDatas":[{"brokerAddrs":{0:"127.0.0.1:10911"},"brokerName":"interopBroker","cluster":"DefaultCluster","enableActingMaster":false}],"filterServerTable":{},"queueDatas":[{"brokerName":"interopBroker","perm":6,"readQueueNums":4,"topicSysFlag":0,"writeQueueNums":4}],"topicQueueMappingByBroker":{"interopBroker":{"topic":"TopicA","scope":"ScopeA","totalQueues":4,"bname":"interopBroker","epoch":1,"dirty":false,"currIdMap":{0:0}}}}"#;
305
306        let decoded = TopicRouteData::decode(java_body).expect("decode Java fastjson route data");
307
308        assert_eq!(decoded.queue_datas.len(), 1);
309        assert_eq!(decoded.queue_datas[0].broker_name, CheetahString::from("interopBroker"));
310        assert_eq!(decoded.broker_datas.len(), 1);
311        assert_eq!(
312            decoded.broker_datas[0].broker_addrs().get(&0),
313            Some(&CheetahString::from("127.0.0.1:10911"))
314        );
315        let mapping = decoded
316            .topic_queue_mapping_by_broker
317            .as_ref()
318            .and_then(|items| items.get(&CheetahString::from("interopBroker")))
319            .expect("topicQueueMappingByBroker alias should decode");
320        assert_eq!(mapping.curr_id_map.as_ref().and_then(|items| items.get(&0)), Some(&0));
321    }
322
323    #[test]
324    fn encode_standard_json_sorts_nested_map_keys() {
325        let mut broker_addrs = HashMap::new();
326        broker_addrs.insert(10, CheetahString::from("10.0.0.10:10911"));
327        broker_addrs.insert(2, CheetahString::from("10.0.0.2:10911"));
328
329        let mut filter_server_table = HashMap::new();
330        filter_server_table.insert(CheetahString::from("z-broker"), vec![CheetahString::from("fs-z")]);
331        filter_server_table.insert(CheetahString::from("a-broker"), vec![CheetahString::from("fs-a")]);
332
333        let mut curr_id_map = HashMap::new();
334        curr_id_map.insert(8, 80);
335        curr_id_map.insert(1, 10);
336
337        let mut topic_queue_mapping_by_broker = HashMap::new();
338        topic_queue_mapping_by_broker.insert(
339            CheetahString::from("mapping-broker"),
340            TopicQueueMappingInfo {
341                topic: Some(CheetahString::from("TestTopic")),
342                scope: Some(CheetahString::from("scope-a")),
343                total_queues: 8,
344                bname: Some(CheetahString::from("broker-a")),
345                epoch: 3,
346                dirty: false,
347                curr_id_map: Some(curr_id_map),
348            },
349        );
350
351        let topic_route_data = TopicRouteData {
352            order_topic_conf: Some(CheetahString::from("order-conf")),
353            queue_datas: vec![QueueData::new(CheetahString::from("broker-a"), 4, 4, 6, 0)],
354            broker_datas: vec![BrokerData::new(
355                CheetahString::from("cluster-a"),
356                CheetahString::from("broker-a"),
357                broker_addrs,
358                Some(CheetahString::from("zone-a")),
359            )],
360            filter_server_table,
361            topic_queue_mapping_by_broker: Some(topic_queue_mapping_by_broker),
362        };
363
364        let encoded = String::from_utf8(topic_route_data.encode_standard_json().unwrap()).unwrap();
365        let broker_addrs_index = encoded.find("\"brokerAddrs\":{\"10\"").unwrap();
366        let broker_addrs_second_index = encoded.find("\"2\":\"10.0.0.2:10911\"").unwrap();
367        let filter_server_index = encoded.find("\"filterServerTable\":{\"a-broker\"").unwrap();
368        let filter_server_second_index = encoded.find("\"z-broker\":[\"fs-z\"]").unwrap();
369        let curr_id_map_index = encoded.find("\"currIdMap\":{\"1\":10").unwrap();
370        let curr_id_map_second_index = encoded.find("\"8\":80").unwrap();
371
372        assert!(broker_addrs_index < broker_addrs_second_index);
373        assert!(filter_server_index < filter_server_second_index);
374        assert!(curr_id_map_index < curr_id_map_second_index);
375    }
376
377    #[test]
378    fn topic_route_data_changed_ignores_broker_and_queue_order() {
379        let queue = |name: &str, count| QueueData::new(name.into(), count, count, 6, 0);
380        let broker = |name: &str| BrokerData::new("cluster".into(), name.into(), HashMap::new(), None);
381        let route_data = TopicRouteData {
382            queue_datas: vec![queue("broker-b", 8), queue("broker-a", 4)],
383            broker_datas: vec![broker("broker-b"), broker("broker-a")],
384            ..Default::default()
385        };
386
387        assert!(route_data.topic_route_data_changed(None));
388        assert!(!route_data.topic_route_data_changed(Some(&route_data)));
389
390        let mut reordered = TopicRouteData::from_existing(&route_data);
391        reordered.queue_datas.reverse();
392        reordered.broker_datas.reverse();
393        assert!(!route_data.topic_route_data_changed(Some(&reordered)));
394
395        let mut changed = TopicRouteData::from_existing(&route_data);
396        changed.order_topic_conf = Some(CheetahString::from("changed"));
397        assert!(route_data.topic_route_data_changed(Some(&changed)));
398    }
399
400    #[test]
401    fn numeric_key_normalization_handles_signed_nested_keys_and_whitespace() {
402        let input = r#"{ 12 : { -3 : "value", 4: [1,-2,3] }, -5 : 6 }"#;
403        let normalized = quote_unquoted_numeric_object_keys(input).unwrap();
404        assert_eq!(
405            serde_json::from_str::<serde_json::Value>(&normalized).unwrap(),
406            serde_json::json!({
407                "12": {"-3": "value", "4": [1,-2,3]}, "-5": 6
408            })
409        );
410    }
411
412    #[test]
413    fn numeric_key_normalization_leaves_strings_and_numeric_values_unchanged() {
414        for input in [
415            r#"{"12": 3, "-4": [5,-6]}"#,
416            r#"{"text":"12: and \"-3:\" and \\4:","value":17}"#,
417        ] {
418            assert_eq!(quote_unquoted_numeric_object_keys(input), None);
419        }
420        let input = r#"{12:"-3: and \"4:\" and \\5:", "values":[6,-7]}"#;
421        let normalized = quote_unquoted_numeric_object_keys(input).unwrap();
422        let expected = r#"{"12":"-3: and \"4:\" and \\5:", "values":[6,-7]}"#;
423        assert_eq!(normalized, expected);
424    }
425
426    #[test]
427    fn decode_java_fastjson_accepts_negative_mapping_queue_ids() {
428        let body = br#"{"brokerDatas":[],"filterServerTable":{},"queueDatas":[],"topicQueueMappingByBroker":{"broker":{"topic":"topic","scope":"scope","totalQueues":1,"bname":"broker","epoch":1,"dirty":false,"currIdMap":{-1:2}}}}"#;
429        let decoded = TopicRouteData::decode(body).unwrap();
430        let mapping = &decoded.topic_queue_mapping_by_broker.as_ref().unwrap()["broker"];
431        assert_eq!(mapping.curr_id_map, Some(HashMap::from([(-1, 2)])));
432    }
433}
rocketmq_protocol/protocol/route/topic_route_data.rs

rocketmq_protocol/protocol/route/
topic_route_data.rs
