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_transport::api::RPCHook;
22
23use crate::admin::default_mq_admin_ext_impl::DefaultMQAdminExtImpl;
24#[cfg(all(feature = "admin-read", not(feature = "admin-mutation")))]
25use crate::admin::mq_admin_read_ext::MQAdminReadExt;
26use crate::base::client_config::ClientConfig;
27use crate::runtime::ClientRuntime;
28use crate::session::ClientSession;
29use crate::session::ClientSessionProvider;
30
31const ADMIN_EXT_GROUP: &str = "admin_ext_group";
32const DEFAULT_TIMEOUT_MILLIS: u64 = 5000;
33
34/// Java-style Admin facade that owns a self-wired `DefaultMQAdminExtImpl`.
35///
36/// `DefaultMQAdminExtImpl` remains the concrete implementation behind the
37/// scoped administration capabilities. This facade requires an
38/// application-owned [`ClientRuntime`] and dereferences to the implementation
39/// for advanced admin operations.
40pub struct DefaultMQAdminExt {
41    session: ClientSession,
42    default_mqadmin_ext_impl: DefaultMQAdminExtImpl,
43}
44
45impl DefaultMQAdminExt {
46    fn build(
47        client_runtime: Arc<ClientRuntime>,
48        client_config: ClientConfig,
49        admin_ext_group: CheetahString,
50        timeout_millis: Duration,
51        rpc_hook: Option<Arc<dyn RPCHook>>,
52    ) -> Self {
53        let default_mqadmin_ext_impl = DefaultMQAdminExtImpl::new(
54            Arc::clone(&client_runtime),
55            rpc_hook,
56            timeout_millis,
57            client_config,
58            admin_ext_group,
59        );
60
61        Self {
62            session: ClientSession::new(client_runtime),
63            default_mqadmin_ext_impl,
64        }
65    }
66
67    pub fn new(client_runtime: Arc<ClientRuntime>) -> Self {
68        Self::with_admin_ext_group(client_runtime, ADMIN_EXT_GROUP)
69    }
70
71    /// Returns the application-owned runtime borrowed by this Admin facade.
72    pub fn client_runtime(&self) -> Arc<ClientRuntime> {
73        self.session.runtime()
74    }
75
76    pub fn with_timeout(client_runtime: Arc<ClientRuntime>, timeout_millis: Duration) -> Self {
77        let client_config = ClientConfig::new();
78        Self::build(
79            client_runtime,
80            client_config,
81            CheetahString::from_static_str(ADMIN_EXT_GROUP),
82            timeout_millis,
83            None,
84        )
85    }
86
87    pub fn with_rpc_hook(client_runtime: Arc<ClientRuntime>, rpc_hook: Arc<dyn RPCHook>) -> Self {
88        Self::with_rpc_hook_and_timeout(client_runtime, rpc_hook, Duration::from_millis(DEFAULT_TIMEOUT_MILLIS))
89    }
90
91    pub fn with_rpc_hook_and_timeout(
92        client_runtime: Arc<ClientRuntime>,
93        rpc_hook: Arc<dyn RPCHook>,
94        timeout_millis: Duration,
95    ) -> Self {
96        let client_config = ClientConfig::new();
97        Self::build(
98            client_runtime,
99            client_config,
100            CheetahString::from_static_str(ADMIN_EXT_GROUP),
101            timeout_millis,
102            Some(rpc_hook),
103        )
104    }
105
106    pub fn with_admin_ext_group(client_runtime: Arc<ClientRuntime>, admin_ext_group: impl Into<CheetahString>) -> Self {
107        let client_config = ClientConfig::new();
108        Self::build(
109            client_runtime,
110            client_config,
111            admin_ext_group.into(),
112            Duration::from_millis(DEFAULT_TIMEOUT_MILLIS),
113            None,
114        )
115    }
116
117    pub fn with_admin_ext_group_and_timeout(
118        client_runtime: Arc<ClientRuntime>,
119        admin_ext_group: impl Into<CheetahString>,
120        timeout_millis: Duration,
121    ) -> Self {
122        let client_config = ClientConfig::new();
123        Self::build(
124            client_runtime,
125            client_config,
126            admin_ext_group.into(),
127            timeout_millis,
128            None,
129        )
130    }
131
132    pub fn with_admin_ext_group_and_rpc_hook(
133        client_runtime: Arc<ClientRuntime>,
134        admin_ext_group: impl Into<CheetahString>,
135        rpc_hook: Arc<dyn RPCHook>,
136    ) -> Self {
137        let client_config = ClientConfig::new();
138        Self::build(
139            client_runtime,
140            client_config,
141            admin_ext_group.into(),
142            Duration::from_millis(DEFAULT_TIMEOUT_MILLIS),
143            Some(rpc_hook),
144        )
145    }
146
147    pub fn with_admin_ext_group_rpc_hook_and_timeout(
148        client_runtime: Arc<ClientRuntime>,
149        admin_ext_group: impl Into<CheetahString>,
150        rpc_hook: Arc<dyn RPCHook>,
151        timeout_millis: Duration,
152    ) -> Self {
153        let client_config = ClientConfig::new();
154        Self::build(
155            client_runtime,
156            client_config,
157            admin_ext_group.into(),
158            timeout_millis,
159            Some(rpc_hook),
160        )
161    }
162
163    pub fn set_namesrv_addr(&mut self, name_serv_addr: impl Into<CheetahString>) {
164        self.default_mqadmin_ext_impl
165            .client_config_mut()
166            .set_namesrv_addr(name_serv_addr.into());
167    }
168
169    #[inline]
170    pub fn is_use_tls(&self) -> bool {
171        self.default_mqadmin_ext_impl.is_use_tls()
172    }
173
174    #[inline]
175    pub fn set_use_tls(&mut self, use_tls: bool) {
176        self.default_mqadmin_ext_impl.set_use_tls(use_tls);
177    }
178
179    #[inline]
180    pub fn client_config(&self) -> &ClientConfig {
181        self.default_mqadmin_ext_impl.client_config()
182    }
183
184    #[inline]
185    pub fn client_config_mut(&mut self) -> &mut ClientConfig {
186        self.default_mqadmin_ext_impl.client_config_mut()
187    }
188
189    #[inline]
190    pub fn inner(&self) -> &DefaultMQAdminExtImpl {
191        &self.default_mqadmin_ext_impl
192    }
193
194    #[inline]
195    pub fn inner_mut(&mut self) -> &mut DefaultMQAdminExtImpl {
196        &mut self.default_mqadmin_ext_impl
197    }
198
199    #[inline]
200    pub fn has_inner(&self) -> bool {
201        self.default_mqadmin_ext_impl.has_inner()
202    }
203
204    #[cfg(any(feature = "admin-read", feature = "admin-mutation"))]
205    pub async fn start(&mut self) -> crate::ClientResult<()> {
206        #[cfg(feature = "admin-mutation")]
207        {
208            return self.default_mqadmin_ext_impl.start_admin().await;
209        }
210        #[cfg(all(feature = "admin-read", not(feature = "admin-mutation")))]
211        {
212            MQAdminReadExt::start(self).await
213        }
214    }
215
216    #[cfg(any(feature = "admin-read", feature = "admin-mutation"))]
217    pub async fn shutdown(&mut self) {
218        #[cfg(feature = "admin-mutation")]
219        {
220            self.default_mqadmin_ext_impl.shutdown_admin().await;
221        }
222        #[cfg(all(feature = "admin-read", not(feature = "admin-mutation")))]
223        {
224            MQAdminReadExt::shutdown(self).await;
225        }
226    }
227}
228
229impl ClientSessionProvider for DefaultMQAdminExt {
230    fn client_session(&self) -> Option<&ClientSession> {
231        Some(&self.session)
232    }
233}
234
235impl AsRef<DefaultMQAdminExtImpl> for DefaultMQAdminExt {
236    fn as_ref(&self) -> &DefaultMQAdminExtImpl {
237        &self.default_mqadmin_ext_impl
238    }
239}
240
241impl AsMut<DefaultMQAdminExtImpl> for DefaultMQAdminExt {
242    fn as_mut(&mut self) -> &mut DefaultMQAdminExtImpl {
243        &mut self.default_mqadmin_ext_impl
244    }
245}
246
247impl Deref for DefaultMQAdminExt {
248    type Target = DefaultMQAdminExtImpl;
249
250    fn deref(&self) -> &Self::Target {
251        &self.default_mqadmin_ext_impl
252    }
253}
254
255impl DerefMut for DefaultMQAdminExt {
256    fn deref_mut(&mut self) -> &mut Self::Target {
257        &mut self.default_mqadmin_ext_impl
258    }
259}
260
261#[cfg(all(test, feature = "admin-full"))]
262mod tests {
263    use std::time::Duration;
264
265    use cheetah_string::CheetahString;
266
267    use crate::admin::capability::RouteAdmin;
268
269    use super::DefaultMQAdminExt;
270
271    fn test_runtime() -> std::sync::Arc<crate::runtime::ClientRuntime> {
272        crate::runtime::test_client_runtime("default-admin-ext-test")
273    }
274
275    #[test]
276    fn constructors_initialize_owned_inner_impl() {
277        let default_admin = DefaultMQAdminExt::new(test_runtime());
278        assert!(default_admin.has_inner());
279        assert!(default_admin.inner().has_inner());
280
281        let timed_admin = DefaultMQAdminExt::with_timeout(test_runtime(), Duration::from_secs(3));
282        assert!(timed_admin.has_inner());
283
284        let grouped_admin = DefaultMQAdminExt::with_admin_ext_group(test_runtime(), "admin-public-api-test");
285        assert!(grouped_admin.has_inner());
286
287        let grouped_timed_admin = DefaultMQAdminExt::with_admin_ext_group_and_timeout(
288            test_runtime(),
289            "admin-public-api-test",
290            Duration::from_secs(3),
291        );
292        assert!(grouped_timed_admin.has_inner());
293    }
294
295    #[test]
296    fn namesrv_addr_updates_shared_client_config_before_start() {
297        let mut admin = DefaultMQAdminExt::new(test_runtime());
298        admin.set_namesrv_addr("127.0.0.1:9876");
299
300        assert_eq!(
301            admin.client_config().get_namesrv_addr().as_deref(),
302            Some("127.0.0.1:9876")
303        );
304    }
305
306    #[test]
307    fn use_tls_updates_shared_admin_client_config_before_start() {
308        let mut admin = DefaultMQAdminExt::new(test_runtime());
309
310        assert!(!admin.is_use_tls());
311        assert!(!admin.inner().is_use_tls());
312
313        admin.set_use_tls(true);
314
315        assert!(admin.is_use_tls());
316        assert!(admin.client_config().is_use_tls());
317        assert!(admin.inner().is_use_tls());
318    }
319
320    #[tokio::test]
321    async fn deref_exposes_impl_admin_trait_methods_without_panicking() {
322        let admin = DefaultMQAdminExt::new(test_runtime());
323
324        let error = admin
325            .get_kv_config(CheetahString::from("namespace"), CheetahString::from("key"))
326            .await
327            .expect_err("unstarted admin should return a typed error instead of panicking");
328
329        assert!(error.is(&rocketmq_error::CLIENT_LIFECYCLE_NOT_STARTED));
330    }
331}
rocketmq_client_rust/admin/default_mq_admin_ext.rs

rocketmq_client_rust/admin/
default_mq_admin_ext.rs
