rocketmq_protocol/
protocol.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::HashMap;
16use std::fmt;
17use std::fmt::Display;
18use std::fmt::Formatter;
19use std::sync::atomic::AtomicI64;
20use std::sync::atomic::Ordering;
21
22use bytes::BytesMut;
23use cheetah_string::CheetahString;
24use serde::ser::SerializeStruct;
25use serde::Deserialize;
26use serde::Serialize;
27use serde::Serializer;
28
29use self::rocketmq_serializable::RocketMQSerializable;
30
31pub mod admin;
32pub mod bodies;
33pub mod body;
34pub mod broker_sync_info;
35pub mod command_custom_header;
36pub mod encoded_frame;
37pub mod filter;
38pub mod forbidden_type;
39pub mod header;
40pub mod header_codec;
41pub mod header_facade;
42pub(crate) mod header_field_merge;
43pub mod headers;
44pub mod heartbeat;
45pub mod heartbeat_facade;
46pub mod namespace_util;
47pub mod namesrv;
48pub mod remoting_command;
49pub mod remoting_command_compat;
50pub mod remoting_command_defaults;
51pub mod request_source;
52pub mod request_type;
53pub mod rocketmq_serializable;
54pub mod route;
55pub mod route_facade;
56pub mod static_topic;
57pub mod subscription;
58pub mod topic;
59
60pub use command_custom_header::CommandCustomHeader;
61pub use command_custom_header::FromMap;
62pub use command_custom_header::HeaderEncodeCapability;
63pub use data_version_compat as data_version_facade;
64pub use encoded_frame::EncodedFrame;
65pub use encoded_frame::EncodedFrameHead;
66pub use remoting_command::RemotingCommand;
67pub use remoting_command_compat as remoting_command_facade;
68
69pub mod data_version_compat;
70
71#[derive(Debug)]
72pub struct DataVersion {
73    state_version: i64,
74    timestamp: i64,
75    counter: AtomicI64,
76}
77
78impl Serialize for DataVersion {
79    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
80    where
81        S: Serializer,
82    {
83        let mut state = serializer.serialize_struct("DataVersion", 3)?;
84        state.serialize_field("stateVersion", &self.state_version)?;
85        state.serialize_field("timestamp", &self.timestamp)?;
86        state.serialize_field("counter", &self.counter.load(Ordering::SeqCst))?;
87        state.end()
88    }
89}
90
91impl<'de> Deserialize<'de> for DataVersion {
92    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
93    where
94        D: serde::Deserializer<'de>,
95    {
96        #[derive(Deserialize)]
97        #[serde(rename_all = "camelCase")]
98        struct Helper {
99            state_version: i64,
100            timestamp: i64,
101            counter: i64,
102        }
103
104        let helper = Helper::deserialize(deserializer)?;
105        Ok(Self::with_values(
106            helper.state_version,
107            helper.timestamp,
108            helper.counter,
109        ))
110    }
111}
112
113impl Clone for DataVersion {
114    fn clone(&self) -> Self {
115        Self::with_values(self.state_version, self.timestamp, self.get_counter())
116    }
117}
118
119impl PartialEq for DataVersion {
120    fn eq(&self, other: &Self) -> bool {
121        self.state_version == other.state_version
122            && self.timestamp == other.timestamp
123            && self.get_counter() == other.get_counter()
124    }
125}
126
127impl Eq for DataVersion {}
128
129impl PartialOrd for DataVersion {
130    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
131        Some(self.cmp(other))
132    }
133}
134
135impl Ord for DataVersion {
136    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
137        self.state_version
138            .cmp(&other.state_version)
139            .then_with(|| self.timestamp.cmp(&other.timestamp))
140            .then_with(|| self.get_counter().cmp(&other.get_counter()))
141    }
142}
143
144impl Default for DataVersion {
145    fn default() -> Self {
146        Self::with_values(0, 0, 0)
147    }
148}
149
150impl DataVersion {
151    pub const fn with_values(state_version: i64, timestamp: i64, counter: i64) -> Self {
152        Self {
153            state_version,
154            timestamp,
155            counter: AtomicI64::new(counter),
156        }
157    }
158
159    pub fn assign_new_one(&mut self, data_version: &Self) {
160        self.timestamp = data_version.timestamp;
161        self.state_version = data_version.state_version;
162        self.counter.store(data_version.get_counter(), Ordering::SeqCst);
163    }
164
165    pub fn get_state_version(&self) -> i64 {
166        self.state_version
167    }
168    pub fn set_state_version(&mut self, value: i64) {
169        self.state_version = value;
170    }
171    pub fn get_timestamp(&self) -> i64 {
172        self.timestamp
173    }
174    pub fn set_timestamp(&mut self, value: i64) {
175        self.timestamp = value;
176    }
177    pub fn get_counter(&self) -> i64 {
178        self.counter.load(Ordering::Relaxed)
179    }
180    pub fn set_counter(&self, value: i64) {
181        self.counter.store(value, Ordering::Relaxed);
182    }
183    pub fn increment_counter(&self) -> i64 {
184        self.counter.fetch_add(1, Ordering::Relaxed) + 1
185    }
186    pub fn state_version(&self) -> i64 {
187        self.state_version
188    }
189    pub fn timestamp(&self) -> i64 {
190        self.timestamp
191    }
192    pub fn counter(&self) -> i64 {
193        self.get_counter()
194    }
195    pub fn next_version_with_timestamp(&mut self, state_version: i64, timestamp: i64) {
196        self.timestamp = timestamp;
197        self.state_version = state_version;
198        self.counter.fetch_add(1, Ordering::SeqCst);
199    }
200}
201
202impl Display for DataVersion {
203    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
204        write!(
205            f,
206            "State Version: {}, Timestamp: {}, Counter: {}",
207            self.state_version,
208            self.timestamp,
209            self.get_counter()
210        )
211    }
212}
213
214pub trait RemotingSerializable {
215    fn encode(&self) -> rocketmq_error::Result<Vec<u8>>;
216    fn serialize_json(&self) -> rocketmq_error::Result<String>;
217    fn serialize_json_pretty(&self) -> rocketmq_error::Result<String>;
218}
219
220pub trait RemotingDeserializable {
221    type Output;
222    fn decode(bytes: &[u8]) -> rocketmq_error::Result<Self::Output>;
223    fn decode_str(value: &str) -> rocketmq_error::Result<Self::Output> {
224        Self::decode(value.as_bytes())
225    }
226    fn decode_string(value: String) -> rocketmq_error::Result<Self::Output> {
227        Self::decode_str(&value)
228    }
229}
230
231impl<T: Serialize> RemotingSerializable for T {
232    fn encode(&self) -> rocketmq_error::Result<Vec<u8>> {
233        serde_json::to_vec(self).map_err(|error| crate::error::serialization_source("serialize", "JSON", error))
234    }
235    fn serialize_json(&self) -> rocketmq_error::Result<String> {
236        serde_json::to_string(self).map_err(|error| crate::error::serialization_source("serialize", "JSON", error))
237    }
238    fn serialize_json_pretty(&self) -> rocketmq_error::Result<String> {
239        serde_json::to_string_pretty(self)
240            .map_err(|error| crate::error::serialization_source("serialize", "JSON", error))
241    }
242}
243
244impl<T: serde::de::DeserializeOwned> RemotingDeserializable for T {
245    type Output = T;
246    fn decode(bytes: &[u8]) -> rocketmq_error::Result<Self::Output> {
247        serde_json::from_slice(bytes).map_err(|error| crate::error::serialization_source("deserialize", "JSON", error))
248    }
249}
250
251/// Legacy handwritten fast-header codec surface.
252///
253/// Production encoding uses
254/// [`command_custom_header::CommandCustomHeader::encode_capability`] and
255/// [`command_custom_header::CommandCustomHeader::encode_direct_binary`].
256#[deprecated(note = "use CommandCustomHeader typed map/direct-binary APIs")]
257pub trait FastCodesHeader {
258    fn write_if_not_null(out: &mut BytesMut, key: &str, value: &str) {
259        if !value.is_empty() {
260            RocketMQSerializable::write_str(out, true, key);
261            RocketMQSerializable::write_str(out, false, value);
262        }
263    }
264    fn encode_fast(&mut self, out: &mut BytesMut);
265    fn decode_fast(&mut self, fields: &HashMap<CheetahString, CheetahString>);
266}
267
268#[repr(u8)]
269#[derive(Debug, PartialEq, Eq)]
270pub enum RemotingCommandType {
271    REQUEST,
272    RESPONSE,
273}
274
275impl RemotingCommandType {
276    pub fn value_of(code: u8) -> Option<Self> {
277        match code {
278            0 => Some(Self::REQUEST),
279            1 => Some(Self::RESPONSE),
280            _ => None,
281        }
282    }
283
284    pub fn get_code(&self) -> u8 {
285        match self {
286            Self::REQUEST => 0,
287            Self::RESPONSE => 1,
288        }
289    }
290
291    pub fn get_type_from_name(name: &str) -> Option<Self> {
292        match name.to_ascii_uppercase().as_str() {
293            "REQUEST" => Some(Self::REQUEST),
294            "RESPONSE" => Some(Self::RESPONSE),
295            _ => None,
296        }
297    }
298}
299
300#[repr(u8)]
301#[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize, Default, Hash, Copy)]
302pub enum LanguageCode {
303    JAVA,
304    CPP,
305    DOTNET,
306    PYTHON,
307    DELPHI,
308    ERLANG,
309    RUBY,
310    OTHER,
311    HTTP,
312    GO,
313    PHP,
314    OMS,
315    #[default]
316    RUST,
317    #[allow(non_camel_case_types, reason = "matches the Java remoting wire language name")]
318    NODE_JS,
319}
320
321impl fmt::Display for LanguageCode {
322    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
323        write!(f, "{self:?}")
324    }
325}
326
327impl From<LanguageCode> for u8 {
328    fn from(code: LanguageCode) -> Self {
329        code.get_code()
330    }
331}
332
333impl From<LanguageCode> for i32 {
334    fn from(code: LanguageCode) -> Self {
335        code.get_code() as i32
336    }
337}
338
339impl From<u32> for LanguageCode {
340    fn from(code: u32) -> Self {
341        u8::try_from(code).map_or(Self::OTHER, Self::from)
342    }
343}
344
345impl From<i32> for LanguageCode {
346    fn from(code: i32) -> Self {
347        u8::try_from(code).map_or(Self::OTHER, Self::from)
348    }
349}
350
351impl From<u8> for LanguageCode {
352    fn from(code: u8) -> Self {
353        Self::value_of(code).unwrap_or(Self::OTHER)
354    }
355}
356
357impl LanguageCode {
358    pub fn value_of(code: u8) -> Option<Self> {
359        Some(match code {
360            0 => Self::JAVA,
361            1 => Self::CPP,
362            2 => Self::DOTNET,
363            3 => Self::PYTHON,
364            4 => Self::DELPHI,
365            5 => Self::ERLANG,
366            6 => Self::RUBY,
367            7 => Self::OTHER,
368            8 => Self::HTTP,
369            9 => Self::GO,
370            10 => Self::PHP,
371            11 => Self::OMS,
372            12 => Self::RUST,
373            13 => Self::NODE_JS,
374            _ => Self::OTHER,
375        })
376    }
377
378    pub fn get_code(&self) -> u8 {
379        *self as u8
380    }
381
382    pub fn get_code_from_name(name: &str) -> Option<Self> {
383        match name.to_ascii_uppercase().as_str() {
384            "JAVA" => Some(Self::JAVA),
385            "CPP" => Some(Self::CPP),
386            "DOTNET" => Some(Self::DOTNET),
387            "PYTHON" => Some(Self::PYTHON),
388            "DELPHI" => Some(Self::DELPHI),
389            "ERLANG" => Some(Self::ERLANG),
390            "RUBY" => Some(Self::RUBY),
391            "OTHER" => Some(Self::OTHER),
392            "HTTP" => Some(Self::HTTP),
393            "GO" => Some(Self::GO),
394            "PHP" => Some(Self::PHP),
395            "OMS" => Some(Self::OMS),
396            "RUST" => Some(Self::RUST),
397            "NODE_JS" => Some(Self::NODE_JS),
398            _ => None,
399        }
400    }
401}
402
403#[repr(u8)]
404#[derive(Debug, PartialEq, Eq, Clone, Copy, Deserialize, Serialize)]
405pub enum SerializeType {
406    JSON,
407    ROCKETMQ,
408}
409
410impl fmt::Display for SerializeType {
411    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
412        write!(f, "{self:?}")
413    }
414}
415
416impl SerializeType {
417    pub fn value_of(code: u8) -> Option<Self> {
418        match code {
419            0 => Some(Self::JSON),
420            1 => Some(Self::ROCKETMQ),
421            _ => None,
422        }
423    }
424
425    pub fn get_code(&self) -> u8 {
426        *self as u8
427    }
428}
429
430#[cfg(test)]
431mod tests {
432    use std::error::Error as StdError;
433
434    use serde::Deserialize;
435    use serde::Serialize;
436    use serde::Serializer;
437
438    use super::RemotingDeserializable;
439    use super::RemotingSerializable;
440
441    struct FailingJsonSerialize;
442
443    impl Serialize for FailingJsonSerialize {
444        fn serialize<S>(&self, _serializer: S) -> Result<S::Ok, S::Error>
445        where
446            S: Serializer,
447        {
448            Err(<S::Error as serde::ser::Error>::custom(
449                "forced JSON serialization failure",
450            ))
451        }
452    }
453
454    #[derive(Debug, Deserialize)]
455    struct DecodedValue;
456
457    fn assert_json_source(error: &rocketmq_error::Error) {
458        let direct_source = StdError::source(error).expect("JSON error source");
459        assert!(direct_source.downcast_ref::<serde_json::Error>().is_some());
460    }
461
462    #[test]
463    fn protocol_json_encode_preserves_source_and_public_boundary() {
464        let error = FailingJsonSerialize.encode().unwrap_err();
465
466        assert_eq!(error.descriptor(), &rocketmq_error::CORE_SERIALIZATION_FAILED);
467        assert_eq!(error.descriptor().public_message(), "Serialization failed");
468        assert_json_source(&error);
469    }
470
471    #[test]
472    fn protocol_json_decode_preserves_source_and_public_boundary() {
473        let error = DecodedValue::decode(b"invalid").unwrap_err();
474
475        assert_eq!(error.descriptor(), &rocketmq_error::CORE_SERIALIZATION_FAILED);
476        assert_eq!(error.descriptor().public_message(), "Serialization failed");
477        assert_json_source(&error);
478    }
479}
rocketmq_protocol/protocol.rs

rocketmq_protocol/
protocol.rs
