rocketmq_client_rust/admin/default_mq_admin_ext_impl/
lifecycle.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::env;
16
17use rocketmq_model::common::FAQUrl;
18use tracing::info;
19
20use super::*;
21
22impl DefaultMQAdminExtImpl {
23    pub fn new(
24        client_runtime: Arc<ClientRuntime>,
25        rpc_hook: Option<Arc<dyn RPCHook>>,
26        timeout_millis: Duration,
27        client_config: ClientConfig,
28        admin_ext_group: CheetahString,
29    ) -> Self {
30        DefaultMQAdminExtImpl {
31            client_pool: client_runtime.pool().clone(),
32            client_pool_token: None,
33            service_state: ServiceState::CreateJust,
34            client_instance: None,
35            rpc_hook,
36            timeout_millis,
37            kv_namespace_to_delete_list: vec![CheetahString::from_static_str(NAMESPACE_ORDER_TOPIC_CONFIG)],
38            client_config,
39            admin_ext_group,
40        }
41    }
42
43    /// Returns whether the facade has a usable concrete implementation.
44    ///
45    /// The implementation is now owned directly, so this compatibility query is always true.
46    pub fn has_inner(&self) -> bool {
47        true
48    }
49
50    #[inline]
51    pub fn client_config(&self) -> &ClientConfig {
52        &self.client_config
53    }
54
55    #[inline]
56    pub fn client_config_mut(&mut self) -> &mut ClientConfig {
57        &mut self.client_config
58    }
59
60    #[inline]
61    pub fn is_use_tls(&self) -> bool {
62        self.client_config.is_use_tls()
63    }
64
65    #[inline]
66    pub fn set_use_tls(&mut self, use_tls: bool) {
67        self.client_config.set_use_tls(use_tls);
68    }
69
70    pub(in crate::admin) fn mq_client_api(&self) -> crate::ClientResult<Arc<MQClientAPIImpl>> {
71        self.client_instance
72            .as_ref()
73            .ok_or(crate::ClientError::not_started())?
74            .get_mq_client_api_impl()
75    }
76
77    pub(in crate::admin) fn remoting_timeout_millis(&self) -> crate::ClientResult<u64> {
78        u64::try_from(self.timeout_millis.as_millis()).map_err(|_| {
79            crate::ClientError::illegal_argument("admin timeout exceeds the supported u64 millisecond range")
80        })
81    }
82
83    pub(in crate::admin) async fn start_admin(&mut self) -> crate::ClientResult<()> {
84        match self.service_state {
85            ServiceState::CreateJust => {
86                self.service_state = ServiceState::StartFailed;
87                self.client_config.change_instance_name_to_pid();
88                if "{}".eq(&self.client_config.socks_proxy_config) {
89                    self.client_config.socks_proxy_config =
90                        env::var(SOCKS_PROXY_JSON).unwrap_or_else(|_| "{}".to_string()).into();
91                }
92                let pooled = self
93                    .client_pool
94                    .get_or_create(self.client_config.clone(), self.rpc_hook.clone())?;
95                let (client_instance, token) = pooled.into_parts();
96                self.client_instance = Some(client_instance);
97                self.client_pool_token = Some(token);
98
99                let group = self.admin_ext_group.clone();
100                let register_ok = self
101                    .client_instance
102                    .as_mut()
103                    .ok_or(crate::ClientError::not_started())?
104                    .register_admin_ext(&group)
105                    .await;
106                if !register_ok {
107                    if let Some(token) = self.client_pool_token.take() {
108                        self.client_pool.release(token).await;
109                    }
110                    self.service_state = ServiceState::StartFailed;
111                    return Err(crate::ClientError::illegal_argument(format!(
112                        "The adminExt group[{}] has created already, specified another name please.{}",
113                        self.admin_ext_group,
114                        FAQUrl::suggest_todo(FAQUrl::GROUP_NAME_DUPLICATE_URL)
115                    )));
116                }
117                if let Err(error) = self
118                    .client_instance
119                    .as_mut()
120                    .ok_or(crate::ClientError::not_started())?
121                    .start()
122                    .await
123                {
124                    if let Some(token) = self.client_pool_token.take() {
125                        self.client_pool.release(token).await;
126                    }
127                    return Err(error);
128                }
129                self.service_state = ServiceState::Running;
130                info!("the adminExt [{}] start OK", self.admin_ext_group);
131                Ok(())
132            }
133            ServiceState::Running | ServiceState::ShutdownAlready | ServiceState::StartFailed => {
134                Err(crate::ClientError::already_started())
135            }
136        }
137    }
138
139    pub(in crate::admin) async fn shutdown_admin(&mut self) {
140        match self.service_state {
141            ServiceState::CreateJust | ServiceState::ShutdownAlready | ServiceState::StartFailed => {}
142            ServiceState::Running => {
143                if let Some(instance) = self.client_instance.as_mut() {
144                    instance.unregister_admin_ext(&self.admin_ext_group).await;
145                }
146                if let Some(token) = self.client_pool_token.take() {
147                    self.client_pool.release(token).await;
148                }
149                self.service_state = ServiceState::ShutdownAlready;
150            }
151        }
152    }
153}
rocketmq_client_rust/admin/default_mq_admin_ext_impl/lifecycle.rs

rocketmq_client_rust/admin/default_mq_admin_ext_impl/
lifecycle.rs
