rocketmq_client_rust/admin/
mq_admin_consumer_observation_read_ext.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
15//! Exact-Broker, read-only Consumer observation capability.
16
17use crate::ClientError;
18use cheetah_string::CheetahString;
19use rocketmq_protocol::code::response_code::ResponseCode;
20use rocketmq_protocol::protocol::admin::consume_stats::ConsumeStats;
21use rocketmq_protocol::protocol::body::consumer_connection::ConsumerConnection;
22
23use super::default_mq_admin_ext::DefaultMQAdminExt;
24use super::mq_admin_read_ext::MQAdminReadExt;
25use super::mq_admin_read_ext::SubscriptionGroupConfigVersioned;
26
27#[derive(Debug, Clone)]
28pub enum ConsumerGroupConfigRead {
29    Present(Box<SubscriptionGroupConfigVersioned>),
30    Absent,
31}
32
33#[derive(Debug, Clone)]
34pub enum ConsumerConnectionRead {
35    Online(ConsumerConnection),
36    Offline,
37}
38
39#[derive(Debug)]
40pub enum ConsumerProgressRead {
41    Observed(ConsumeStats),
42    Absent,
43}
44
45#[allow(async_fn_in_trait)]
46pub trait MQAdminConsumerObservationReadExt: Send {
47    async fn consumer_group_config_at(
48        &self,
49        broker_addr: CheetahString,
50        consumer_group: CheetahString,
51    ) -> crate::ClientResult<ConsumerGroupConfigRead>;
52
53    async fn consumer_connection_at(
54        &self,
55        broker_addr: CheetahString,
56        consumer_group: CheetahString,
57    ) -> crate::ClientResult<ConsumerConnectionRead>;
58
59    async fn consumer_progress_at(
60        &self,
61        broker_addr: CheetahString,
62        consumer_group: CheetahString,
63    ) -> crate::ClientResult<ConsumerProgressRead>;
64}
65
66impl MQAdminConsumerObservationReadExt for DefaultMQAdminExt {
67    async fn consumer_group_config_at(
68        &self,
69        broker_addr: CheetahString,
70        consumer_group: CheetahString,
71    ) -> crate::ClientResult<ConsumerGroupConfigRead> {
72        match MQAdminReadExt::subscription_group_config_with_version(self, broker_addr, consumer_group).await {
73            Ok(config) => Ok(ConsumerGroupConfigRead::Present(Box::new(config))),
74            Err(error) if broker_response_is(&error, ResponseCode::SubscriptionGroupNotExist) => {
75                Ok(ConsumerGroupConfigRead::Absent)
76            }
77            Err(error) => Err(error),
78        }
79    }
80
81    async fn consumer_connection_at(
82        &self,
83        broker_addr: CheetahString,
84        consumer_group: CheetahString,
85    ) -> crate::ClientResult<ConsumerConnectionRead> {
86        match MQAdminReadExt::observe_consumer_connection_at(self, consumer_group, broker_addr).await {
87            Ok(connection) if connection.get_connection_set().is_empty() => Ok(ConsumerConnectionRead::Offline),
88            Ok(connection) => Ok(ConsumerConnectionRead::Online(connection)),
89            Err(error) if broker_response_is(&error, ResponseCode::ConsumerNotOnline) => {
90                Ok(ConsumerConnectionRead::Offline)
91            }
92            Err(error) => Err(error),
93        }
94    }
95
96    async fn consumer_progress_at(
97        &self,
98        broker_addr: CheetahString,
99        consumer_group: CheetahString,
100    ) -> crate::ClientResult<ConsumerProgressRead> {
101        match MQAdminReadExt::examine_consume_stats(self, consumer_group, None, None, Some(broker_addr), None).await {
102            Ok(stats) => Ok(ConsumerProgressRead::Observed(stats)),
103            Err(error) if broker_response_is(&error, ResponseCode::SubscriptionGroupNotExist) => {
104                Ok(ConsumerProgressRead::Absent)
105            }
106            Err(error) => Err(error),
107        }
108    }
109}
110
111fn broker_response_is(error: &ClientError, expected: ResponseCode) -> bool {
112    error.broker_response_code() == Some(expected.to_i32())
113}
114
115#[cfg(test)]
116mod tests {
117    use super::*;
118
119    #[test]
120    fn only_exact_broker_response_codes_are_closed_states() {
121        let missing = ClientError::broker_operation_failed(
122            "BROKER_OPERATION",
123            ResponseCode::SubscriptionGroupNotExist.to_i32(),
124            "must not be inspected",
125        );
126        let offline = ClientError::broker_operation_failed(
127            "BROKER_OPERATION",
128            ResponseCode::ConsumerNotOnline.to_i32(),
129            "must not be inspected",
130        );
131        let unavailable = ClientError::broker_operation_failed(
132            "BROKER_OPERATION",
133            ResponseCode::SystemError.to_i32(),
134            "consumer not online",
135        );
136
137        assert!(broker_response_is(&missing, ResponseCode::SubscriptionGroupNotExist));
138        assert!(broker_response_is(&offline, ResponseCode::ConsumerNotOnline));
139        assert!(!broker_response_is(&unavailable, ResponseCode::ConsumerNotOnline));
140    }
141}
rocketmq_client_rust/admin/mq_admin_consumer_observation_read_ext.rs

rocketmq_client_rust/admin/
mq_admin_consumer_observation_read_ext.rs
