rocketmq_client_rust/implementation/mq_client_api_impl/
route.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 super::route_error::route_lookup_error;
16use super::*;
17
18pub struct RouteClient<'a> {
19    api: &'a MQClientAPIImpl,
20}
21
22impl RouteClient<'_> {
23    pub async fn topic_route_info(
24        &self,
25        topic: &CheetahString,
26        timeout_millis: u64,
27    ) -> crate::ClientResult<Option<TopicRouteData>> {
28        self.api
29            .get_topic_route_info_from_name_server(topic, timeout_millis)
30            .await
31    }
32}
33
34impl MQClientAPIImpl {
35    #[must_use]
36    pub fn route_client(&self) -> RouteClient<'_> {
37        RouteClient { api: self }
38    }
39}
40
41impl MQClientAPIImpl {
42    #[inline]
43    pub async fn get_default_topic_route_info_from_name_server(
44        &self,
45        timeout_millis: u64,
46    ) -> crate::ClientResult<Option<TopicRouteData>> {
47        self.get_topic_route_info_from_name_server_detail(
48            TopicValidator::AUTO_CREATE_TOPIC_KEY_TOPIC,
49            timeout_millis,
50            false,
51        )
52        .await
53    }
54
55    #[inline]
56    pub async fn get_topic_route_info_from_name_server(
57        &self,
58        topic: &str,
59        timeout_millis: u64,
60    ) -> crate::ClientResult<Option<TopicRouteData>> {
61        self.get_topic_route_info_from_name_server_detail(topic, timeout_millis, true)
62            .await
63    }
64
65    #[inline]
66    pub async fn get_topic_route_info_from_name_server_detail(
67        &self,
68        topic: &str,
69        timeout_millis: u64,
70        allow_topic_not_exist: bool,
71    ) -> crate::ClientResult<Option<TopicRouteData>> {
72        let deadline = RequestDeadline::from_timeout_millis(timeout_millis);
73        self.get_topic_route_info_once(topic, deadline, allow_topic_not_exist)
74            .await
75            .map_err(route_lookup_error)
76    }
77
78    pub(crate) async fn get_topic_route_info_once(
79        &self,
80        topic: &str,
81        deadline: RequestDeadline,
82        allow_topic_not_exist: bool,
83    ) -> Result<Option<TopicRouteData>, RetryInput> {
84        let request_header = GetRouteInfoRequestHeader {
85            topic: CheetahString::from_slice(topic),
86            accept_standard_json_only: None,
87            topic_request_header: None,
88        };
89        let request = self.create_request_command(RequestCode::GetRouteinfoByTopic, request_header);
90        let outcome = self
91            .remoting_client
92            .invoke_request_with_deadline(None, request, deadline)
93            .await
94            .map_err(RetryInput::Transport)?;
95        match outcome {
96            OutboundRequestOutcome::Response(mut result) => {
97                let code = result.code();
98                let response_code = ResponseCode::from(code);
99                match response_code {
100                    ResponseCode::Success => {
101                        let body = result.take_body();
102                        if let Some(body_inner) = body {
103                            let route_data = TopicRouteData::decode(body_inner.as_ref()).map_err(RetryInput::from)?;
104                            return Ok(Some(route_data));
105                        }
106                    }
107                    ResponseCode::TopicNotExist => {
108                        if allow_topic_not_exist {
109                            warn!("get Topic [{}] RouteInfoFromNameServer is not exist value", topic);
110                            return Ok(None);
111                        }
112                    }
113                    _ => {
114                        return Err(RetryInput::Response {
115                            code,
116                            retry_after: None,
117                            terminal_error: mq_client_err!(
118                                code,
119                                result.remark().cloned().unwrap_or_default().to_string()
120                            ),
121                        });
122                    }
123                }
124                Err(RetryInput::Response {
125                    code,
126                    retry_after: None,
127                    terminal_error: mq_client_err!(code, result.remark().cloned().unwrap_or_default().to_string()),
128                })
129            }
130            OutboundRequestOutcome::Rejected(rejection) => Err(RetryInput::Rejected(rejection)),
131            OutboundRequestOutcome::Contract(contract) => Err(RetryInput::Contract(contract)),
132        }
133    }
134
135    pub(crate) async fn get_default_topic_route_info_once(
136        &self,
137        deadline: RequestDeadline,
138    ) -> Result<Option<TopicRouteData>, RetryInput> {
139        self.get_topic_route_info_once(TopicValidator::AUTO_CREATE_TOPIC_KEY_TOPIC, deadline, false)
140            .await
141    }
142
143    pub fn get_name_server_address_list(&self) -> Vec<CheetahString> {
144        self.remoting_client.get_name_server_address_list()
145    }
146}
rocketmq_client_rust/implementation/mq_client_api_impl/route.rs

rocketmq_client_rust/implementation/mq_client_api_impl/
route.rs
