rocketmq_client_rust/
lib.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
15#![allow(
16    dead_code,
17    reason = "legacy private Client items remain under the non-growth lint debt registry"
18)]
19#![recursion_limit = "256"]
20
21extern crate core;
22
23// Define macros at crate root so they're available throughout the crate
24/// Create a client error with optional response code
25#[macro_export]
26macro_rules! mq_client_err {
27    // Handle errors with a custom ResponseCode and formatted string
28    ($response_code:expr, $fmt:expr, $($arg:expr),*) => {{
29        let formatted_msg = format!($fmt, $($arg),*);
30        $crate::ClientError::from($crate::MQClientException::new_with_code(
31            $response_code as i32,
32            formatted_msg,
33        ))
34    }};
35
36    ($response_code:expr, $error_message:expr) => {{
37        $crate::ClientError::from($crate::MQClientException::new_with_code(
38            $response_code as i32,
39            $error_message,
40        ))
41    }};
42
43    // Handle errors without a ResponseCode, using only the error message (accepts both &str and String)
44    ($error_message:expr) => {{
45        $crate::ClientError::from($crate::MQClientException::new($error_message))
46    }};
47}
48
49/// Create a broker operation error
50#[macro_export]
51macro_rules! client_broker_err {
52    // Handle errors with a custom ResponseCode and formatted string
53    ($response_code:expr, $error_message:expr, $broker_addr:expr) => {{
54        $crate::ClientError::broker_operation_failed("BROKER_OPERATION", $response_code as i32, $error_message)
55            .with_broker_addr($broker_addr)
56    }};
57    // Handle errors without a ResponseCode, using only the error message
58    ($response_code:expr, $error_message:expr) => {{
59        $crate::ClientError::broker_operation_failed("BROKER_OPERATION", $response_code as i32, $error_message)
60    }};
61}
62
63mod admin;
64mod base;
65mod client_error;
66mod common;
67mod config_support;
68mod consumer;
69mod exception;
70mod factory;
71mod hook;
72mod implementation;
73mod latency;
74mod legacy;
75mod lock;
76mod nameserver_discovery;
77pub mod prelude;
78mod producer;
79mod public_api;
80mod runtime;
81mod session;
82mod stat;
83#[cfg(any(test, feature = "test-support"))]
84pub mod test_support;
85mod trace;
86mod types;
87mod utils;
88
89pub use crate::base::query_result::QueryResult;
90pub use crate::base::MQAdmin;
91pub use crate::base::MqClientAdmin;
92pub use crate::base::MqClientAdminInner;
93pub use crate::client_error::ClientError;
94pub use crate::client_error::ClientResult;
95pub use crate::common::acl::AclConstants;
96pub use crate::common::acl::AclException;
97pub use crate::common::acl::AclSigner;
98pub use crate::common::acl::AclUtils;
99pub use crate::common::acl::Permission;
100pub use crate::common::acl::SigningAlgorithm;
101pub use crate::common::acl_client_rpc_hook::AclClientRPCHook;
102pub use crate::common::admin_tool_result::AdminToolResult;
103pub use crate::common::admin_tools_result_code_enum::AdminToolsResultCodeEnum;
104pub use crate::common::nameserver_access_config::NameserverAccessConfig;
105pub use crate::common::session_credentials::SessionCredentials;
106pub use crate::common::thread_local_index::ThreadLocalIndex;
107pub use crate::consumer::pop_retry_policy::pop_retry_subscription_topics;
108pub use crate::consumer::pop_retry_policy::resolve_pop_retry_topic;
109
110mod cluster_session {
111    use std::collections::HashMap;
112    use std::net::SocketAddr;
113    use std::sync::Arc;
114    use std::sync::Mutex;
115
116    use cheetah_string::CheetahString;
117    use rocketmq_protocol::protocol::remoting_command::RemotingCommand;
118    use rocketmq_security_api::OutboundSigner;
119    use rocketmq_security_api::SecurityRequestView;
120    use rocketmq_transport::api::RPCHook;
121
122    pub use rocketmq_model::common::boundary_type::BoundaryType;
123    pub use rocketmq_model::common::message::message_ext::MessageExt;
124    pub use rocketmq_model::common::message::message_queue::MessageQueue;
125    pub use rocketmq_model::common::message::message_queue_assignment::MessageQueueAssignment;
126
127    pub type ClientRpcHook = dyn RPCHook;
128
129    /// Opaque compatibility handle for the Client instance owned by the Proxy
130    /// Cluster worker.
131    ///
132    /// The standard shared owner remains private to Client; the Cluster adapter
133    /// can only use the instance API and keeps cloned handles on its managed worker.
134    pub struct ClientInstanceHandle {
135        inner: Arc<crate::factory::mq_client_instance::MQClientInstance>,
136        pool: crate::implementation::mq_client_manager::ClientPool,
137        token: Mutex<Option<crate::implementation::mq_client_manager::ClientPoolToken>>,
138        operation_gate: Arc<tokio::sync::Mutex<()>>,
139    }
140
141    /// Isolates a managed Client owner in its injected [`crate::ClientPool`] while
142    /// preserving the caller-provided instance name as a readable prefix.
143    pub fn client_config_for_managed_domain(
144        domain_id: u64,
145        mut client_config: crate::base::client_config::ClientConfig,
146    ) -> crate::base::client_config::ClientConfig {
147        client_config.set_instance_name(CheetahString::from_string(format!(
148            "{}#managed-domain-{domain_id}",
149            client_config.instance_name
150        )));
151        client_config
152    }
153
154    impl ClientInstanceHandle {
155        pub fn get_or_create(
156            client_runtime: &crate::runtime::ClientRuntime,
157            domain_id: u64,
158            client_config: crate::base::client_config::ClientConfig,
159            rpc_hook: Option<Arc<ClientRpcHook>>,
160        ) -> crate::ClientResult<Self> {
161            let client_config = client_config_for_managed_domain(domain_id, client_config);
162            let pool = client_runtime.pool().clone();
163            let (inner, token) = pool.get_or_create(client_config, rpc_hook)?.into_parts();
164            Ok(Self {
165                inner,
166                pool,
167                token: Mutex::new(Some(token)),
168                operation_gate: Arc::new(tokio::sync::Mutex::new(())),
169            })
170        }
171
172        pub async fn start(&self) -> crate::ClientResult<()> {
173            let _operation = self.operation().await;
174            self.inner.start().await
175        }
176
177        /// Invokes one canonical Remoting command on a Broker selected by a managed caller.
178        pub async fn invoke_remoting(
179            &self,
180            broker_addr: &CheetahString,
181            request: RemotingCommand,
182            timeout_millis: u64,
183        ) -> crate::ClientResult<RemotingCommand> {
184            let _operation = self.operation().await;
185            self.inner
186                .get_mq_client_api_impl()?
187                .invoke(broker_addr, request, timeout_millis)
188                .await
189        }
190
191        /// Returns low-cardinality NameServer discovery state without endpoint identities.
192        pub async fn nameserver_discovery_status(&self) -> Option<crate::NameServerDiscoveryStatus> {
193            let _operation = self.operation().await;
194            self.inner.nameserver_discovery_status().await
195        }
196
197        pub async fn topic_route(
198            &self,
199            topic: &str,
200            timeout_millis: u64,
201        ) -> crate::ClientResult<Option<rocketmq_protocol::protocol::route::topic_route_data::TopicRouteData>> {
202            let _operation = self.operation().await;
203            self.inner
204                .get_mq_client_api_impl()?
205                .get_topic_route_info_from_name_server(topic, timeout_millis)
206                .await
207        }
208
209        pub async fn lock_batch_mq(
210            &self,
211            broker_addr: &str,
212            request: rocketmq_protocol::protocol::body::request::lock_batch_request_body::LockBatchRequestBody,
213            timeout_millis: u64,
214        ) -> crate::ClientResult<std::collections::HashSet<MessageQueue>> {
215            let _operation = self.operation().await;
216            self.inner
217                .get_mq_client_api_impl()?
218                .lock_batch_mq(broker_addr, request, timeout_millis)
219                .await
220        }
221
222        pub async fn unlock_batch_mq(
223            &self,
224            broker_addr: &CheetahString,
225            request: rocketmq_protocol::protocol::body::unlock_batch_request_body::UnlockBatchRequestBody,
226            timeout_millis: u64,
227        ) -> crate::ClientResult<()> {
228            let _operation = self.operation().await;
229            self.inner
230                .get_mq_client_api_impl()?
231                .unlock_batch_mq(broker_addr, request, timeout_millis, false)
232                .await
233        }
234
235        #[allow(
236            clippy::too_many_arguments,
237            reason = "mirrors the RocketMQ query-assignment wire contract"
238        )]
239        pub async fn query_assignment(
240            &self,
241            broker_addr: &CheetahString,
242            topic: CheetahString,
243            consumer_group: CheetahString,
244            client_id: CheetahString,
245            strategy_name: CheetahString,
246            message_model: rocketmq_protocol::protocol::heartbeat::message_model::MessageModel,
247            timeout_millis: u64,
248        ) -> crate::ClientResult<Option<Vec<MessageQueueAssignment>>> {
249            let _operation = self.operation().await;
250            self.inner
251                .get_mq_client_api_impl()?
252                .query_assignment(
253                    broker_addr,
254                    topic,
255                    consumer_group,
256                    client_id,
257                    strategy_name,
258                    message_model,
259                    rocketmq_transport::api::RequestDeadline::from_timeout_millis(timeout_millis),
260                )
261                .await
262                .map_err(assignment_query_error)
263                .map(|assignments| assignments.map(|items| items.into_iter().collect()))
264        }
265
266        pub async fn pop_message(
267            &self,
268            broker_name: &CheetahString,
269            broker_addr: &CheetahString,
270            request: rocketmq_protocol::protocol::header::pop_message_request_header::PopMessageRequestHeader,
271            timeout_millis: u64,
272        ) -> crate::ClientResult<crate::consumer::pop_result::PopResult> {
273            let _operation = self.operation().await;
274            let (sender, receiver) = tokio::sync::oneshot::channel();
275            self.inner
276                .get_mq_client_api_impl()?
277                .pop_message_async(
278                    broker_name,
279                    broker_addr,
280                    request,
281                    timeout_millis,
282                    OwnedPopCallback { sender: Some(sender) },
283                )
284                .await?;
285            receiver.await.unwrap_or_else(|source| {
286                Err(crate::ClientError::internal(
287                    "receive client pop callback result",
288                    source,
289                ))
290            })
291        }
292
293        pub async fn ack_message(
294            &self,
295            broker_addr: &CheetahString,
296            request: rocketmq_protocol::protocol::header::ack_message_request_header::AckMessageRequestHeader,
297            timeout_millis: u64,
298        ) -> crate::ClientResult<crate::consumer::ack_result::AckResult> {
299            let _operation = self.operation().await;
300            self.inner
301                .get_mq_client_api_impl()?
302                .ack_message(broker_addr, request, timeout_millis)
303                .await
304        }
305
306        pub async fn batch_ack_message(
307            &self,
308            broker_addr: &CheetahString,
309            request: rocketmq_protocol::protocol::body::batch_ack_message_request_body::BatchAckMessageRequestBody,
310            timeout_millis: u64,
311        ) -> crate::ClientResult<crate::consumer::ack_result::AckResult> {
312            let _operation = self.operation().await;
313            self.inner
314                .get_mq_client_api_impl()?
315                .batch_ack_message(broker_addr, request, timeout_millis)
316                .await
317        }
318
319        #[cfg(feature = "admin-mutation")]
320        pub async fn sync_lite_subscription(
321            &self,
322            broker_addr: &CheetahString,
323            request: rocketmq_model::common::lite::LiteSubscriptionDTO,
324            timeout_millis: u64,
325        ) -> crate::ClientResult<()> {
326            let _operation = self.operation().await;
327            self.inner
328                .get_mq_client_api_impl()?
329                .sync_lite_subscription_async(broker_addr, request, timeout_millis)
330                .await
331        }
332
333        pub async fn change_invisible_time(
334            &self,
335            broker_name: &CheetahString,
336            broker_addr: &CheetahString,
337            request: rocketmq_protocol::protocol::header::change_invisible_time_request_header::ChangeInvisibleTimeRequestHeader,
338            timeout_millis: u64,
339        ) -> crate::ClientResult<crate::consumer::ack_result::AckResult> {
340            let _operation = self.operation().await;
341            let (sender, receiver) = tokio::sync::oneshot::channel();
342            self.inner
343                .get_mq_client_api_impl()?
344                .change_invisible_time_async(
345                    broker_name,
346                    broker_addr,
347                    request,
348                    timeout_millis,
349                    OwnedAckCallback::new(sender),
350                )
351                .await?;
352            receiver.await.unwrap_or_else(|source| {
353                Err(crate::ClientError::internal(
354                    "receive client change-invisible callback result",
355                    source,
356                ))
357            })
358        }
359
360        pub async fn end_transaction(
361            &self,
362            broker_addr: &CheetahString,
363            request: rocketmq_protocol::protocol::header::end_transaction_request_header::EndTransactionRequestHeader,
364            remark: CheetahString,
365            timeout_millis: u64,
366        ) -> crate::ClientResult<()> {
367            let _operation = self.operation().await;
368            self.inner
369                .get_mq_client_api_impl()?
370                .end_transaction_oneway(broker_addr, request, remark, timeout_millis)
371                .await
372        }
373
374        pub async fn find_subscribe_broker_addr(
375            &self,
376            broker_name: &CheetahString,
377            broker_id: u64,
378            only_this_broker: bool,
379        ) -> Option<CheetahString> {
380            let _operation = self.operation().await;
381            let instance = self.inner.clone();
382            instance
383                .find_broker_address_in_subscribe(broker_name, broker_id, only_this_broker)
384                .await
385                .map(|found| found.broker_addr)
386        }
387
388        pub async fn refresh_topic_route(&self, topic: &CheetahString) -> bool {
389            let _operation = self.operation().await;
390            let instance = self.inner.clone();
391            instance.update_topic_route_info_from_name_server_topic(topic).await
392        }
393
394        pub async fn broker_name_for_queue(&self, queue: &MessageQueue) -> CheetahString {
395            let _operation = self.operation().await;
396            self.inner.get_broker_name_from_message_queue(queue).await
397        }
398
399        pub async fn pull_outcome_from_broker(
400            &self,
401            broker_addr: &str,
402            request: rocketmq_protocol::protocol::header::pull_message_request_header::PullMessageRequestHeader,
403            timeout_millis: u64,
404        ) -> crate::ClientResult<rocketmq_model::result::PullOutcome<MessageExt>> {
405            let _operation = self.operation().await;
406            let result = self
407                .inner
408                .pull_message_from_broker(broker_addr, request, timeout_millis)
409                .await?;
410            Ok((&result).into())
411        }
412
413        #[allow(clippy::too_many_arguments, reason = "mirrors the RocketMQ send-back wire contract")]
414        pub async fn consumer_send_message_back(
415            &self,
416            broker_addr: &str,
417            broker_name: Option<&str>,
418            message: &MessageExt,
419            consumer_group: &str,
420            delay_level: i32,
421            timeout_millis: u64,
422            max_consume_retry_times: i32,
423        ) -> crate::ClientResult<()> {
424            let _operation = self.operation().await;
425            self.inner
426                .consumer_send_message_back(
427                    broker_addr,
428                    broker_name,
429                    message,
430                    consumer_group,
431                    delay_level,
432                    timeout_millis,
433                    max_consume_retry_times,
434                )
435                .await
436        }
437
438        pub async fn update_consumer_offset(
439            &self,
440            broker_addr: &CheetahString,
441            request: rocketmq_protocol::protocol::header::update_consumer_offset_header::UpdateConsumerOffsetRequestHeader,
442            timeout_millis: u64,
443        ) -> crate::ClientResult<()> {
444            let _operation = self.operation().await;
445            self.inner
446                .update_consumer_offset(broker_addr, request, timeout_millis)
447                .await
448        }
449
450        pub async fn query_consumer_offset(
451            &self,
452            broker_addr: &str,
453            request: rocketmq_protocol::protocol::header::query_consumer_offset_request_header::QueryConsumerOffsetRequestHeader,
454            timeout_millis: u64,
455        ) -> crate::ClientResult<i64> {
456            let _operation = self.operation().await;
457            self.inner
458                .query_consumer_offset(broker_addr, request, timeout_millis)
459                .await
460        }
461
462        pub async fn min_offset(
463            &self,
464            broker_addr: &str,
465            queue: &MessageQueue,
466            timeout_millis: u64,
467        ) -> crate::ClientResult<i64> {
468            let _operation = self.operation().await;
469            self.inner.get_min_offset(broker_addr, queue, timeout_millis).await
470        }
471
472        pub async fn max_offset(
473            &self,
474            broker_addr: &str,
475            queue: &MessageQueue,
476            timeout_millis: u64,
477        ) -> crate::ClientResult<i64> {
478            let _operation = self.operation().await;
479            self.inner.get_max_offset(broker_addr, queue, timeout_millis).await
480        }
481
482        pub async fn search_offset(
483            &self,
484            broker_addr: &str,
485            queue: &MessageQueue,
486            timestamp: i64,
487            boundary_type: BoundaryType,
488            timeout_millis: u64,
489        ) -> crate::ClientResult<i64> {
490            let _operation = self.operation().await;
491            self.inner
492                .search_offset_by_timestamp(broker_addr, queue, timestamp, boundary_type, timeout_millis)
493                .await
494        }
495
496        pub async fn topic_config(
497            &self,
498            broker_addr: &CheetahString,
499            topic: CheetahString,
500            timeout_millis: u64,
501        ) -> crate::ClientResult<rocketmq_model::common::config::TopicConfig> {
502            let _operation = self.operation().await;
503            self.inner.get_topic_config(broker_addr, topic, timeout_millis).await
504        }
505
506        pub async fn subscription_group_config(
507            &self,
508            broker_addr: &CheetahString,
509            group: CheetahString,
510            timeout_millis: u64,
511        ) -> crate::ClientResult<
512            rocketmq_protocol::protocol::subscription::subscription_group_config::SubscriptionGroupConfig,
513        > {
514            let _operation = self.operation().await;
515            self.inner
516                .get_subscription_group_config(broker_addr, group, timeout_millis)
517                .await
518        }
519
520        pub async fn broker_cluster_info(
521            &self,
522            timeout_millis: u64,
523        ) -> crate::ClientResult<rocketmq_protocol::protocol::body::broker_body::cluster_info::ClusterInfo> {
524            let _operation = self.operation().await;
525            self.inner.get_broker_cluster_info(timeout_millis).await
526        }
527
528        pub async fn user(
529            &self,
530            broker_addr: CheetahString,
531            username: CheetahString,
532            timeout_millis: u64,
533        ) -> crate::ClientResult<Option<rocketmq_protocol::protocol::body::user_info::UserInfo>> {
534            let _operation = self.operation().await;
535            self.inner.get_user(broker_addr, username, timeout_millis).await
536        }
537
538        pub async fn acl(
539            &self,
540            broker_addr: CheetahString,
541            subject: CheetahString,
542            timeout_millis: u64,
543        ) -> crate::ClientResult<Option<rocketmq_protocol::protocol::body::acl_info::AclInfo>> {
544            let _operation = self.operation().await;
545            self.inner.get_acl(broker_addr, subject, timeout_millis).await
546        }
547
548        pub async fn shutdown_owned(&self) {
549            let token = self
550                .token
551                .lock()
552                .unwrap_or_else(|poisoned| poisoned.into_inner())
553                .take();
554            let Some(token) = token else {
555                return;
556            };
557            let _operation = self.operation().await;
558            self.pool.release(token).await;
559        }
560
561        async fn operation(&self) -> tokio::sync::OwnedMutexGuard<()> {
562            self.operation_gate.clone().lock_owned().await
563        }
564    }
565
566    fn assignment_query_error(input: crate::common::retry_policy::RetryInput) -> crate::ClientError {
567        use crate::common::retry_policy::RetryInput;
568        use crate::ClientError;
569        use rocketmq_error::fields;
570        use rocketmq_error::Error;
571        use rocketmq_error::ErrorContext;
572        use rocketmq_transport::api::OutboundRequestContractReason;
573        use rocketmq_transport::api::OutboundRequestRejectionReason;
574
575        match input {
576            RetryInput::Transport(error) => ClientError::from_shared(error.into_shared_error()),
577            RetryInput::Rejected(rejection) => match rejection.reason() {
578                OutboundRequestRejectionReason::DeadlineExpired => {
579                    let mut context = ErrorContext::new().with_text(fields::OPERATION_DIAGNOSTIC, "query_assignment");
580                    if let Some(timeout_millis) = rejection.timeout_millis() {
581                        context = context.with_u64(fields::TIMEOUT_MS, timeout_millis);
582                    }
583                    ClientError::from_shared(Arc::new(
584                        Error::new(&rocketmq_error::CORE_OPERATION_TIMED_OUT).with_context(context),
585                    ))
586                }
587                OutboundRequestRejectionReason::ClientStopping => ClientError::not_started(),
588                OutboundRequestRejectionReason::QueueSaturated => ClientError::from_shared(Arc::new(Error::new(
589                    &rocketmq_error::TRANSPORT_ADMISSION_QUEUE_SATURATED,
590                ))),
591                OutboundRequestRejectionReason::Cancelled
592                | OutboundRequestRejectionReason::SessionClosed
593                | OutboundRequestRejectionReason::EndpointUnavailable => {
594                    ClientError::from_shared(Arc::new(Error::new(&rocketmq_error::TRANSPORT_CONNECTION_FAILED)))
595                }
596            },
597            RetryInput::Contract(contract) => match contract.reason() {
598                OutboundRequestContractReason::NameServerEndpointMissing => {
599                    ClientError::from_shared(Arc::new(Error::new(&rocketmq_error::TRANSPORT_CONNECTION_FAILED)))
600                }
601            },
602            RetryInput::Response { terminal_error, .. } | RetryInput::BusinessError(terminal_error) => terminal_error,
603            // The query-assignment wire operation cannot construct producer send status or
604            // route-selection inputs; MQClientInstance owns route selection before this facade.
605            RetryInput::SendStatus(_) | RetryInput::RouteUnavailable => {
606                unreachable!("query-assignment API returned an unrelated retry input")
607            }
608        }
609    }
610
611    impl Drop for ClientInstanceHandle {
612        fn drop(&mut self) {
613            if self
614                .token
615                .get_mut()
616                .unwrap_or_else(|poisoned| poisoned.into_inner())
617                .is_some()
618            {
619                tracing::warn!(
620                    client_id = %self.inner.client_id,
621                    "managed Client handle dropped before asynchronous shutdown completed"
622                );
623            }
624        }
625    }
626
627    struct OwnedPopCallback {
628        sender: Option<tokio::sync::oneshot::Sender<crate::ClientResult<crate::consumer::pop_result::PopResult>>>,
629    }
630
631    impl crate::consumer::pop_callback::PopCallback for OwnedPopCallback {
632        async fn on_success(&mut self, pop_result: crate::consumer::pop_result::PopResult) {
633            if let Some(sender) = self.sender.take() {
634                let _ = sender.send(Ok(pop_result));
635            }
636        }
637
638        fn on_error(&mut self, error: crate::ClientError) {
639            if let Some(sender) = self.sender.take() {
640                let _ = sender.send(Err(error));
641            }
642        }
643    }
644
645    struct OwnedAckCallback {
646        sender:
647            Mutex<Option<tokio::sync::oneshot::Sender<crate::ClientResult<crate::consumer::ack_result::AckResult>>>>,
648    }
649
650    impl OwnedAckCallback {
651        fn new(
652            sender: tokio::sync::oneshot::Sender<crate::ClientResult<crate::consumer::ack_result::AckResult>>,
653        ) -> Self {
654            Self {
655                sender: Mutex::new(Some(sender)),
656            }
657        }
658
659        fn send(&self, result: crate::ClientResult<crate::consumer::ack_result::AckResult>) {
660            let mut sender = match self.sender.lock() {
661                Ok(sender) => sender,
662                Err(poisoned) => poisoned.into_inner(),
663            };
664            if let Some(sender) = sender.take() {
665                let _ = sender.send(result);
666            }
667        }
668    }
669
670    impl crate::consumer::ack_callback::AckCallback for OwnedAckCallback {
671        fn on_success(&self, ack_result: crate::consumer::ack_result::AckResult) {
672            self.send(Ok(ack_result));
673        }
674
675        fn on_exception(&self, error: crate::ClientError) {
676            self.send(Err(error));
677        }
678    }
679
680    pub fn rpc_hook_from_outbound_signer(signer: Arc<dyn OutboundSigner>) -> Arc<ClientRpcHook> {
681        Arc::new(OutboundSignerRpcHook { signer })
682    }
683
684    struct OutboundSignerRpcHook {
685        signer: Arc<dyn OutboundSigner>,
686    }
687
688    impl RPCHook for OutboundSignerRpcHook {
689        fn do_before_request(
690            &self,
691            _remote_addr: SocketAddr,
692            request: &mut RemotingCommand,
693        ) -> Result<(), rocketmq_error::SharedError> {
694            // RPCHook callers normally serialize the custom header first, but
695            // keep the adapter correct when invoked directly or by another
696            // Client transport path.
697            request.make_custom_header_to_net();
698            let empty_fields = HashMap::new();
699            let fields = request.ext_fields().unwrap_or(&empty_fields);
700            let signature = self
701                .signer
702                .sign(SecurityRequestView::new(
703                    request.code(),
704                    request.version(),
705                    fields,
706                    request.body().map(AsRef::as_ref),
707                    None,
708                ))
709                .map_err(|source| source.into_shared_error())?;
710
711            request.ensure_ext_fields_initialized();
712            for (key, value) in signature.fields() {
713                request.add_ext_field(key.clone(), value.expose_secret().clone());
714            }
715            Ok(())
716        }
717
718        fn do_after_response(
719            &self,
720            _remote_addr: SocketAddr,
721            _request: &RemotingCommand,
722            _response: &mut RemotingCommand,
723        ) -> Result<(), rocketmq_error::SharedError> {
724            Ok(())
725        }
726    }
727
728    #[cfg(test)]
729    mod tests {
730        use std::sync::Arc;
731
732        use super::rpc_hook_from_outbound_signer;
733        use super::ClientInstanceHandle;
734        use super::OutboundSigner;
735        use super::RemotingCommand;
736        use super::SecurityRequestView;
737        use rocketmq_protocol::code::request_code::RequestCode;
738        use rocketmq_protocol::protocol::header::client_request_header::GetRouteInfoRequestHeader;
739        use rocketmq_security_api::Secret;
740        use rocketmq_security_api::SecurityOperation;
741        use rocketmq_security_api::SecurityProviderError;
742        use rocketmq_security_api::SecurityProviderFailure;
743        use rocketmq_security_api::Signature;
744
745        struct FixedSigner;
746
747        impl OutboundSigner for FixedSigner {
748            fn sign(&self, request: SecurityRequestView<'_>) -> Result<Signature, SecurityProviderError> {
749                assert_eq!(
750                    request.fields().get("topic").map(|value| value.as_str()),
751                    Some("TopicA")
752                );
753                Ok(Signature::new(vec![
754                    ("AccessKey".into(), Secret::new("cluster-client".into())),
755                    ("Signature".into(), Secret::new("redacted-signature".into())),
756                ]))
757            }
758        }
759
760        struct FailingSigner;
761
762        impl OutboundSigner for FailingSigner {
763            fn sign(&self, _request: SecurityRequestView<'_>) -> Result<Signature, SecurityProviderError> {
764                Err(SecurityProviderError::caused_by(
765                    SecurityProviderFailure::OperationFailed,
766                    SecurityOperation::SignRequest,
767                    std::io::Error::other("secret signing diagnostic"),
768                ))
769            }
770        }
771
772        #[test]
773        fn outbound_signer_adapter_applies_fields_to_client_rpc_request() {
774            let hook = rpc_hook_from_outbound_signer(Arc::new(FixedSigner));
775            let mut request = RemotingCommand::create_request_command(
776                RequestCode::GetRouteinfoByTopic,
777                GetRouteInfoRequestHeader::new("TopicA", None),
778            );
779
780            hook.do_before_request("127.0.0.1:9876".parse().unwrap(), &mut request)
781                .unwrap();
782
783            let fields = request.ext_fields().unwrap();
784            assert_eq!(
785                fields.get("AccessKey").map(|value| value.as_str()),
786                Some("cluster-client")
787            );
788            assert_eq!(
789                fields.get("Signature").map(|value| value.as_str()),
790                Some("redacted-signature")
791            );
792        }
793
794        #[test]
795        fn outbound_signer_adapter_redacts_signer_failures() {
796            let hook = rpc_hook_from_outbound_signer(Arc::new(FailingSigner));
797            let mut request = RemotingCommand::create_request_command(
798                RequestCode::GetRouteinfoByTopic,
799                GetRouteInfoRequestHeader::new("TopicA", None),
800            );
801
802            let error = hook
803                .do_before_request("127.0.0.1:9876".parse().unwrap(), &mut request)
804                .expect_err("signing failure must fail closed");
805            let message = error.to_string();
806            assert!(message.contains("auth.security_provider.operation_failed"));
807            assert!(!message.contains("secret signing diagnostic"));
808            assert_eq!(error.descriptor(), &rocketmq_error::SECURITY_PROVIDER_OPERATION_FAILED);
809            let io = std::error::Error::source(error.as_ref()).expect("I/O cause must remain available");
810            assert!(io.downcast_ref::<std::io::Error>().is_some());
811        }
812
813        #[tokio::test]
814        async fn managed_client_is_shared_until_the_last_owner_releases_it() {
815            let runtime = crate::runtime::test_client_runtime("managed-client-shared-test");
816            let domain_id = 71_001;
817            let mut config = crate::base::client_config::ClientConfig::default();
818            config.set_instance_name("managed-client-shared".into());
819
820            let first = ClientInstanceHandle::get_or_create(&runtime, domain_id, config.clone(), None).unwrap();
821            let second = ClientInstanceHandle::get_or_create(&runtime, domain_id, config, None).unwrap();
822
823            assert!(Arc::ptr_eq(&first.inner, &second.inner));
824            assert_eq!(runtime.pool().instance_count(), 1);
825
826            first.shutdown_owned().await;
827            assert_eq!(runtime.pool().instance_count(), 1);
828
829            second.shutdown_owned().await;
830            assert_eq!(runtime.pool().instance_count(), 0);
831        }
832
833        #[tokio::test]
834        async fn managed_client_rejects_conflicting_configuration_for_the_same_identity() {
835            let runtime = crate::runtime::test_client_runtime("managed-client-conflict-test");
836            let domain_id = 71_002;
837            let mut first_config = crate::base::client_config::ClientConfig::default();
838            first_config.set_instance_name("managed-client-conflict".into());
839            first_config.set_namesrv_addr("127.0.0.1:9876".into());
840            let mut second_config = first_config.clone();
841            second_config.set_namesrv_addr("127.0.0.2:9876".into());
842
843            let first = ClientInstanceHandle::get_or_create(&runtime, domain_id, first_config, None).unwrap();
844            let error = ClientInstanceHandle::get_or_create(&runtime, domain_id, second_config, None)
845                .err()
846                .expect("conflicting managed configuration must fail closed");
847
848            assert!(error.is(&rocketmq_error::CORE_ARGUMENT_INVALID));
849            first.shutdown_owned().await;
850        }
851
852        #[tokio::test]
853        async fn managed_client_isolated_across_runtime_domains() {
854            let mut config = crate::base::client_config::ClientConfig::default();
855            config.set_instance_name("managed-client-cross-domain".into());
856            let first_runtime = crate::runtime::test_client_runtime("managed-client-domain-a");
857            let second_runtime = crate::runtime::test_client_runtime("managed-client-domain-b");
858            let first = ClientInstanceHandle::get_or_create(&first_runtime, 71_004, config.clone(), None).unwrap();
859            let second = ClientInstanceHandle::get_or_create(&second_runtime, 71_004, config, None).unwrap();
860
861            assert!(!Arc::ptr_eq(&first.inner, &second.inner));
862            assert_eq!(first_runtime.pool().instance_count(), 1);
863            assert_eq!(second_runtime.pool().instance_count(), 1);
864
865            first.shutdown_owned().await;
866            assert_eq!(first_runtime.pool().instance_count(), 0);
867            assert_eq!(second_runtime.pool().instance_count(), 1);
868            second.shutdown_owned().await;
869            assert_eq!(second_runtime.pool().instance_count(), 0);
870        }
871
872        #[tokio::test]
873        async fn releasing_the_last_lease_prevents_manager_aba_reuse() {
874            let runtime = crate::runtime::test_client_runtime("managed-client-aba-test");
875            let mut config = crate::base::client_config::ClientConfig::default();
876            config.set_instance_name("managed-client-aba".into());
877            let first = ClientInstanceHandle::get_or_create(&runtime, 71_008, config.clone(), None).unwrap();
878            first.shutdown_owned().await;
879
880            let second = ClientInstanceHandle::get_or_create(&runtime, 71_008, config, None).unwrap();
881            assert!(!Arc::ptr_eq(&first.inner, &second.inner));
882            second.shutdown_owned().await;
883        }
884    }
885}
886pub use crate::consumer::consumer_impl::pull_request_ext::PullResultExt;
887pub use crate::consumer::listener::message_listener_concurrently::ArcMessageListenerConcurrently;
888pub use crate::consumer::listener::message_listener_orderly::ArcMessageListenerOrderly;
889pub use crate::consumer::notify_result::NotifyResult;
890pub use crate::consumer::pull_result::PullOutcome;
891pub use crate::consumer::AbstractAllocateMessageQueueStrategy;
892pub use crate::consumer::AckCallback;
893pub use crate::consumer::AckCallbackFn;
894pub use crate::consumer::AckResult;
895pub use crate::consumer::AckStatus;
896pub use crate::consumer::AllocateMachineRoomNearby;
897pub use crate::consumer::AllocateMessageQueueAveragely;
898pub use crate::consumer::AllocateMessageQueueAveragelyByCircle;
899pub use crate::consumer::AllocateMessageQueueByConfig;
900pub use crate::consumer::AllocateMessageQueueByMachineRoom;
901pub use crate::consumer::AllocateMessageQueueByMachineRoomNearby;
902pub use crate::consumer::AllocateMessageQueueConsistentHash;
903pub use crate::consumer::AllocateMessageQueueStrategy;
904pub use crate::consumer::ArcMessageQueueListener;
905pub use crate::consumer::ClassicPullCallback;
906pub use crate::consumer::ConsumeConcurrentlyContext;
907pub use crate::consumer::ConsumeConcurrentlyStatus;
908pub use crate::consumer::ConsumeOrderlyContext;
909pub use crate::consumer::ConsumeOrderlyStatus;
910pub use crate::consumer::ConsumerTuningProfile;
911pub use crate::consumer::ControllableOffset;
912pub use crate::consumer::DefaultLitePullConsumerBuilder;
913#[allow(deprecated)]
914pub use crate::consumer::DefaultMQPullConsumer;
915pub use crate::consumer::DefaultMQPullConsumerBuilder;
916pub use crate::consumer::DefaultMQPushConsumerBuilder;
917pub use crate::consumer::HashFunction;
918pub use crate::consumer::LocalFileOffsetStore;
919pub use crate::consumer::MQConsumer;
920pub use crate::consumer::MQConsumerInner;
921#[allow(deprecated)]
922pub use crate::consumer::MQPullConsumer;
923pub use crate::consumer::MQPushConsumer;
924pub use crate::consumer::MachineRoomResolver;
925pub use crate::consumer::MessageListener;
926pub use crate::consumer::MessageListenerConcurrently;
927pub use crate::consumer::MessageListenerOrderly;
928pub use crate::consumer::MessageQueueListener;
929pub use crate::consumer::MessageSelector;
930pub use crate::consumer::OffsetSerialize;
931pub use crate::consumer::OffsetSerializeWrapper;
932pub use crate::consumer::OffsetStore;
933pub use crate::consumer::PopCallback;
934pub use crate::consumer::PopCallbackFn;
935pub use crate::consumer::PopResult;
936pub use crate::consumer::PopStatus;
937pub use crate::consumer::PullCallback;
938pub use crate::consumer::PullCallbackFn;
939pub use crate::consumer::PullOptions;
940pub use crate::consumer::PullResult;
941pub use crate::consumer::PullStatus;
942pub use crate::consumer::ReadOffsetType;
943pub use crate::consumer::RemoteBrokerOffsetStore;
944pub use crate::consumer::TopicMessageQueueChangeListener;
945pub use crate::exception::MQBrokerException;
946pub use crate::exception::MQClientException;
947pub use crate::exception::OffsetNotFoundException;
948pub use crate::exception::RequestTimeoutException;
949pub use crate::hook::check_forbidden_context::CheckForbiddenContext;
950pub use crate::hook::check_forbidden_hook::CheckForbiddenHook;
951pub use crate::hook::consume_message_context::ConsumeMessageContext;
952pub use crate::hook::consume_message_hook::ConsumeMessageHook;
953pub use crate::hook::consume_message_hook::ConsumeMessageHookArc;
954pub use crate::hook::namespace_rpc_hook::NamespaceRpcHook;
955pub use crate::implementation::mq_client_api_factory::MQClientAPIFactory;
956pub use crate::implementation::mq_client_api_impl::AdminClient;
957pub use crate::implementation::mq_client_api_impl::ConsumerClient;
958pub use crate::implementation::mq_client_api_impl::ProducerClient;
959pub use crate::implementation::mq_client_api_impl::RouteClient;
960pub use crate::implementation::mq_client_api_impl::TransactionClient;
961pub use crate::implementation::mq_client_manager::ClientPool;
962pub use crate::latency::BrokerFilter;
963pub use crate::latency::MQFaultStrategy;
964pub use crate::latency::Resolver;
965pub use crate::latency::ServiceDetector;
966#[allow(deprecated)]
967pub use crate::legacy::ConsumeMessageOpenTracingHookImpl;
968pub use crate::legacy::ConsumeRequest;
969#[allow(deprecated)]
970pub use crate::legacy::DefaultMQPullConsumerImpl;
971pub use crate::legacy::DoNothingClientRemotingProcessor;
972#[allow(deprecated)]
973pub use crate::legacy::EndTransactionOpenTracingHookImpl;
974#[allow(deprecated)]
975pub use crate::legacy::MQHelper;
976#[allow(deprecated)]
977pub use crate::legacy::MQPullConsumerScheduleService;
978#[allow(deprecated)]
979pub use crate::legacy::PullTaskCallback;
980#[allow(deprecated)]
981pub use crate::legacy::PullTaskContext;
982#[allow(deprecated)]
983pub use crate::legacy::PullTaskImpl;
984pub use crate::legacy::RebalanceImpl;
985#[allow(deprecated)]
986pub use crate::legacy::RebalancePullImpl;
987#[allow(deprecated)]
988pub use crate::legacy::SendMessageOpenTracingHookImpl;
989#[allow(deprecated)]
990pub use crate::legacy::TransactionCheckListener;
991pub use crate::lock::ReadWriteCASLock;
992pub use crate::producer::send_callback::ArcSendCallback;
993pub use crate::producer::transaction_listener::ArcTransactionListener;
994pub use crate::producer::BatchSendCallbackRequest;
995pub use crate::producer::BatchSendRequest;
996pub use crate::producer::JavaHashCode;
997pub use crate::producer::LocalTransactionState;
998pub use crate::producer::MessageQueueSelector;
999pub use crate::producer::MessageQueueSelectorFn;
1000pub use crate::producer::ProducerLifecycle;
1001pub use crate::producer::ProducerQueryRequest;
1002pub use crate::producer::ProducerQueryResponse;
1003pub use crate::producer::RecallRequest;
1004pub use crate::producer::RequestCallback;
1005pub use crate::producer::RequestReplyCallbackRequest;
1006pub use crate::producer::RequestReplyRequest;
1007pub use crate::producer::SelectMessageQueueByHash;
1008pub use crate::producer::SelectMessageQueueByMachineRoom;
1009pub use crate::producer::SelectMessageQueueByRandom;
1010pub use crate::producer::SelectedRequestReplyCallbackRequest;
1011pub use crate::producer::SelectedRequestReplyRequest;
1012pub use crate::producer::SelectedSendCallbackRequest;
1013pub use crate::producer::SelectedSendRequest;
1014pub use crate::producer::SendCallback;
1015pub use crate::producer::SendCallbackRequest;
1016pub use crate::producer::SendDestination;
1017pub use crate::producer::SendMode;
1018pub use crate::producer::SendRequest;
1019pub use crate::producer::SendResult;
1020pub use crate::producer::SendStatus;
1021pub use crate::producer::TopicCreateRequest;
1022pub use crate::producer::TransactionListener;
1023pub use crate::producer::TransactionMQProducer;
1024pub use crate::producer::TransactionMQProducerBuilder;
1025pub use crate::producer::TransactionSendRequest;
1026pub use crate::producer::TransactionSendResult;
1027pub use crate::runtime::ClientMetrics;
1028pub use crate::runtime::ClientRuntime;
1029pub use crate::runtime::ClientRuntimeConfig;
1030pub use crate::runtime::TelemetryHandle;
1031pub use crate::trace::async_trace_dispatcher::AsyncTraceDispatcher;
1032pub use crate::trace::trace_data_encoder::TraceDataEncoder;
1033pub use crate::trace::trace_dispatcher::ArcTraceDispatcher;
1034pub use crate::trace::trace_dispatcher::TraceDispatcher;
1035pub use crate::trace::trace_dispatcher::Type as TraceDispatcherOperation;
1036pub use crate::trace::trace_dispatcher_type::TraceDispatcherType;
1037pub use crate::trace::trace_type::TraceType;
1038pub use crate::utils::message_util::MessageUtil;
1039pub use public_api::*;
1040
1041#[cfg(test)]
1042mod tests {
1043    #[test]
1044    fn mq_client_err_without_response_code_preserves_message() {
1045        let err = crate::mq_client_err!("simple client error");
1046
1047        assert!(err.is(&rocketmq_error::CORE_ARGUMENT_INVALID));
1048        assert!(!err.to_string().contains("simple client error"));
1049    }
1050}
rocketmq_client_rust/lib.rs

rocketmq_client_rust/
lib.rs
