rocketmq-common 0.9.0

Rust implementation of Apache rocketmq common
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125

use crate::common::compression::compression_type::CompressionType;
use rocketmq_error::RocketMQResult;

// Meaning of each bit in the system flag
///
/// | bit    | 7 | 6 | 5         | 4        | 3           | 2                | 1                | 0                |
/// |--------|---|---|-----------|----------|-------------|------------------|------------------|------------------|
/// | byte 1 |   |   | STOREHOST | BORNHOST | TRANSACTION | TRANSACTION      | MULTI_TAGS       | COMPRESSED       |
/// | byte 2 |   |   |           |          |             | COMPRESSION_TYPE | COMPRESSION_TYPE | COMPRESSION_TYPE |
/// | byte 3 |   |   |           |          |             |                  |                  |                  |
/// | byte 4 |   |   |           |          |             |                  |                  |
pub struct MessageSysFlag;

impl MessageSysFlag {
    pub const COMPRESSED_FLAG: i32 = 0x1;
    pub const MULTI_TAGS_FLAG: i32 = 0x1 << 1;

    //Transaction related flag
    pub const TRANSACTION_NOT_TYPE: i32 = 0;
    pub const TRANSACTION_PREPARED_TYPE: i32 = 0x1 << 2;
    pub const TRANSACTION_COMMIT_TYPE: i32 = 0x2 << 2;
    pub const TRANSACTION_ROLLBACK_TYPE: i32 = 0x3 << 2;

    //Flag of the message properties
    pub const BORNHOST_V6_FLAG: i32 = 0x1 << 4;
    pub const STOREHOSTADDRESS_V6_FLAG: i32 = 0x1 << 5;

    //Mark the flag for batch to avoid conflict
    pub const NEED_UNWRAP_FLAG: i32 = 0x1 << 6;
    pub const INNER_BATCH_FLAG: i32 = 0x1 << 7;

    // COMPRESSION_TYPE
    pub const COMPRESSION_LZ4_TYPE: i32 = 0x1 << 8;
    pub const COMPRESSION_ZSTD_TYPE: i32 = 0x2 << 8;
    pub const COMPRESSION_ZLIB_TYPE: i32 = 0x3 << 8;
    pub const COMPRESSION_TYPE_COMPARATOR: i32 = 0x7 << 8;

    #[inline]
    pub const fn get_transaction_value(flag: i32) -> i32 {
        flag & Self::TRANSACTION_ROLLBACK_TYPE
    }

    #[inline]
    pub const fn reset_transaction_value(flag: i32, transaction_type: i32) -> i32 {
        (flag & !Self::TRANSACTION_ROLLBACK_TYPE) | transaction_type
    }

    #[inline]
    pub const fn clear_compressed_flag(flag: i32) -> i32 {
        flag & !Self::COMPRESSED_FLAG
    }

    #[inline]
    pub fn get_compression_type(flag: i32) -> RocketMQResult<CompressionType> {
        let compression_type_value = (flag & Self::COMPRESSION_TYPE_COMPARATOR) >> 8;
        CompressionType::find_by_value(compression_type_value)
    }

    #[inline]
    pub fn try_get_compression_type(flag: i32) -> RocketMQResult<CompressionType> {
        Self::get_compression_type(flag)
    }

    #[inline]
    pub const fn check(flag: i32, expected_flag: i32) -> bool {
        (flag & expected_flag) != 0
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn get_transaction_value_returns_correct_value() {
        let flag = MessageSysFlag::TRANSACTION_COMMIT_TYPE;
        assert_eq!(
            MessageSysFlag::get_transaction_value(flag),
            MessageSysFlag::TRANSACTION_COMMIT_TYPE
        );
    }

    #[test]
    fn reset_transaction_value_resets_correctly() {
        let flag = MessageSysFlag::TRANSACTION_COMMIT_TYPE;
        let new_flag = MessageSysFlag::reset_transaction_value(flag, MessageSysFlag::TRANSACTION_ROLLBACK_TYPE);
        assert_eq!(new_flag, MessageSysFlag::TRANSACTION_ROLLBACK_TYPE);
    }

    #[test]
    fn clear_compressed_flag_clears_correctly() {
        let flag = MessageSysFlag::COMPRESSED_FLAG;
        let new_flag = MessageSysFlag::clear_compressed_flag(flag);
        assert_eq!(new_flag, 0);
    }

    #[test]
    fn get_compression_type_returns_correct_type() {
        let flag = MessageSysFlag::COMPRESSION_LZ4_TYPE;
        assert_eq!(
            MessageSysFlag::get_compression_type(flag).expect("LZ4 flag should decode"),
            CompressionType::LZ4
        );
    }

    #[test]
    fn try_get_compression_type_rejects_unknown_type_bits() {
        let flag = MessageSysFlag::COMPRESSED_FLAG | (0x7 << 8);
        let error = MessageSysFlag::try_get_compression_type(flag).expect_err("unknown compression bits should error");

        assert!(error.to_string().contains("unknown compression type value: 7"));
    }

    #[test]
    fn check_returns_true_when_flag_is_set() {
        let flag = MessageSysFlag::COMPRESSED_FLAG;
        assert!(MessageSysFlag::check(flag, MessageSysFlag::COMPRESSED_FLAG));
    }

    #[test]
    fn check_returns_false_when_flag_is_not_set() {
        let flag = 0;
        assert!(!MessageSysFlag::check(flag, MessageSysFlag::COMPRESSED_FLAG));
    }
}