rocketmq_common/common/sys_flag/
pull_sys_flag.rs

1/*
2 * Licensed to the Apache Software Foundation (ASF) under one or more
3 * contributor license agreements.  See the NOTICE file distributed with
4 * this work for additional information regarding copyright ownership.
5 * The ASF licenses this file to You under the Apache License, Version 2.0
6 * (the "License"); you may not use this file except in compliance with
7 * the License.  You may obtain a copy of the License at
8 *
9 *     http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 */
17pub struct PullSysFlag;
18
19impl PullSysFlag {
20    const FLAG_COMMIT_OFFSET: u32 = 0x1;
21    const FLAG_SUSPEND: u32 = 0x1 << 1;
22    const FLAG_SUBSCRIPTION: u32 = 0x1 << 2;
23    const FLAG_CLASS_FILTER: u32 = 0x1 << 3;
24    const FLAG_LITE_PULL_MESSAGE: u32 = 0x1 << 4;
25
26    pub fn build_sys_flag(
27        commit_offset: bool,
28        suspend: bool,
29        subscription: bool,
30        class_filter: bool,
31    ) -> u32 {
32        let mut flag = 0;
33
34        if commit_offset {
35            flag |= Self::FLAG_COMMIT_OFFSET;
36        }
37
38        if suspend {
39            flag |= Self::FLAG_SUSPEND;
40        }
41
42        if subscription {
43            flag |= Self::FLAG_SUBSCRIPTION;
44        }
45
46        if class_filter {
47            flag |= Self::FLAG_CLASS_FILTER;
48        }
49
50        flag
51    }
52
53    pub fn build_sys_flag_with_lite_pull(
54        commit_offset: bool,
55        suspend: bool,
56        subscription: bool,
57        class_filter: bool,
58        lite_pull: bool,
59    ) -> u32 {
60        let mut flag = Self::build_sys_flag(commit_offset, suspend, subscription, class_filter);
61
62        if lite_pull {
63            flag |= Self::FLAG_LITE_PULL_MESSAGE;
64        }
65
66        flag
67    }
68
69    pub fn clear_commit_offset_flag(sys_flag: u32) -> u32 {
70        sys_flag & (!Self::FLAG_COMMIT_OFFSET)
71    }
72
73    pub fn has_commit_offset_flag(sys_flag: u32) -> bool {
74        (sys_flag & Self::FLAG_COMMIT_OFFSET) == Self::FLAG_COMMIT_OFFSET
75    }
76
77    pub fn has_suspend_flag(sys_flag: u32) -> bool {
78        (sys_flag & Self::FLAG_SUSPEND) == Self::FLAG_SUSPEND
79    }
80
81    pub fn clear_suspend_flag(sys_flag: u32) -> u32 {
82        sys_flag & (!Self::FLAG_SUSPEND)
83    }
84
85    pub fn has_subscription_flag(sys_flag: u32) -> bool {
86        (sys_flag & Self::FLAG_SUBSCRIPTION) == Self::FLAG_SUBSCRIPTION
87    }
88
89    pub fn build_sys_flag_with_subscription(sys_flag: u32) -> u32 {
90        sys_flag | Self::FLAG_SUBSCRIPTION
91    }
92
93    pub fn has_class_filter_flag(sys_flag: u32) -> bool {
94        (sys_flag & Self::FLAG_CLASS_FILTER) == Self::FLAG_CLASS_FILTER
95    }
96
97    pub fn has_lite_pull_flag(sys_flag: u32) -> bool {
98        (sys_flag & Self::FLAG_LITE_PULL_MESSAGE) == Self::FLAG_LITE_PULL_MESSAGE
99    }
100}
101
102#[cfg(test)]
103mod tests {
104    use super::*;
105
106    #[test]
107    fn build_sys_flag_sets_correct_flags() {
108        let flag = PullSysFlag::build_sys_flag(true, true, true, true);
109        assert_eq!(flag, 0b1111);
110    }
111
112    #[test]
113    fn build_sys_flag_with_lite_pull_sets_correct_flags() {
114        let flag = PullSysFlag::build_sys_flag_with_lite_pull(true, true, true, true, true);
115        assert_eq!(flag, 0b11111);
116    }
117
118    #[test]
119    fn clear_commit_offset_flag_clears_correct_flag() {
120        let flag = PullSysFlag::clear_commit_offset_flag(0b1111);
121        assert_eq!(flag, 0b1110);
122    }
123
124    #[test]
125    fn has_commit_offset_flag_returns_correct_result() {
126        assert!(PullSysFlag::has_commit_offset_flag(0b1));
127        assert!(!PullSysFlag::has_commit_offset_flag(0b10));
128    }
129
130    #[test]
131    fn has_suspend_flag_returns_correct_result() {
132        assert!(PullSysFlag::has_suspend_flag(0b10));
133        assert!(!PullSysFlag::has_suspend_flag(0b1));
134    }
135
136    #[test]
137    fn clear_suspend_flag_clears_correct_flag() {
138        let flag = PullSysFlag::clear_suspend_flag(0b1111);
139        assert_eq!(flag, 0b1101);
140    }
141
142    #[test]
143    fn has_subscription_flag_returns_correct_result() {
144        assert!(PullSysFlag::has_subscription_flag(0b100));
145        assert!(!PullSysFlag::has_subscription_flag(0b1));
146    }
147
148    #[test]
149    fn build_sys_flag_with_subscription_sets_correct_flag() {
150        let flag = PullSysFlag::build_sys_flag_with_subscription(0b1);
151        assert_eq!(flag, 0b101);
152    }
153
154    #[test]
155    fn has_class_filter_flag_returns_correct_result() {
156        assert!(PullSysFlag::has_class_filter_flag(0b1000));
157        assert!(!PullSysFlag::has_class_filter_flag(0b1));
158    }
159
160    #[test]
161    fn has_lite_pull_flag_returns_correct_result() {
162        assert!(PullSysFlag::has_lite_pull_flag(0b10000));
163        assert!(!PullSysFlag::has_lite_pull_flag(0b1));
164    }
165}
rocketmq_common/common/sys_flag/pull_sys_flag.rs

rocketmq_common/common/sys_flag/
pull_sys_flag.rs
