rocketmq_protocol/protocol/remoting_command/
decode.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 bytes::BytesMut;
16
17use super::parse_header_length;
18use super::parse_serialize_type;
19use super::RemotingCommand;
20use super::SerializeType;
21use crate::rocketmq_serializable::RocketMQSerializable;
22
23mod json;
24
25use json::try_decode_json_header;
26use json::try_decode_json_header_bytes;
27
28impl RemotingCommand {
29    /// Decodes one command with the historical 16 MiB announced-payload ceiling.
30    ///
31    /// Transport owners with an explicit total-wire limit should use
32    /// [`Self::decode_with_max_frame_bytes`] so inbound and outbound policy stays symmetric.
33    #[inline]
34    pub fn decode(src: &mut BytesMut) -> rocketmq_error::Result<Option<RemotingCommand>> {
35        Self::decode_with_max_frame_bytes(src, 16 * 1024 * 1024 + 4)
36    }
37
38    /// Decodes one command bounded by a caller-owned complete wire-frame limit.
39    ///
40    /// `max_frame_bytes` includes the four-byte announced-length prefix.
41    ///
42    /// # Errors
43    ///
44    /// Returns a serialization error for a negative, overflowing, oversized, or malformed frame.
45    #[inline]
46    pub fn decode_with_max_frame_bytes(
47        src: &mut BytesMut,
48        max_frame_bytes: usize,
49    ) -> rocketmq_error::Result<Option<RemotingCommand>> {
50        const FRAME_HEADER_SIZE: usize = 4;
51        const SERIALIZE_TYPE_SIZE: usize = 4;
52        const MIN_PAYLOAD_SIZE: usize = SERIALIZE_TYPE_SIZE; // Minimum: just serialize_type field
53
54        let available = src.len();
55
56        // Early return if not enough data for frame header
57        if available < FRAME_HEADER_SIZE {
58            return Ok(None);
59        }
60
61        // Read total size without advancing the buffer (peek)
62        let announced_size = i32::from_be_bytes([src[0], src[1], src[2], src[3]]);
63        let total_size = usize::try_from(announced_size).map_err(|_| {
64            crate::error::serialization_decode_failed(
65                "remoting_command",
66                format!("Invalid negative frame size {announced_size}"),
67            )
68        })?;
69        let full_frame_size = total_size.checked_add(FRAME_HEADER_SIZE).ok_or_else(|| {
70            crate::error::serialization_decode_failed(
71                "remoting_command",
72                format!("Frame size {total_size} overflows the wire envelope"),
73            )
74        })?;
75
76        if full_frame_size > max_frame_bytes {
77            return Err(crate::error::serialization_decode_failed(
78                "remoting_command",
79                format!("Wire frame size {full_frame_size} exceeds configured limit {max_frame_bytes}"),
80            ));
81        }
82
83        // Wait for complete frame
84        if available < full_frame_size {
85            return Ok(None);
86        }
87
88        // Now validate minimum total_size (we have the complete frame)
89        if total_size < MIN_PAYLOAD_SIZE {
90            return Err(crate::error::serialization_decode_failed(
91                "remoting_command",
92                format!("Invalid total_size {total_size}, minimum required is {MIN_PAYLOAD_SIZE}"),
93            ));
94        }
95
96        // Extract the complete frame before validating the marked header so
97        // malformed complete frames preserve the established consume-on-error
98        // behavior.
99        let frame_data = src.split_to(full_frame_size);
100        let ori_header_length = i32::from_be_bytes([frame_data[4], frame_data[5], frame_data[6], frame_data[7]]);
101        let header_length = parse_header_length(ori_header_length);
102
103        // Validate header length
104        if header_length > total_size - SERIALIZE_TYPE_SIZE {
105            return Err(crate::error::serialization_decode_failed(
106                "remoting_command",
107                format!("Invalid header length {header_length}, total size {total_size}"),
108            ));
109        }
110
111        let protocol_type = parse_serialize_type(ori_header_length)?;
112        let frame = frame_data.freeze();
113        let header_start = FRAME_HEADER_SIZE + SERIALIZE_TYPE_SIZE;
114        let header_end = header_start + header_length;
115        let mut cmd = match protocol_type {
116            SerializeType::ROCKETMQ => {
117                RocketMQSerializable::rocket_mq_protocol_decode_bytes(frame.slice(header_start..header_end))?
118            }
119            SerializeType::JSON => {
120                if let Some(cmd) = try_decode_json_header_bytes(frame.slice(header_start..header_end)) {
121                    cmd
122                } else {
123                    // Unsupported JSON shapes retain the Serde/SIMD
124                    // compatibility path. Copying is confined to this cold
125                    // fallback after consuming the complete frame.
126                    let mut header_data = BytesMut::from(&frame[header_start..header_end]);
127                    Self::decode_json_header_fallback(&mut header_data, header_length)?
128                }
129            }
130        };
131        cmd.set_serialize_type_ref(protocol_type);
132        if header_end < frame.len() {
133            cmd.set_body_mut_ref(frame.slice(header_end..));
134        }
135        Ok(Some(cmd))
136    }
137
138    #[cold]
139    #[inline(never)]
140    fn decode_json_header_fallback(
141        src: &mut BytesMut,
142        _header_length: usize,
143    ) -> rocketmq_error::Result<RemotingCommand> {
144        #[cfg(feature = "simd")]
145        {
146            let mut slice = src.split_to(_header_length).to_vec();
147            simd_json::from_slice::<RemotingCommand>(&mut slice)
148                .map_err(|error| crate::error::serialization_source("decode", "json", error))
149        }
150
151        #[cfg(not(feature = "simd"))]
152        {
153            serde_json::from_slice::<RemotingCommand>(src)
154                .map_err(|error| crate::error::serialization_source("decode", "json", error))
155        }
156    }
157
158    /// Optimized header decoding with type-based dispatch
159    #[inline]
160    pub fn header_decode(
161        src: &mut BytesMut,
162        header_length: usize,
163        type_: SerializeType,
164    ) -> rocketmq_error::Result<Option<RemotingCommand>> {
165        match type_ {
166            SerializeType::JSON => {
167                if let Some(cmd) = try_decode_json_header(src, header_length) {
168                    return Ok(Some(cmd.set_serialize_type(SerializeType::JSON)));
169                }
170                let cmd = Self::decode_json_header_fallback(src, header_length)?;
171                Ok(Some(cmd.set_serialize_type(SerializeType::JSON)))
172            }
173            SerializeType::ROCKETMQ => {
174                // Deserialize binary header
175                let cmd = RocketMQSerializable::rocket_mq_protocol_decode(src, header_length)?;
176                Ok(Some(cmd.set_serialize_type(SerializeType::ROCKETMQ)))
177            }
178        }
179    }
180}
rocketmq_protocol/protocol/remoting_command/decode.rs

rocketmq_protocol/protocol/remoting_command/
decode.rs
