rocketmq_protocol/rpc/
rpc_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
20#[derive(Clone, Debug, Serialize, Deserialize, Default, RequestHeaderCodec)]
21#[header(
22    type_id = "rocketmq_protocol::rpc::rpc_request_header::RpcRequestHeader",
23    java_class = "org.apache.rocketmq.remoting.rpc.RpcRequestHeader"
24)]
25pub struct RpcRequestHeader {
26    // the namespace name
27    #[serde(rename = "ns", alias = "namespace")]
28    #[header(key = "ns", alias = "namespace", alias_conflict = "prefer_canonical")]
29    pub namespace: Option<CheetahString>,
30    // if the data has been namespaced
31    #[serde(rename = "nsd", alias = "namespaced")]
32    #[header(key = "nsd", alias = "namespaced", alias_conflict = "prefer_canonical")]
33    pub namespaced: Option<bool>,
34    // the abstract remote addr name, usually the physical broker name
35    #[serde(rename = "bname", alias = "brokerName")]
36    #[header(key = "bname", alias = "brokerName", alias_conflict = "prefer_canonical")]
37    pub broker_name: Option<CheetahString>,
38    // oneway
39    #[serde(rename = "oway", alias = "oneway")]
40    #[header(key = "oway", alias = "oneway", alias_conflict = "prefer_canonical")]
41    pub oneway: Option<bool>,
42}
43
44impl RpcRequestHeader {
45    pub fn new(
46        namespace: Option<CheetahString>,
47        namespaced: Option<bool>,
48        broker_name: Option<CheetahString>,
49        oneway: Option<bool>,
50    ) -> Self {
51        Self {
52            namespace,
53            namespaced,
54            broker_name,
55            oneway,
56        }
57    }
58}
59
60#[cfg(test)]
61mod tests {
62    use std::collections::HashMap;
63
64    use super::*;
65    use crate::protocol::command_custom_header::CommandCustomHeader;
66    use crate::protocol::command_custom_header::FromMap;
67
68    #[test]
69    fn encodes_java_canonical_short_keys() {
70        let header = RpcRequestHeader::new(Some("tenant".into()), Some(true), Some("broker-a".into()), Some(false));
71
72        let map = header.to_map().unwrap();
73        assert_eq!(map.get("ns"), Some(&"tenant".into()));
74        assert_eq!(map.get("nsd"), Some(&"true".into()));
75        assert_eq!(map.get("bname"), Some(&"broker-a".into()));
76        assert_eq!(map.get("oway"), Some(&"false".into()));
77        for legacy_key in ["namespace", "namespaced", "brokerName", "oneway"] {
78            assert!(!map.contains_key(legacy_key));
79        }
80    }
81
82    #[test]
83    fn decodes_legacy_long_keys() {
84        let map = HashMap::from([
85            ("namespace".into(), "tenant".into()),
86            ("namespaced".into(), "true".into()),
87            ("brokerName".into(), "broker-a".into()),
88            ("oneway".into(), "false".into()),
89        ]);
90
91        let header = <RpcRequestHeader as FromMap>::from(&map).unwrap();
92        assert_eq!(header.namespace.as_deref(), Some("tenant"));
93        assert_eq!(header.namespaced, Some(true));
94        assert_eq!(header.broker_name.as_deref(), Some("broker-a"));
95        assert_eq!(header.oneway, Some(false));
96    }
97
98    #[test]
99    fn canonical_keys_win_alias_conflicts_independent_of_insertion_order() {
100        let entries = [
101            ("ns", "canonical-ns"),
102            ("namespace", "legacy-ns"),
103            ("nsd", "true"),
104            ("namespaced", "false"),
105            ("bname", "canonical-broker"),
106            ("brokerName", "legacy-broker"),
107            ("oway", "false"),
108            ("oneway", "true"),
109        ];
110
111        for reverse in [false, true] {
112            let mut map = HashMap::new();
113            let ordered: Vec<_> = if reverse {
114                entries.iter().rev().copied().collect()
115            } else {
116                entries.to_vec()
117            };
118            for (key, value) in ordered {
119                map.insert(key.into(), value.into());
120            }
121
122            let header = <RpcRequestHeader as FromMap>::from(&map).unwrap();
123            assert_eq!(header.namespace.as_deref(), Some("canonical-ns"));
124            assert_eq!(header.namespaced, Some(true));
125            assert_eq!(header.broker_name.as_deref(), Some("canonical-broker"));
126            assert_eq!(header.oneway, Some(false));
127        }
128    }
129}
rocketmq_protocol/rpc/rpc_request_header.rs

rocketmq_protocol/rpc/
rpc_request_header.rs
