rocketmq_protocol/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::RequestHeaderCodec;
17use serde::Deserialize;
18use serde::Serialize;
19
20use crate::protocol::header::namesrv::topic_operation_header::TopicRequestHeader;
21
22#[derive(Debug, Default, Serialize, Deserialize, RequestHeaderCodec)]
23#[header(
24    type_id = "rocketmq_protocol::protocol::header::notification_request_header::NotificationRequestHeader",
25    java_class = "org.apache.rocketmq.remoting.protocol.header.NotificationRequestHeader",
26    fast
27)]
28#[serde(rename_all = "camelCase")]
29pub struct NotificationRequestHeader {
30    #[header(required)]
31    pub consumer_group: CheetahString,
32
33    #[header(required)]
34    pub topic: CheetahString,
35
36    #[header(required)]
37    pub queue_id: i32,
38
39    #[header(required)]
40    pub poll_time: i64,
41
42    #[header(required)]
43    pub born_time: i64,
44
45    /// Indicates if the message is ordered; defaults to false.
46    #[serde(default)]
47    #[header(default, default_semantic = "literal:false")]
48    pub order: bool,
49
50    /// Attempt ID
51    #[serde(skip_serializing_if = "Option::is_none")]
52    pub attempt_id: Option<CheetahString>,
53
54    #[serde(skip_serializing_if = "Option::is_none")]
55    pub exp_type: Option<CheetahString>,
56
57    #[serde(rename = "exp", skip_serializing_if = "Option::is_none")]
58    pub exp: Option<CheetahString>,
59
60    #[serde(default)]
61    #[header(default, default_semantic = "literal:false")]
62    pub is_lite_consumer: bool,
63
64    #[serde(skip_serializing_if = "Option::is_none")]
65    pub client_id: Option<CheetahString>,
66
67    #[serde(flatten)]
68    #[header(flatten, presence = "always")]
69    pub topic_request_header: Option<TopicRequestHeader>,
70}
71
72#[cfg(test)]
73mod tests {
74    use serde_json;
75
76    use super::*;
77    use crate::protocol::header_codec::HeaderCodec;
78    use crate::HeaderMap;
79
80    #[test]
81    fn test_notification_request_header_serialization() {
82        let header = NotificationRequestHeader {
83            consumer_group: CheetahString::from("consumer_group_1"),
84            topic: CheetahString::from("test_topic"),
85            queue_id: 10,
86            poll_time: 1234567890,
87            born_time: 1234567891,
88            order: true,
89            attempt_id: Some(CheetahString::from("attempt_1")),
90            exp_type: Some(CheetahString::from("TAG")),
91            exp: Some(CheetahString::from("tag-a")),
92            is_lite_consumer: false,
93            client_id: Some(CheetahString::from("client-a")),
94            topic_request_header: None,
95        };
96
97        let serialized = serde_json::to_string(&header).expect("Failed to serialize header");
98        assert!(serialized.contains("\"expType\":\"TAG\""));
99        assert!(serialized.contains("\"exp\":\"tag-a\""));
100
101        let deserialized: NotificationRequestHeader =
102            serde_json::from_str(&serialized).expect("Failed to deserialize header");
103        assert_eq!(header.queue_id, deserialized.queue_id);
104        assert_eq!(deserialized.exp_type.as_deref(), Some("TAG"));
105        assert_eq!(deserialized.exp.as_deref(), Some("tag-a"));
106    }
107
108    #[test]
109    fn notification_header_defaults_and_required_fields_match_the_wire_contract() {
110        let mut fields = HeaderMap::from([
111            ("consumerGroup".into(), "group-a".into()),
112            ("topic".into(), "topic-a".into()),
113            ("queueId".into(), "3".into()),
114            ("pollTime".into(), "15000".into()),
115            ("bornTime".into(), "1720000000000".into()),
116        ]);
117
118        let decoded = <NotificationRequestHeader as HeaderCodec>::decode_from_map(&fields).unwrap();
119        assert!(!decoded.order);
120        assert!(!decoded.is_lite_consumer);
121        assert_eq!(decoded.topic_request_header.as_ref().and_then(|header| header.lo), None);
122
123        let encoded = <NotificationRequestHeader as crate::CommandCustomHeader>::to_map(&decoded).unwrap();
124        assert_eq!(encoded.get("order").map(CheetahString::as_str), Some("false"));
125        assert_eq!(encoded.get("isLiteConsumer").map(CheetahString::as_str), Some("false"));
126
127        fields.remove("consumerGroup");
128        assert!(<NotificationRequestHeader as HeaderCodec>::decode_from_map(&fields).is_err());
129    }
130}
rocketmq_protocol/protocol/header/notification_request_header.rs

rocketmq_protocol/protocol/header/
notification_request_header.rs
