rocketmq_client_rust/utils/
message_util.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::sync::LazyLock;
16
17use bytes::Bytes;
18use cheetah_string::CheetahString;
19use rocketmq_model::common::message::message_accessor::MessageAccessor;
20use rocketmq_model::common::message::message_single::Message;
21use rocketmq_model::common::message::MessageConst;
22use rocketmq_model::common::mix_all;
23
24use crate::common::client_error_code::ClientErrorCode;
25
26// Cached static strings to avoid repeated allocations
27static PROPERTY_CLUSTER: LazyLock<CheetahString> =
28    LazyLock::new(|| CheetahString::from_static_str(MessageConst::PROPERTY_CLUSTER));
29static PROPERTY_MESSAGE_TYPE: LazyLock<CheetahString> =
30    LazyLock::new(|| CheetahString::from_static_str(MessageConst::PROPERTY_MESSAGE_TYPE));
31static REPLY_MESSAGE_FLAG: LazyLock<CheetahString> =
32    LazyLock::new(|| CheetahString::from_static_str(mix_all::REPLY_MESSAGE_FLAG));
33static PROPERTY_MESSAGE_REPLY_TO_CLIENT: LazyLock<CheetahString> =
34    LazyLock::new(|| CheetahString::from_static_str(MessageConst::PROPERTY_MESSAGE_REPLY_TO_CLIENT));
35static PROPERTY_CORRELATION_ID: LazyLock<CheetahString> =
36    LazyLock::new(|| CheetahString::from_static_str(MessageConst::PROPERTY_CORRELATION_ID));
37static PROPERTY_MESSAGE_TTL: LazyLock<CheetahString> =
38    LazyLock::new(|| CheetahString::from_static_str(MessageConst::PROPERTY_MESSAGE_TTL));
39
40pub struct MessageUtil;
41
42impl MessageUtil {
43    pub fn create_reply_message(request_message: &Message, body: &[u8]) -> crate::ClientResult<Message> {
44        // Early return: check required cluster property
45        let Some(cluster) = request_message.property(&PROPERTY_CLUSTER) else {
46            return Err(mq_client_err!(
47                ClientErrorCode::CREATE_REPLY_MESSAGE_EXCEPTION,
48                format!(
49                    "create reply message fail, requestMessage error, property[{}] is null.",
50                    MessageConst::PROPERTY_CLUSTER
51                )
52            ));
53        };
54
55        let mut reply_message = Message::default();
56        reply_message.set_body(Some(Bytes::copy_from_slice(body)));
57
58        let reply_topic = mix_all::get_reply_topic(cluster);
59        reply_message.set_topic(CheetahString::from_string(reply_topic));
60
61        // Set message type using cached static string
62        MessageAccessor::put_property(
63            &mut reply_message,
64            PROPERTY_MESSAGE_TYPE.clone(),
65            REPLY_MESSAGE_FLAG.clone(),
66        );
67
68        // Copy optional properties if present
69        if let Some(reply_to) = request_message.property(&PROPERTY_MESSAGE_REPLY_TO_CLIENT) {
70            MessageAccessor::put_property(
71                &mut reply_message,
72                PROPERTY_MESSAGE_REPLY_TO_CLIENT.clone(),
73                CheetahString::from_slice(reply_to),
74            );
75        }
76
77        if let Some(correlation_id) = request_message.property(&PROPERTY_CORRELATION_ID) {
78            MessageAccessor::put_property(
79                &mut reply_message,
80                PROPERTY_CORRELATION_ID.clone(),
81                CheetahString::from_slice(correlation_id),
82            );
83        }
84
85        if let Some(ttl) = request_message.property(&PROPERTY_MESSAGE_TTL) {
86            MessageAccessor::put_property(
87                &mut reply_message,
88                PROPERTY_MESSAGE_TTL.clone(),
89                CheetahString::from_slice(ttl),
90            );
91        }
92
93        Ok(reply_message)
94    }
95
96    pub fn get_reply_to_client(reply_message: &Message) -> Option<CheetahString> {
97        reply_message
98            .property(&PROPERTY_MESSAGE_REPLY_TO_CLIENT)
99            .map(CheetahString::from_slice)
100    }
101}
102
103#[cfg(test)]
104mod tests {
105    use super::*;
106    use crate::test_support::error_assertions::client_exception;
107
108    #[test]
109    fn create_reply_message_uses_java_reply_topic_and_copies_request_properties() {
110        let mut request = Message::default();
111        MessageAccessor::put_property(
112            &mut request,
113            PROPERTY_CLUSTER.clone(),
114            CheetahString::from_static_str("ClusterA"),
115        );
116        MessageAccessor::put_property(
117            &mut request,
118            PROPERTY_MESSAGE_REPLY_TO_CLIENT.clone(),
119            CheetahString::from_static_str("client-1"),
120        );
121        MessageAccessor::put_property(
122            &mut request,
123            PROPERTY_CORRELATION_ID.clone(),
124            CheetahString::from_static_str("corr-1"),
125        );
126        MessageAccessor::put_property(
127            &mut request,
128            PROPERTY_MESSAGE_TTL.clone(),
129            CheetahString::from_static_str("3000"),
130        );
131
132        let reply =
133            MessageUtil::create_reply_message(&request, b"reply-body").expect("reply message should be created");
134
135        assert_eq!(reply.topic().as_str(), "ClusterA_REPLY_TOPIC");
136        assert_eq!(
137            reply.body().as_ref().map(|body| body.as_ref()),
138            Some(b"reply-body".as_slice())
139        );
140        assert_eq!(
141            reply.property(&PROPERTY_MESSAGE_TYPE),
142            Some(mix_all::REPLY_MESSAGE_FLAG)
143        );
144        assert_eq!(reply.property(&PROPERTY_MESSAGE_REPLY_TO_CLIENT), Some("client-1"));
145        assert_eq!(reply.property(&PROPERTY_CORRELATION_ID), Some("corr-1"));
146        assert_eq!(reply.property(&PROPERTY_MESSAGE_TTL), Some("3000"));
147    }
148
149    #[test]
150    fn create_reply_message_requires_cluster_property_like_java() {
151        let request = Message::default();
152
153        let error =
154            MessageUtil::create_reply_message(&request, b"reply-body").expect_err("missing cluster should fail");
155
156        let exception = client_exception(&error);
157        assert_eq!(
158            exception.response_code(),
159            ClientErrorCode::CREATE_REPLY_MESSAGE_EXCEPTION
160        );
161        assert!(exception.to_string().contains(MessageConst::PROPERTY_CLUSTER));
162    }
163
164    fn request_with_cluster_only() -> Message {
165        let mut request = Message::default();
166        MessageAccessor::put_property(
167            &mut request,
168            PROPERTY_CLUSTER.clone(),
169            CheetahString::from_static_str("ClusterA"),
170        );
171        request
172    }
173
174    #[test]
175    fn create_reply_message_with_only_cluster_leaves_optional_properties_absent() {
176        let request = request_with_cluster_only();
177
178        let reply = MessageUtil::create_reply_message(&request, b"body").expect("reply should be created");
179
180        assert_eq!(reply.property(&PROPERTY_MESSAGE_REPLY_TO_CLIENT), None);
181        assert_eq!(reply.property(&PROPERTY_CORRELATION_ID), None);
182        assert_eq!(reply.property(&PROPERTY_MESSAGE_TTL), None);
183        assert_eq!(MessageUtil::get_reply_to_client(&reply), None);
184    }
185
186    #[test]
187    fn create_reply_message_copies_each_optional_property_independently() {
188        let mut reply_to_only = request_with_cluster_only();
189        MessageAccessor::put_property(
190            &mut reply_to_only,
191            PROPERTY_MESSAGE_REPLY_TO_CLIENT.clone(),
192            CheetahString::from_static_str("client-1"),
193        );
194        let reply = MessageUtil::create_reply_message(&reply_to_only, b"body").expect("reply should be created");
195        assert_eq!(
196            MessageUtil::get_reply_to_client(&reply),
197            Some(CheetahString::from_static_str("client-1"))
198        );
199        assert_eq!(reply.property(&PROPERTY_CORRELATION_ID), None);
200        assert_eq!(reply.property(&PROPERTY_MESSAGE_TTL), None);
201
202        let mut correlation_only = request_with_cluster_only();
203        MessageAccessor::put_property(
204            &mut correlation_only,
205            PROPERTY_CORRELATION_ID.clone(),
206            CheetahString::from_static_str("corr-1"),
207        );
208        let reply = MessageUtil::create_reply_message(&correlation_only, b"body").expect("reply should be created");
209        assert_eq!(MessageUtil::get_reply_to_client(&reply), None);
210        assert_eq!(reply.property(&PROPERTY_CORRELATION_ID), Some("corr-1"));
211        assert_eq!(reply.property(&PROPERTY_MESSAGE_TTL), None);
212
213        let mut ttl_only = request_with_cluster_only();
214        MessageAccessor::put_property(
215            &mut ttl_only,
216            PROPERTY_MESSAGE_TTL.clone(),
217            CheetahString::from_static_str("5000"),
218        );
219        let reply = MessageUtil::create_reply_message(&ttl_only, b"body").expect("reply should be created");
220        assert_eq!(MessageUtil::get_reply_to_client(&reply), None);
221        assert_eq!(reply.property(&PROPERTY_CORRELATION_ID), None);
222        assert_eq!(reply.property(&PROPERTY_MESSAGE_TTL), Some("5000"));
223    }
224
225    #[test]
226    fn create_reply_message_uses_supplied_body_and_ignores_unrelated_request_properties() {
227        let mut request = request_with_cluster_only();
228        request.set_body(Some(Bytes::copy_from_slice(b"request-body")));
229        MessageAccessor::put_property(
230            &mut request,
231            CheetahString::from_static_str("customKey"),
232            CheetahString::from_static_str("customValue"),
233        );
234
235        let reply = MessageUtil::create_reply_message(&request, b"reply-body-supplied")
236            .expect("reply message should be created");
237
238        assert_eq!(
239            reply.body().as_ref().map(|body| body.as_ref()),
240            Some(b"reply-body-supplied".as_slice())
241        );
242        assert_eq!(reply.property(&CheetahString::from_static_str("customKey")), None);
243    }
244
245    #[test]
246    fn create_reply_message_copies_body_independently_of_caller_buffer_and_supports_empty_body() {
247        let request = request_with_cluster_only();
248
249        let mut buffer = vec![1u8, 2, 3];
250        let reply = MessageUtil::create_reply_message(&request, &buffer).expect("reply should be created");
251        buffer[0] = 99;
252        buffer.push(4);
253        assert_eq!(
254            reply.body().as_ref().map(|body| body.as_ref()),
255            Some([1u8, 2, 3].as_slice())
256        );
257
258        let empty_reply = MessageUtil::create_reply_message(&request, &[]).expect("empty body reply should be created");
259        assert_eq!(
260            empty_reply.body().as_ref().map(|body| body.as_ref()),
261            Some([].as_slice())
262        );
263    }
264}
rocketmq_client_rust/utils/message_util.rs

rocketmq_client_rust/utils/
message_util.rs
