rocketmq_protocol/protocol/body/
process_queue_info.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 serde::Deserialize;
16use serde::Serialize;
17
18#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize)]
19#[serde(rename_all = "camelCase")]
20pub struct ProcessQueueInfo {
21    pub commit_offset: i64,
22    pub cached_msg_min_offset: u64,
23    pub cached_msg_max_offset: u64,
24    pub cached_msg_count: u32,
25    pub cached_msg_size_in_mib: u32,
26
27    pub transaction_msg_min_offset: u64,
28    pub transaction_msg_max_offset: u64,
29    pub transaction_msg_count: u32,
30
31    pub locked: bool,
32    pub try_unlock_times: u64,
33    pub last_lock_timestamp: u64,
34
35    pub droped: bool,
36    pub last_pull_timestamp: u64,
37    pub last_consume_timestamp: u64,
38}
39
40impl std::fmt::Display for ProcessQueueInfo {
41    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
42        write!(
43            f,
44            "ProcessQueueInfo [commit_offset: {}, cached_msg_min_offset: {}, cached_msg_max_offset: {}, \
45             cached_msg_count: {}, cached_msg_size_in_mib: {}, transaction_msg_min_offset: {}, \
46             transaction_msg_max_offset: {}, transaction_msg_count: {}, locked: {}, try_unlock_times: {}, \
47             last_lock_timestamp: {}, droped: {}, last_pull_timestamp: {}, last_consume_timestamp: {}]",
48            self.commit_offset,
49            self.cached_msg_min_offset,
50            self.cached_msg_max_offset,
51            self.cached_msg_count,
52            self.cached_msg_size_in_mib,
53            self.transaction_msg_min_offset,
54            self.transaction_msg_max_offset,
55            self.transaction_msg_count,
56            self.locked,
57            self.try_unlock_times,
58            self.last_lock_timestamp,
59            self.droped,
60            self.last_pull_timestamp,
61            self.last_consume_timestamp
62        )
63    }
64}
65
66#[cfg(test)]
67mod tests {
68    use super::*;
69
70    #[test]
71    fn process_queue_display() {
72        let process_queue_info = ProcessQueueInfo {
73            commit_offset: 100,
74            cached_msg_min_offset: 90,
75            cached_msg_max_offset: 110,
76            cached_msg_count: 20,
77            cached_msg_size_in_mib: 5,
78            transaction_msg_min_offset: 80,
79            transaction_msg_max_offset: 120,
80            transaction_msg_count: 40,
81            locked: true,
82            try_unlock_times: 3,
83            last_lock_timestamp: 1620000000,
84            droped: false,
85            last_pull_timestamp: 1620000100,
86            last_consume_timestamp: 1620000200,
87        };
88
89        let display_output = format!("{}", process_queue_info);
90        assert!(display_output.contains("ProcessQueueInfo"));
91        assert!(display_output.contains("commit_offset: 100"));
92        assert!(display_output.contains("cached_msg_min_offset: 90"));
93        assert!(display_output.contains("cached_msg_max_offset: 110"));
94        assert!(display_output.contains("cached_msg_count: 20"));
95        assert!(display_output.contains("cached_msg_size_in_mib: 5"));
96        assert!(display_output.contains("transaction_msg_min_offset: 80"));
97        assert!(display_output.contains("transaction_msg_max_offset: 120"));
98        assert!(display_output.contains("transaction_msg_count: 40"));
99        assert!(display_output.contains("locked: true"));
100        assert!(display_output.contains("try_unlock_times: 3"));
101        assert!(display_output.contains("last_lock_timestamp: 1620000000"));
102        assert!(display_output.contains("droped: false"));
103        assert!(display_output.contains("last_pull_timestamp: 1620000100"));
104        assert!(display_output.contains("last_consume_timestamp: 1620000200"));
105    }
106}
rocketmq_protocol/protocol/body/process_queue_info.rs

rocketmq_protocol/protocol/body/
process_queue_info.rs
