rocketmq_common/common/sys_flag/
message_sys_flag.rs

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

rocketmq_common/common/sys_flag/
message_sys_flag.rs
