rocketmq_remoting/protocol/header/
notification_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::RequestHeaderCodecV2;
17use serde::Deserialize;
18use serde::Serialize;
19
20use crate::protocol::header::namesrv::topic_operation_header::TopicRequestHeader;
21
22#[derive(Debug, Serialize, Deserialize, RequestHeaderCodecV2)]
23pub struct NotificationRequestHeader {
24    #[serde(rename = "consumerGroup")]
25    #[required]
26    pub consumer_group: CheetahString,
27
28    #[serde(rename = "topic")]
29    #[required]
30    pub topic: CheetahString,
31
32    #[serde(rename = "queueId")]
33    #[required]
34    pub queue_id: i32,
35
36    #[serde(rename = "pollTime")]
37    #[required]
38    pub poll_time: i64,
39
40    #[serde(rename = "bornTime")]
41    #[required]
42    pub born_time: i64,
43
44    /// Indicates if the message is ordered; defaults to false.
45    #[serde(default)]
46    pub order: bool,
47
48    /// Attempt ID
49    #[serde(rename = "attemptId", skip_serializing_if = "Option::is_none")]
50    pub attempt_id: Option<CheetahString>,
51
52    #[serde(rename = "expType", skip_serializing_if = "Option::is_none")]
53    pub exp_type: Option<CheetahString>,
54
55    #[serde(rename = "exp", skip_serializing_if = "Option::is_none")]
56    pub exp: Option<CheetahString>,
57
58    #[serde(flatten)]
59    pub topic_request_header: Option<TopicRequestHeader>,
60}
61
62#[cfg(test)]
63mod tests {
64    use serde_json;
65
66    use super::*;
67
68    #[test]
69    fn test_notification_request_header_serialization() {
70        let header = NotificationRequestHeader {
71            consumer_group: CheetahString::from("consumer_group_1"),
72            topic: CheetahString::from("test_topic"),
73            queue_id: 10,
74            poll_time: 1234567890,
75            born_time: 1234567891,
76            order: true,
77            attempt_id: Some(CheetahString::from("attempt_1")),
78            exp_type: Some(CheetahString::from("TAG")),
79            exp: Some(CheetahString::from("tag-a")),
80            topic_request_header: None,
81        };
82
83        let serialized = serde_json::to_string(&header).expect("Failed to serialize header");
84        assert!(serialized.contains("\"expType\":\"TAG\""));
85        assert!(serialized.contains("\"exp\":\"tag-a\""));
86
87        let deserialized: NotificationRequestHeader =
88            serde_json::from_str(&serialized).expect("Failed to deserialize header");
89        assert_eq!(header.queue_id, deserialized.queue_id);
90        assert_eq!(deserialized.exp_type.as_deref(), Some("TAG"));
91        assert_eq!(deserialized.exp.as_deref(), Some("tag-a"));
92    }
93
94    #[test]
95    fn test_notification_request_header_default_order() {
96        let header = NotificationRequestHeader {
97            consumer_group: CheetahString::from("consumer_group_1"),
98            topic: CheetahString::from("test_topic"),
99            queue_id: 10,
100            poll_time: 1234567890,
101            born_time: 1234567891,
102            order: false, // Defaults to false
103            attempt_id: None,
104            exp_type: None,
105            exp: None,
106            topic_request_header: None,
107        };
108
109        let serialized = serde_json::to_string(&header).expect("Failed to serialize header");
110
111        let deserialized: NotificationRequestHeader =
112            serde_json::from_str(&serialized).expect("Failed to deserialize header");
113        assert_eq!(header.queue_id, deserialized.queue_id);
114    }
115}
rocketmq_remoting/protocol/header/notification_request_header.rs

rocketmq_remoting/protocol/header/
notification_request_header.rs
