rocketmq_model/common/sys_flag/
message_sys_flag.rs

1// Copyright 2026 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 crate::common::compression::compression_type::CompressionType;
16use rocketmq_error::Result;
17
18// Meaning of each bit in the system flag
19///
20/// | bit    | 7 | 6 | 5         | 4        | 3           | 2                | 1                | 0                |
21/// |--------|---|---|-----------|----------|-------------|------------------|------------------|------------------|
22/// | byte 1 |   |   | STOREHOST | BORNHOST | TRANSACTION | TRANSACTION      | MULTI_TAGS       | COMPRESSED       |
23/// | byte 2 |   |   |           |          |             | COMPRESSION_TYPE | COMPRESSION_TYPE | COMPRESSION_TYPE |
24/// | byte 3 |   |   |           |          |             |                  |                  |                  |
25/// | byte 4 |   |   |           |          |             |                  |                  |
26pub struct MessageSysFlag;
27
28impl MessageSysFlag {
29    pub const COMPRESSED_FLAG: i32 = 0x1;
30    pub const MULTI_TAGS_FLAG: i32 = 0x1 << 1;
31
32    //Transaction related flag
33    pub const TRANSACTION_NOT_TYPE: i32 = 0;
34    pub const TRANSACTION_PREPARED_TYPE: i32 = 0x1 << 2;
35    pub const TRANSACTION_COMMIT_TYPE: i32 = 0x2 << 2;
36    pub const TRANSACTION_ROLLBACK_TYPE: i32 = 0x3 << 2;
37
38    //Flag of the message properties
39    pub const BORNHOST_V6_FLAG: i32 = 0x1 << 4;
40    pub const STOREHOSTADDRESS_V6_FLAG: i32 = 0x1 << 5;
41
42    //Mark the flag for batch to avoid conflict
43    pub const NEED_UNWRAP_FLAG: i32 = 0x1 << 6;
44    pub const INNER_BATCH_FLAG: i32 = 0x1 << 7;
45
46    // COMPRESSION_TYPE
47    pub const COMPRESSION_LZ4_TYPE: i32 = 0x1 << 8;
48    pub const COMPRESSION_ZSTD_TYPE: i32 = 0x2 << 8;
49    pub const COMPRESSION_ZLIB_TYPE: i32 = 0x3 << 8;
50    pub const COMPRESSION_TYPE_COMPARATOR: i32 = 0x7 << 8;
51
52    #[inline]
53    pub const fn get_transaction_value(flag: i32) -> i32 {
54        flag & Self::TRANSACTION_ROLLBACK_TYPE
55    }
56
57    #[inline]
58    pub const fn reset_transaction_value(flag: i32, transaction_type: i32) -> i32 {
59        (flag & !Self::TRANSACTION_ROLLBACK_TYPE) | transaction_type
60    }
61
62    #[inline]
63    pub const fn clear_compressed_flag(flag: i32) -> i32 {
64        flag & !Self::COMPRESSED_FLAG
65    }
66
67    #[inline]
68    pub fn get_compression_type(flag: i32) -> Result<CompressionType> {
69        let compression_type_value = (flag & Self::COMPRESSION_TYPE_COMPARATOR) >> 8;
70        CompressionType::find_by_value(compression_type_value)
71    }
72
73    #[inline]
74    pub fn try_get_compression_type(flag: i32) -> Result<CompressionType> {
75        Self::get_compression_type(flag)
76    }
77
78    #[inline]
79    pub const fn check(flag: i32, expected_flag: i32) -> bool {
80        (flag & expected_flag) != 0
81    }
82}
83
84#[cfg(test)]
85mod tests {
86    use super::*;
87
88    #[test]
89    fn get_transaction_value_returns_correct_value() {
90        let flag = MessageSysFlag::TRANSACTION_COMMIT_TYPE;
91        assert_eq!(
92            MessageSysFlag::get_transaction_value(flag),
93            MessageSysFlag::TRANSACTION_COMMIT_TYPE
94        );
95    }
96
97    #[test]
98    fn reset_transaction_value_resets_correctly() {
99        let flag = MessageSysFlag::TRANSACTION_COMMIT_TYPE;
100        let new_flag = MessageSysFlag::reset_transaction_value(flag, MessageSysFlag::TRANSACTION_ROLLBACK_TYPE);
101        assert_eq!(new_flag, MessageSysFlag::TRANSACTION_ROLLBACK_TYPE);
102    }
103
104    #[test]
105    fn clear_compressed_flag_clears_correctly() {
106        let flag = MessageSysFlag::COMPRESSED_FLAG;
107        let new_flag = MessageSysFlag::clear_compressed_flag(flag);
108        assert_eq!(new_flag, 0);
109    }
110
111    #[test]
112    fn get_compression_type_returns_correct_type() {
113        let flag = MessageSysFlag::COMPRESSION_LZ4_TYPE;
114        assert_eq!(
115            MessageSysFlag::get_compression_type(flag).expect("LZ4 flag should decode"),
116            CompressionType::LZ4
117        );
118    }
119
120    #[test]
121    fn try_get_compression_type_rejects_unknown_type_bits() {
122        let flag = MessageSysFlag::COMPRESSED_FLAG | (0x7 << 8);
123        let error = MessageSysFlag::try_get_compression_type(flag).expect_err("unknown compression bits should error");
124
125        assert_eq!(error.descriptor(), &rocketmq_error::CORE_SERIALIZATION_FAILED);
126    }
127
128    #[test]
129    fn check_returns_true_when_flag_is_set() {
130        let flag = MessageSysFlag::COMPRESSED_FLAG;
131        assert!(MessageSysFlag::check(flag, MessageSysFlag::COMPRESSED_FLAG));
132    }
133
134    #[test]
135    fn check_returns_false_when_flag_is_not_set() {
136        let flag = 0;
137        assert!(!MessageSysFlag::check(flag, MessageSysFlag::COMPRESSED_FLAG));
138    }
139}
rocketmq_model/common/sys_flag/message_sys_flag.rs

rocketmq_model/common/sys_flag/
message_sys_flag.rs
