rocketmq_protocol/protocol/remoting_command/
encode.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::Bytes;
16use bytes::BytesMut;
17
18use super::RemotingCommand;
19use super::SerializeType;
20use crate::rocketmq_serializable::RocketMQSerializable;
21
22mod json;
23mod rocketmq;
24
25impl RemotingCommand {
26    #[inline]
27    pub(super) fn try_header_encode(&mut self) -> rocketmq_error::Result<Bytes> {
28        match self.serialize_type {
29            SerializeType::ROCKETMQ => {
30                let mut encoded = BytesMut::new();
31                RocketMQSerializable::try_rocketmq_protocol_encode(self, &mut encoded)
32                    .map_err(crate::protocol::header_codec::into_error)?;
33                Ok(encoded.freeze())
34            }
35            SerializeType::JSON => {
36                self.try_make_custom_header_to_net()
37                    .map_err(crate::protocol::header_codec::into_error)?;
38                #[cfg(feature = "simd")]
39                {
40                    simd_json::to_vec(self)
41                        .map(Bytes::from)
42                        .map_err(|source| crate::error::serialization_source("encode", "remoting-command", source))
43                }
44                #[cfg(not(feature = "simd"))]
45                {
46                    serde_json::to_vec(self)
47                        .map(Bytes::from)
48                        .map_err(|source| crate::error::serialization_source("encode", "remoting-command", source))
49                }
50            }
51        }
52    }
53
54    /// Encodes the frame header and rolls the destination back on failure.
55    ///
56    /// # Errors
57    ///
58    /// Returns the custom-header validation or direct-binary encoding failure.
59    #[inline]
60    pub fn try_fast_header_encode(&mut self, dst: &mut BytesMut) -> rocketmq_error::Result<()> {
61        let body_length = self.body.as_ref().map_or(0, Bytes::len);
62        self.try_fast_header_encode_with_body_length(dst, body_length)
63    }
64
65    #[inline]
66    pub(crate) fn try_fast_header_encode_with_body_length(
67        &mut self,
68        dst: &mut BytesMut,
69        body_length: usize,
70    ) -> rocketmq_error::Result<()> {
71        let checkpoint = dst.len();
72        let result = match self.body.as_ref() {
73            Some(body) if body.len() != body_length => Err(crate::error::serialization_encode_failed(
74                "remoting-command",
75                "explicit body length does not match the in-memory body",
76            )),
77            _ => self.try_fast_header_encode_inner(dst, body_length),
78        };
79        if result.is_err() {
80            dst.truncate(checkpoint);
81        }
82        result
83    }
84
85    #[inline]
86    fn try_fast_header_encode_inner(&mut self, dst: &mut BytesMut, body_length: usize) -> rocketmq_error::Result<()> {
87        match self.serialize_type {
88            SerializeType::JSON => self.fast_encode_json(dst, body_length),
89            SerializeType::ROCKETMQ => self.fast_encode_rocketmq(dst, body_length),
90        }
91    }
92}
rocketmq_protocol/protocol/remoting_command/encode.rs

rocketmq_protocol/protocol/remoting_command/
encode.rs
