rocketmq_client_rust/admin/
default_mq_admin_ext.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 std::ops::Deref;
16use std::ops::DerefMut;
17use std::sync::Arc;
18use std::time::Duration;
19
20use cheetah_string::CheetahString;
21use rocketmq_remoting::runtime::RPCHook;
22use rocketmq_rust::ArcMut;
23
24use crate::admin::default_mq_admin_ext_impl::DefaultMQAdminExtImpl;
25use crate::admin::mq_admin_ext_async::MQAdminExt;
26use crate::base::client_config::ClientConfig;
27
28const ADMIN_EXT_GROUP: &str = "admin_ext_group";
29const DEFAULT_TIMEOUT_MILLIS: u64 = 5000;
30
31/// Java-style Admin facade that owns a self-wired `DefaultMQAdminExtImpl`.
32///
33/// `DefaultMQAdminExtImpl` remains the concrete implementation of the
34/// `MQAdminExt` trait. This facade provides the Java-like no-argument
35/// constructors and lifecycle methods while dereferencing to the implementation
36/// for advanced admin operations.
37pub struct DefaultMQAdminExt {
38    client_config: ArcMut<ClientConfig>,
39    default_mqadmin_ext_impl: ArcMut<DefaultMQAdminExtImpl>,
40}
41
42impl DefaultMQAdminExt {
43    fn build(
44        client_config: ArcMut<ClientConfig>,
45        admin_ext_group: CheetahString,
46        timeout_millis: Duration,
47        rpc_hook: Option<Arc<dyn RPCHook>>,
48    ) -> Self {
49        let mut default_mqadmin_ext_impl = ArcMut::new(DefaultMQAdminExtImpl::new(
50            rpc_hook,
51            timeout_millis,
52            client_config.clone(),
53            admin_ext_group,
54        ));
55        let inner = default_mqadmin_ext_impl.clone();
56        default_mqadmin_ext_impl.set_inner(inner);
57
58        Self {
59            client_config,
60            default_mqadmin_ext_impl,
61        }
62    }
63
64    pub fn new() -> Self {
65        Self::with_admin_ext_group(ADMIN_EXT_GROUP)
66    }
67
68    pub fn with_timeout(timeout_millis: Duration) -> Self {
69        let client_config = ArcMut::new(ClientConfig::new());
70        Self::build(
71            client_config,
72            CheetahString::from_static_str(ADMIN_EXT_GROUP),
73            timeout_millis,
74            None,
75        )
76    }
77
78    pub fn with_rpc_hook(rpc_hook: Arc<dyn RPCHook>) -> Self {
79        Self::with_rpc_hook_and_timeout(rpc_hook, Duration::from_millis(DEFAULT_TIMEOUT_MILLIS))
80    }
81
82    pub fn with_rpc_hook_and_timeout(rpc_hook: Arc<dyn RPCHook>, timeout_millis: Duration) -> Self {
83        let client_config = ArcMut::new(ClientConfig::new());
84        Self::build(
85            client_config,
86            CheetahString::from_static_str(ADMIN_EXT_GROUP),
87            timeout_millis,
88            Some(rpc_hook),
89        )
90    }
91
92    pub fn with_admin_ext_group(admin_ext_group: impl Into<CheetahString>) -> Self {
93        let client_config = ArcMut::new(ClientConfig::new());
94        Self::build(
95            client_config,
96            admin_ext_group.into(),
97            Duration::from_millis(DEFAULT_TIMEOUT_MILLIS),
98            None,
99        )
100    }
101
102    pub fn with_admin_ext_group_and_timeout(
103        admin_ext_group: impl Into<CheetahString>,
104        timeout_millis: Duration,
105    ) -> Self {
106        let client_config = ArcMut::new(ClientConfig::new());
107        Self::build(client_config, admin_ext_group.into(), timeout_millis, None)
108    }
109
110    pub fn with_admin_ext_group_and_rpc_hook(
111        admin_ext_group: impl Into<CheetahString>,
112        rpc_hook: Arc<dyn RPCHook>,
113    ) -> Self {
114        let client_config = ArcMut::new(ClientConfig::new());
115        Self::build(
116            client_config,
117            admin_ext_group.into(),
118            Duration::from_millis(DEFAULT_TIMEOUT_MILLIS),
119            Some(rpc_hook),
120        )
121    }
122
123    pub fn set_namesrv_addr(&mut self, name_serv_addr: impl Into<CheetahString>) {
124        self.client_config.set_namesrv_addr(name_serv_addr.into());
125    }
126
127    #[inline]
128    pub fn is_use_tls(&self) -> bool {
129        self.client_config.is_use_tls()
130    }
131
132    #[inline]
133    pub fn set_use_tls(&mut self, use_tls: bool) {
134        self.client_config.set_use_tls(use_tls);
135        self.default_mqadmin_ext_impl.set_use_tls(use_tls);
136    }
137
138    #[inline]
139    pub fn client_config(&self) -> &ArcMut<ClientConfig> {
140        &self.client_config
141    }
142
143    #[inline]
144    pub fn client_config_mut(&mut self) -> &mut ArcMut<ClientConfig> {
145        &mut self.client_config
146    }
147
148    #[inline]
149    pub fn inner(&self) -> &ArcMut<DefaultMQAdminExtImpl> {
150        &self.default_mqadmin_ext_impl
151    }
152
153    #[inline]
154    pub fn inner_mut(&mut self) -> &mut ArcMut<DefaultMQAdminExtImpl> {
155        &mut self.default_mqadmin_ext_impl
156    }
157
158    #[inline]
159    pub fn has_inner(&self) -> bool {
160        self.default_mqadmin_ext_impl.has_inner()
161    }
162
163    pub async fn start(&mut self) -> rocketmq_error::RocketMQResult<()> {
164        MQAdminExt::start(self.default_mqadmin_ext_impl.as_mut()).await
165    }
166
167    pub async fn shutdown(&mut self) {
168        MQAdminExt::shutdown(self.default_mqadmin_ext_impl.as_mut()).await;
169    }
170}
171
172impl Default for DefaultMQAdminExt {
173    fn default() -> Self {
174        Self::new()
175    }
176}
177
178impl AsRef<DefaultMQAdminExtImpl> for DefaultMQAdminExt {
179    fn as_ref(&self) -> &DefaultMQAdminExtImpl {
180        self.default_mqadmin_ext_impl.as_ref()
181    }
182}
183
184impl AsMut<DefaultMQAdminExtImpl> for DefaultMQAdminExt {
185    fn as_mut(&mut self) -> &mut DefaultMQAdminExtImpl {
186        self.default_mqadmin_ext_impl.as_mut()
187    }
188}
189
190impl Deref for DefaultMQAdminExt {
191    type Target = DefaultMQAdminExtImpl;
192
193    fn deref(&self) -> &Self::Target {
194        self.default_mqadmin_ext_impl.as_ref()
195    }
196}
197
198impl DerefMut for DefaultMQAdminExt {
199    fn deref_mut(&mut self) -> &mut Self::Target {
200        self.default_mqadmin_ext_impl.as_mut()
201    }
202}
203
204#[cfg(test)]
205mod tests {
206    use std::time::Duration;
207
208    use cheetah_string::CheetahString;
209    use rocketmq_error::RocketMQError;
210
211    use crate::admin::mq_admin_ext_async::MQAdminExt;
212
213    use super::DefaultMQAdminExt;
214
215    #[test]
216    fn constructors_initialize_self_wired_inner_impl() {
217        let default_admin = DefaultMQAdminExt::new();
218        assert!(default_admin.has_inner());
219        assert!(default_admin.inner().has_inner());
220
221        let timed_admin = DefaultMQAdminExt::with_timeout(Duration::from_secs(3));
222        assert!(timed_admin.has_inner());
223
224        let grouped_admin = DefaultMQAdminExt::with_admin_ext_group("admin-public-api-test");
225        assert!(grouped_admin.has_inner());
226
227        let grouped_timed_admin =
228            DefaultMQAdminExt::with_admin_ext_group_and_timeout("admin-public-api-test", Duration::from_secs(3));
229        assert!(grouped_timed_admin.has_inner());
230    }
231
232    #[test]
233    fn namesrv_addr_updates_shared_client_config_before_start() {
234        let mut admin = DefaultMQAdminExt::new();
235        admin.set_namesrv_addr("127.0.0.1:9876");
236
237        assert_eq!(
238            admin.client_config().get_namesrv_addr().as_deref(),
239            Some("127.0.0.1:9876")
240        );
241    }
242
243    #[test]
244    fn use_tls_updates_shared_admin_client_config_before_start() {
245        let mut admin = DefaultMQAdminExt::new();
246
247        assert!(!admin.is_use_tls());
248        assert!(!admin.inner().is_use_tls());
249
250        admin.set_use_tls(true);
251
252        assert!(admin.is_use_tls());
253        assert!(admin.client_config().is_use_tls());
254        assert!(admin.inner().is_use_tls());
255    }
256
257    #[tokio::test]
258    async fn deref_exposes_impl_admin_trait_methods_without_panicking() {
259        let admin = DefaultMQAdminExt::new();
260
261        let error = admin
262            .get_kv_config(CheetahString::from("namespace"), CheetahString::from("key"))
263            .await
264            .expect_err("unstarted admin should return a typed error instead of panicking");
265
266        assert!(matches!(error, RocketMQError::ClientNotStarted));
267    }
268}
rocketmq_client_rust/admin/default_mq_admin_ext.rs

rocketmq_client_rust/admin/
default_mq_admin_ext.rs
