rocketmq_client_rust/base/
client_config.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;
16use std::sync::atomic::AtomicBool;
17use std::sync::atomic::Ordering;
18use std::sync::Arc;
19use std::time::Duration;
20
21use crate::config_support::name_server_address_utils::NameServerAddressUtils;
22use crate::config_support::name_server_target::parse_legacy_namesrv_addr;
23use crate::config_support::name_server_target::NormalizedNameServerTargets;
24use crate::ClientError;
25use crate::ClientResult;
26use cheetah_string::CheetahString;
27use rocketmq_model::common::message::message_queue::MessageQueue;
28use rocketmq_protocol::protocol::namespace_util::NamespaceUtil;
29use rocketmq_protocol::protocol::request_type::RequestType;
30use rocketmq_protocol::protocol::LanguageCode;
31use rocketmq_runtime::common::time_utils::current_nano;
32use rocketmq_transport::api::NetworkUtil;
33use rocketmq_transport::api::SocksProxyConfig;
34use rocketmq_transport::api::TlsConfig;
35
36use crate::base::access_channel::AccessChannel;
37
38#[derive(Clone)]
39pub struct ClientConfig {
40    pub namesrv_addr: Option<CheetahString>,
41    pub client_ip: Option<CheetahString>,
42    pub instance_name: CheetahString,
43    pub client_callback_executor_threads: usize,
44    pub namespace: Option<CheetahString>,
45    pub namespace_initialized: Arc<AtomicBool>,
46    pub namespace_v2: Option<CheetahString>,
47    pub access_channel: AccessChannel,
48    pub poll_name_server_interval: u32,
49    pub heartbeat_broker_interval: u32,
50    pub persist_consumer_offset_interval: u32,
51    pub pull_time_delay_millis_when_exception: u32,
52    pub unit_mode: bool,
53    pub unit_name: Option<CheetahString>,
54    pub decode_read_body: bool,
55    pub decode_decompress_body: bool,
56    pub vip_channel_enabled: bool,
57    pub use_heartbeat_v2: bool,
58    pub enable_concurrent_heartbeat: bool,
59    pub use_tls: bool,
60    pub tls_config: TlsConfig,
61    pub socks_proxy_config: CheetahString,
62    pub mq_client_api_timeout: u64,
63    pub detect_timeout: u32,
64    pub detect_interval: u32,
65    pub language: LanguageCode,
66    pub enable_stream_request_type: bool,
67    pub send_latency_enable: bool,
68    pub start_detector_enable: bool,
69    pub enable_heartbeat_channel_event_listener: bool,
70    pub enable_trace: bool,
71    pub trace_topic: Option<CheetahString>,
72    pub trace_msg_batch_num: usize,
73    pub max_page_size_in_get_metadata: usize,
74    /// Thread pool size for concurrent heartbeat operations.
75    /// Only effective when enable_concurrent_heartbeat is true.
76    /// Default: number of CPU cores (matches Java: Runtime.getRuntime().availableProcessors())
77    // java client has this option, but we keep it for compatibility and future will remove it if it's not needed
78    pub concurrent_heartbeat_thread_pool_size: usize,
79    /// Pull request worker shard count for PullMessageService.
80    /// Default: min(number of CPU cores, 8).
81    pub pull_message_service_shards: usize,
82}
83
84impl Default for ClientConfig {
85    fn default() -> Self {
86        Self::new()
87    }
88}
89
90impl ClientConfig {
91    pub const SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY: &'static str = "com.rocketmq.sendMessageWithVIPChannel";
92    pub const SOCKS_PROXY_CONFIG: &'static str = "com.rocketmq.socks.proxy.config";
93    pub const DECODE_READ_BODY: &'static str = "com.rocketmq.read.body";
94    pub const DECODE_DECOMPRESS_BODY: &'static str = "com.rocketmq.decompress.body";
95    pub const SEND_LATENCY_ENABLE: &'static str = "com.rocketmq.sendLatencyEnable";
96    pub const START_DETECTOR_ENABLE: &'static str = "com.rocketmq.startDetectorEnable";
97    pub const HEART_BEAT_V2: &'static str = "com.rocketmq.heartbeat.v2";
98    pub const ENABLE_CONCURRENT_HEARTBEAT: &'static str = "com.rocketmq.enableConcurrentHeartbeat";
99
100    pub fn new() -> Self {
101        ClientConfig {
102            namesrv_addr: NameServerAddressUtils::get_name_server_addresses().map(|addr| addr.into()),
103            client_ip: NetworkUtil::get_local_address().map(|addr| addr.into()),
104            instance_name: env::var("rocketmq.client.name")
105                .unwrap_or_else(|_| "DEFAULT".to_string())
106                .into(),
107            client_callback_executor_threads: num_cpus::get(),
108            namespace: None,
109            namespace_initialized: Arc::new(AtomicBool::new(false)),
110            namespace_v2: None,
111            access_channel: AccessChannel::Local,
112            poll_name_server_interval: Duration::from_secs(30).as_millis() as u32,
113            heartbeat_broker_interval: Duration::from_secs(30).as_millis() as u32,
114            persist_consumer_offset_interval: Duration::from_secs(5).as_millis() as u32,
115            pull_time_delay_millis_when_exception: 1000,
116            unit_mode: false,
117            unit_name: None,
118            decode_read_body: env::var(Self::DECODE_READ_BODY)
119                .unwrap_or_else(|_| "true".to_string())
120                .parse::<bool>()
121                .unwrap_or(true),
122            decode_decompress_body: env::var(Self::DECODE_DECOMPRESS_BODY)
123                .unwrap_or_else(|_| "true".to_string())
124                .parse::<bool>()
125                .unwrap_or(true),
126            vip_channel_enabled: env::var(Self::SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY)
127                .unwrap_or_else(|_| "false".to_string())
128                .parse::<bool>()
129                .unwrap_or(false),
130            use_heartbeat_v2: env::var(Self::HEART_BEAT_V2)
131                .unwrap_or_else(|_| "false".to_string())
132                .parse::<bool>()
133                .unwrap_or(false),
134            enable_concurrent_heartbeat: env::var(Self::ENABLE_CONCURRENT_HEARTBEAT)
135                .unwrap_or_else(|_| "false".to_string())
136                .parse::<bool>()
137                .unwrap_or(false),
138            use_tls: false,
139            tls_config: TlsConfig::default(),
140            socks_proxy_config: env::var(Self::SOCKS_PROXY_CONFIG)
141                .unwrap_or_else(|_| "{}".to_string())
142                .into(),
143            mq_client_api_timeout: Duration::from_secs(3).as_millis() as u64,
144            detect_timeout: 200,
145            detect_interval: Duration::from_secs(2).as_millis() as u32,
146            language: LanguageCode::RUST,
147            enable_stream_request_type: false,
148            send_latency_enable: env::var(Self::SEND_LATENCY_ENABLE)
149                .unwrap_or_else(|_| "false".to_string())
150                .parse::<bool>()
151                .unwrap_or(false),
152            start_detector_enable: env::var(Self::START_DETECTOR_ENABLE)
153                .unwrap_or_else(|_| "false".to_string())
154                .parse::<bool>()
155                .unwrap_or(false),
156            enable_heartbeat_channel_event_listener: true,
157            enable_trace: false,
158            trace_topic: None,
159            trace_msg_batch_num: 10,
160            max_page_size_in_get_metadata: 2000,
161            concurrent_heartbeat_thread_pool_size: num_cpus::get(),
162            pull_message_service_shards: num_cpus::get().clamp(1, 8),
163        }
164    }
165}
166
167impl ClientConfig {
168    #[inline]
169    pub fn with_namespace(&mut self, resource: impl Into<CheetahString>) -> CheetahString {
170        let resource = resource.into();
171        let namespace = self.get_namespace().unwrap_or_default();
172
173        // Fast path: no namespace needed, return resource directly
174        if namespace.is_empty() {
175            return resource;
176        }
177
178        // Fast path: resource already has namespace, return directly
179        if NamespaceUtil::is_already_with_namespace(resource.as_str(), namespace.as_str()) {
180            return resource;
181        }
182
183        NamespaceUtil::wrap_namespace(namespace, resource)
184    }
185
186    #[inline]
187    pub fn queue_with_namespace(&mut self, mut queue: MessageQueue) -> MessageQueue {
188        if let Some(namespace) = self.get_namespace() {
189            if !namespace.is_empty() {
190                let topic = NamespaceUtil::wrap_namespace(namespace.as_str(), queue.topic_str());
191                queue.set_topic(topic);
192                return queue;
193            }
194        }
195        queue
196    }
197
198    pub(crate) fn queue_with_resolved_namespace(&self, mut queue: MessageQueue) -> MessageQueue {
199        if let Some(namespace) = self.resolved_namespace() {
200            if !namespace.is_empty() {
201                let topic = NamespaceUtil::wrap_namespace(namespace.as_str(), queue.topic_str());
202                queue.set_topic(topic);
203            }
204        }
205        queue
206    }
207
208    #[inline]
209    pub fn queues_with_namespace<I>(&mut self, queues: I) -> Vec<MessageQueue>
210    where
211        I: IntoIterator<Item = MessageQueue>,
212    {
213        queues
214            .into_iter()
215            .map(|queue| self.queue_with_namespace(queue))
216            .collect()
217    }
218
219    #[inline]
220    pub fn get_namespace(&mut self) -> Option<CheetahString> {
221        let namespace_initialized = self.namespace_initialized.load(Ordering::Acquire);
222        if namespace_initialized {
223            return self.namespace.clone();
224        }
225
226        self.namespace = self.resolved_namespace();
227        self.namespace_initialized.store(true, Ordering::Release);
228        self.namespace.clone()
229    }
230
231    pub(crate) fn resolved_namespace(&self) -> Option<CheetahString> {
232        if self.namespace_initialized.load(Ordering::Acquire) {
233            return self.namespace.clone();
234        }
235        if let Some(namespace) = self.namespace.as_ref() {
236            return Some(namespace.clone());
237        }
238        self.namesrv_addr
239            .as_ref()
240            .filter(|address| NameServerAddressUtils::validate_instance_endpoint(address.as_ref()))
241            .and_then(|address| NameServerAddressUtils::parse_instance_id_from_endpoint(address.as_ref()))
242            .map(Into::into)
243    }
244
245    #[inline]
246    pub fn change_instance_name_to_pid(&mut self) {
247        if self.instance_name == "DEFAULT" {
248            self.instance_name = format!("{}#{}", std::process::id(), current_nano()).into();
249        }
250    }
251    #[inline]
252    pub fn set_instance_name(&mut self, instance_name: CheetahString) {
253        self.instance_name = instance_name;
254    }
255    #[inline]
256    pub fn set_namesrv_addr(&mut self, namesrv_addr: CheetahString) {
257        self.namesrv_addr = Some(namesrv_addr);
258        self.namespace_initialized.store(false, Ordering::Release);
259    }
260
261    #[inline]
262    pub fn build_mq_client_id(&self) -> String {
263        // Pre-allocate capacity to avoid reallocations
264        let estimated_capacity = self.client_ip.as_ref().map(|ip| ip.len()).unwrap_or(15)
265            + self.instance_name.len()
266            + self.unit_name.as_ref().map(|un| un.len() + 1).unwrap_or(0)
267            + if self.enable_stream_request_type { 8 } else { 0 }
268            + 3; // For '@' separators
269
270        let mut sb = String::with_capacity(estimated_capacity);
271        if let Some(ref client_ip) = self.client_ip {
272            sb.push_str(client_ip.as_str());
273        }
274
275        sb.push('@');
276        sb.push_str(self.instance_name.as_str());
277        if let Some(ref unit_name) = self.unit_name {
278            if !unit_name.is_empty() {
279                sb.push('@');
280                sb.push_str(unit_name.as_str());
281            }
282        }
283
284        if self.enable_stream_request_type {
285            sb.push('@');
286            sb.push_str(RequestType::Stream.to_string().as_str());
287        }
288
289        sb
290    }
291
292    #[inline]
293    pub fn get_namesrv_addr(&self) -> Option<CheetahString> {
294        if let Ok(Some(targets)) = self.normalized_namesrv_targets() {
295            return Some(CheetahString::from_slice(targets.canonical()));
296        }
297        if let Some(namesrv_addr) = self
298            .namesrv_addr
299            .as_ref()
300            .filter(|addr| !addr.is_empty() && NameServerAddressUtils::is_name_srv_endpoint(addr.as_str()))
301        {
302            NameServerAddressUtils::get_name_srv_addr_from_namesrv_endpoint(namesrv_addr.as_str())
303                .map(|addr| addr.into())
304        } else {
305            self.namesrv_addr.clone()
306        }
307    }
308
309    pub(crate) fn normalized_namesrv_targets(&self) -> ClientResult<Option<NormalizedNameServerTargets>> {
310        self.namesrv_addr.as_deref().map(parse_legacy_namesrv_addr).transpose()
311    }
312
313    pub(crate) fn normalize_namesrv_addr(&mut self) -> ClientResult<()> {
314        if let Some(targets) = self.normalized_namesrv_targets()? {
315            self.namesrv_addr = Some(CheetahString::from_slice(targets.canonical()));
316        }
317        Ok(())
318    }
319
320    // ============ Comprehensive Getters and Setters ============
321
322    #[inline]
323    pub fn get_client_ip(&self) -> Option<&CheetahString> {
324        self.client_ip.as_ref()
325    }
326
327    #[inline]
328    pub fn set_client_ip(&mut self, client_ip: CheetahString) {
329        self.client_ip = Some(client_ip);
330    }
331
332    #[inline]
333    pub fn get_instance_name(&self) -> &CheetahString {
334        &self.instance_name
335    }
336
337    #[inline]
338    pub fn get_client_callback_executor_threads(&self) -> usize {
339        self.client_callback_executor_threads
340    }
341
342    #[inline]
343    pub fn set_client_callback_executor_threads(&mut self, threads: usize) {
344        self.client_callback_executor_threads = threads;
345    }
346
347    #[inline]
348    pub fn get_namespace_v2(&self) -> Option<&CheetahString> {
349        self.namespace_v2.as_ref()
350    }
351
352    #[inline]
353    pub fn set_namespace_v2(&mut self, namespace_v2: CheetahString) {
354        self.namespace_v2 = Some(namespace_v2);
355    }
356
357    #[inline]
358    pub fn set_namespace(&mut self, namespace: CheetahString) {
359        self.namespace = Some(namespace);
360        self.namespace_initialized.store(true, Ordering::Release);
361    }
362
363    #[inline]
364    pub fn get_access_channel(&self) -> AccessChannel {
365        self.access_channel
366    }
367
368    #[inline]
369    pub fn set_access_channel(&mut self, access_channel: AccessChannel) {
370        self.access_channel = access_channel;
371    }
372
373    #[inline]
374    pub fn get_poll_name_server_interval(&self) -> u32 {
375        self.poll_name_server_interval
376    }
377
378    #[inline]
379    pub fn set_poll_name_server_interval(&mut self, interval: u32) {
380        self.poll_name_server_interval = interval;
381    }
382
383    #[inline]
384    pub fn get_heartbeat_broker_interval(&self) -> u32 {
385        self.heartbeat_broker_interval
386    }
387
388    #[inline]
389    pub fn set_heartbeat_broker_interval(&mut self, interval: u32) {
390        self.heartbeat_broker_interval = interval;
391    }
392
393    #[inline]
394    pub fn get_persist_consumer_offset_interval(&self) -> u32 {
395        self.persist_consumer_offset_interval
396    }
397
398    #[inline]
399    pub fn set_persist_consumer_offset_interval(&mut self, interval: u32) {
400        self.persist_consumer_offset_interval = interval;
401    }
402
403    #[inline]
404    pub fn get_pull_time_delay_millis_when_exception(&self) -> u32 {
405        self.pull_time_delay_millis_when_exception
406    }
407
408    #[inline]
409    pub fn get_pull_time_delay_mills_when_exception(&self) -> u32 {
410        self.get_pull_time_delay_millis_when_exception()
411    }
412
413    #[inline]
414    pub fn set_pull_time_delay_millis_when_exception(&mut self, delay: u32) {
415        self.pull_time_delay_millis_when_exception = delay;
416    }
417
418    #[inline]
419    pub fn set_pull_time_delay_mills_when_exception(&mut self, delay: u32) {
420        self.set_pull_time_delay_millis_when_exception(delay);
421    }
422
423    #[inline]
424    pub fn get_unit_name(&self) -> Option<&CheetahString> {
425        self.unit_name.as_ref()
426    }
427
428    #[inline]
429    pub fn set_unit_name(&mut self, unit_name: CheetahString) {
430        self.unit_name = Some(unit_name);
431    }
432
433    #[inline]
434    pub fn is_unit_mode(&self) -> bool {
435        self.unit_mode
436    }
437
438    #[inline]
439    pub fn set_unit_mode(&mut self, unit_mode: bool) {
440        self.unit_mode = unit_mode;
441    }
442
443    #[inline]
444    pub fn is_decode_read_body(&self) -> bool {
445        self.decode_read_body
446    }
447
448    #[inline]
449    pub fn set_decode_read_body(&mut self, decode_read_body: bool) {
450        self.decode_read_body = decode_read_body;
451    }
452
453    #[inline]
454    pub fn is_decode_decompress_body(&self) -> bool {
455        self.decode_decompress_body
456    }
457
458    #[inline]
459    pub fn set_decode_decompress_body(&mut self, decode_decompress_body: bool) {
460        self.decode_decompress_body = decode_decompress_body;
461    }
462
463    #[inline]
464    pub fn is_vip_channel_enabled(&self) -> bool {
465        self.vip_channel_enabled
466    }
467
468    #[inline]
469    pub fn set_vip_channel_enabled(&mut self, enabled: bool) {
470        self.vip_channel_enabled = enabled;
471    }
472
473    #[inline]
474    pub fn is_use_heartbeat_v2(&self) -> bool {
475        self.use_heartbeat_v2
476    }
477
478    #[inline]
479    pub fn set_use_heartbeat_v2(&mut self, use_heartbeat_v2: bool) {
480        self.use_heartbeat_v2 = use_heartbeat_v2;
481    }
482
483    #[inline]
484    pub fn is_use_tls(&self) -> bool {
485        self.use_tls
486    }
487
488    #[inline]
489    pub fn set_use_tls(&mut self, use_tls: bool) {
490        self.use_tls = use_tls;
491        self.tls_config.enable = use_tls;
492    }
493
494    #[inline]
495    pub fn tls_config(&self) -> &TlsConfig {
496        &self.tls_config
497    }
498
499    #[inline]
500    pub fn set_tls_config(&mut self, mut tls_config: TlsConfig) {
501        tls_config.enable = self.use_tls;
502        self.tls_config = tls_config;
503    }
504
505    #[inline]
506    pub fn set_tls_test_mode_enable(&mut self, enabled: bool) {
507        self.tls_config.test_mode_enable = enabled;
508    }
509
510    #[inline]
511    pub fn set_tls_client_auth_server(&mut self, enabled: bool) {
512        self.tls_config.client.auth_server = enabled;
513    }
514
515    #[inline]
516    pub fn set_tls_client_trust_cert_path(&mut self, path: impl Into<String>) {
517        self.tls_config.client.trust_cert_path = Some(path.into());
518    }
519
520    #[inline]
521    pub fn set_tls_client_cert_path(&mut self, path: impl Into<String>) {
522        self.tls_config.client.cert_path = Some(path.into());
523    }
524
525    #[inline]
526    pub fn set_tls_client_key_path(&mut self, path: impl Into<String>) {
527        self.tls_config.client.key_path = Some(path.into());
528    }
529
530    #[inline]
531    pub fn get_socks_proxy_config(&self) -> &CheetahString {
532        &self.socks_proxy_config
533    }
534
535    #[inline]
536    pub fn set_socks_proxy_config(&mut self, config: CheetahString) {
537        self.socks_proxy_config = config;
538    }
539
540    /// Parses the Java-compatible SOCKS proxy JSON into the transport-owned typed model.
541    pub fn parse_socks_proxy_config(&self) -> ClientResult<SocksProxyConfig> {
542        SocksProxyConfig::parse_java_json(&self.socks_proxy_config).map_err(ClientError::from)
543    }
544
545    #[inline]
546    pub fn get_language(&self) -> LanguageCode {
547        self.language
548    }
549
550    #[inline]
551    pub fn set_language(&mut self, language: LanguageCode) {
552        self.language = language;
553    }
554
555    #[inline]
556    pub fn get_mq_client_api_timeout(&self) -> u64 {
557        self.mq_client_api_timeout
558    }
559
560    #[inline]
561    pub fn set_mq_client_api_timeout(&mut self, timeout: u64) {
562        self.mq_client_api_timeout = timeout;
563    }
564
565    #[inline]
566    pub fn get_detect_timeout(&self) -> u32 {
567        self.detect_timeout
568    }
569
570    #[inline]
571    pub fn set_detect_timeout(&mut self, timeout: u32) {
572        self.detect_timeout = timeout;
573    }
574
575    #[inline]
576    pub fn get_detect_interval(&self) -> u32 {
577        self.detect_interval
578    }
579
580    #[inline]
581    pub fn set_detect_interval(&mut self, interval: u32) {
582        self.detect_interval = interval;
583    }
584
585    #[inline]
586    pub fn is_enable_stream_request_type(&self) -> bool {
587        self.enable_stream_request_type
588    }
589
590    #[inline]
591    pub fn set_enable_stream_request_type(&mut self, enabled: bool) {
592        self.enable_stream_request_type = enabled;
593    }
594
595    #[inline]
596    pub fn is_send_latency_enable(&self) -> bool {
597        self.send_latency_enable
598    }
599
600    #[inline]
601    pub fn set_send_latency_enable(&mut self, enabled: bool) {
602        self.send_latency_enable = enabled;
603    }
604
605    #[inline]
606    pub fn is_start_detector_enable(&self) -> bool {
607        self.start_detector_enable
608    }
609
610    #[inline]
611    pub fn set_start_detector_enable(&mut self, enabled: bool) {
612        self.start_detector_enable = enabled;
613    }
614
615    #[inline]
616    pub fn is_enable_heartbeat_channel_event_listener(&self) -> bool {
617        self.enable_heartbeat_channel_event_listener
618    }
619
620    #[inline]
621    pub fn set_enable_heartbeat_channel_event_listener(&mut self, enabled: bool) {
622        self.enable_heartbeat_channel_event_listener = enabled;
623    }
624
625    #[inline]
626    pub fn is_enable_trace(&self) -> bool {
627        self.enable_trace
628    }
629
630    #[inline]
631    pub fn set_enable_trace(&mut self, enabled: bool) {
632        self.enable_trace = enabled;
633    }
634
635    #[inline]
636    pub fn get_trace_topic(&self) -> Option<&CheetahString> {
637        self.trace_topic.as_ref()
638    }
639
640    #[inline]
641    pub fn set_trace_topic(&mut self, topic: CheetahString) {
642        self.trace_topic = Some(topic);
643    }
644
645    #[inline]
646    pub fn get_trace_msg_batch_num(&self) -> usize {
647        self.trace_msg_batch_num
648    }
649
650    #[inline]
651    pub fn set_trace_msg_batch_num(&mut self, num: usize) {
652        self.trace_msg_batch_num = num;
653    }
654
655    #[inline]
656    pub fn get_max_page_size_in_get_metadata(&self) -> usize {
657        self.max_page_size_in_get_metadata
658    }
659
660    #[inline]
661    pub fn set_max_page_size_in_get_metadata(&mut self, size: usize) {
662        self.max_page_size_in_get_metadata = size;
663    }
664
665    #[inline]
666    pub fn get_concurrent_heartbeat_thread_pool_size(&self) -> usize {
667        self.concurrent_heartbeat_thread_pool_size
668    }
669
670    #[inline]
671    pub fn is_enable_concurrent_heartbeat(&self) -> bool {
672        self.enable_concurrent_heartbeat
673    }
674
675    #[inline]
676    pub fn set_enable_concurrent_heartbeat(&mut self, enabled: bool) {
677        self.enable_concurrent_heartbeat = enabled;
678    }
679
680    #[inline]
681    pub fn set_concurrent_heartbeat_thread_pool_size(&mut self, size: usize) {
682        self.concurrent_heartbeat_thread_pool_size = size;
683    }
684
685    #[inline]
686    pub fn get_pull_message_service_shards(&self) -> usize {
687        self.pull_message_service_shards
688    }
689
690    #[inline]
691    pub fn set_pull_message_service_shards(&mut self, shards: usize) {
692        self.pull_message_service_shards = shards;
693    }
694
695    // ============ Utility Methods ============
696
697    /// Clones the configuration
698    #[inline]
699    pub fn clone_client_config(&self) -> Self {
700        self.clone()
701    }
702
703    /// Resets client config from another instance
704    pub fn reset_client_config(&mut self, other: &ClientConfig) {
705        self.namesrv_addr = other.namesrv_addr.clone();
706        self.client_ip = other.client_ip.clone();
707        self.instance_name = other.instance_name.clone();
708        self.client_callback_executor_threads = other.client_callback_executor_threads;
709        self.namespace = other.namespace.clone();
710        self.namespace_v2 = other.namespace_v2.clone();
711        self.access_channel = other.access_channel;
712        self.poll_name_server_interval = other.poll_name_server_interval;
713        self.heartbeat_broker_interval = other.heartbeat_broker_interval;
714        self.persist_consumer_offset_interval = other.persist_consumer_offset_interval;
715        self.pull_time_delay_millis_when_exception = other.pull_time_delay_millis_when_exception;
716        self.unit_mode = other.unit_mode;
717        self.unit_name = other.unit_name.clone();
718        self.decode_read_body = other.decode_read_body;
719        self.decode_decompress_body = other.decode_decompress_body;
720        self.vip_channel_enabled = other.vip_channel_enabled;
721        self.use_heartbeat_v2 = other.use_heartbeat_v2;
722        self.use_tls = other.use_tls;
723        self.tls_config = other.tls_config.clone();
724        self.socks_proxy_config = other.socks_proxy_config.clone();
725        self.language = other.language;
726        self.mq_client_api_timeout = other.mq_client_api_timeout;
727        self.detect_timeout = other.detect_timeout;
728        self.detect_interval = other.detect_interval;
729        self.enable_stream_request_type = other.enable_stream_request_type;
730        self.send_latency_enable = other.send_latency_enable;
731        self.start_detector_enable = other.start_detector_enable;
732        self.enable_heartbeat_channel_event_listener = other.enable_heartbeat_channel_event_listener;
733        self.enable_trace = other.enable_trace;
734        self.trace_topic = other.trace_topic.clone();
735        self.trace_msg_batch_num = other.trace_msg_batch_num;
736        self.max_page_size_in_get_metadata = other.max_page_size_in_get_metadata;
737        self.enable_concurrent_heartbeat = other.enable_concurrent_heartbeat;
738        self.concurrent_heartbeat_thread_pool_size = other.concurrent_heartbeat_thread_pool_size;
739        self.pull_message_service_shards = other.pull_message_service_shards;
740    }
741
742    /// Deprecated: Use with_namespace instead
743    #[inline]
744    #[deprecated(note = "Use with_namespace for namespace wrapping")]
745    pub fn without_namespace(&mut self, resource: &str) -> CheetahString {
746        if let Some(namespace) = self.get_namespace().as_deref() {
747            NamespaceUtil::without_namespace_with_namespace(resource, namespace).into()
748        } else {
749            NamespaceUtil::without_namespace(resource).into()
750        }
751    }
752
753    /// Creates a new builder for constructing a ClientConfig with a fluent API
754    ///
755    /// # Example
756    ///
757    /// ```rust
758    /// use rocketmq_client_rust::ClientConfig;
759    ///
760    /// let config = ClientConfig::builder()
761    ///     .namesrv_addr("localhost:9876")
762    ///     .instance_name("my_producer")
763    ///     .enable_tls(true)
764    ///     .build()?;
765    /// # Ok::<(), rocketmq_client_rust::ClientError>(())
766    /// ```
767    #[inline]
768    pub fn builder() -> crate::base::client_config_builder::ClientConfigBuilder {
769        crate::base::client_config_builder::ClientConfigBuilder::new()
770    }
771}
772
773impl std::fmt::Display for ClientConfig {
774    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
775        write!(
776            f,
777            "ClientConfig {{ namesrv_addr: {:?}, client_ip: {:?}, instance_name: {}, \
778             client_callback_executor_threads: {}, namespace: {:?}, namespace_v2: {:?}, access_channel: {}, \
779             poll_name_server_interval: {}, heartbeat_broker_interval: {}, persist_consumer_offset_interval: {}, \
780             pull_time_delay_millis_when_exception: {}, unit_mode: {}, unit_name: {:?}, decode_read_body: {}, \
781             decode_decompress_body: {}, vip_channel_enabled: {}, use_heartbeat_v2: {}, use_tls: {}, \
782             socks_proxy_config: {}, mq_client_api_timeout: {}, detect_timeout: {}, detect_interval: {}, language: \
783             {:?}, enable_stream_request_type: {}, send_latency_enable: {}, start_detector_enable: {}, \
784             enable_heartbeat_channel_event_listener: {}, enable_trace: {}, trace_topic: {:?}, trace_msg_batch_num: \
785             {}, max_page_size_in_get_metadata: {}, enable_concurrent_heartbeat: {}, \
786             concurrent_heartbeat_thread_pool_size: {}, pull_message_service_shards: {} }}",
787            self.namesrv_addr,
788            self.client_ip,
789            self.instance_name,
790            self.client_callback_executor_threads,
791            self.namespace,
792            self.namespace_v2,
793            self.access_channel,
794            self.poll_name_server_interval,
795            self.heartbeat_broker_interval,
796            self.persist_consumer_offset_interval,
797            self.pull_time_delay_millis_when_exception,
798            self.unit_mode,
799            self.unit_name,
800            self.decode_read_body,
801            self.decode_decompress_body,
802            self.vip_channel_enabled,
803            self.use_heartbeat_v2,
804            self.use_tls,
805            self.socks_proxy_config,
806            self.mq_client_api_timeout,
807            self.detect_timeout,
808            self.detect_interval,
809            self.language,
810            self.enable_stream_request_type,
811            self.send_latency_enable,
812            self.start_detector_enable,
813            self.enable_heartbeat_channel_event_listener,
814            self.enable_trace,
815            self.trace_topic,
816            self.trace_msg_batch_num,
817            self.max_page_size_in_get_metadata,
818            self.enable_concurrent_heartbeat,
819            self.concurrent_heartbeat_thread_pool_size,
820            self.pull_message_service_shards
821        )
822    }
823}
824
825#[cfg(test)]
826mod tests {
827    use super::*;
828
829    #[test]
830    fn java_compatible_aliases_delegate_to_client_config_fields() {
831        let mut config = ClientConfig::default();
832
833        config.set_pull_time_delay_mills_when_exception(2345);
834        assert_eq!(config.get_pull_time_delay_mills_when_exception(), 2345);
835        assert_eq!(config.get_pull_time_delay_millis_when_exception(), 2345);
836
837        config.set_enable_concurrent_heartbeat(false);
838        assert!(!config.is_enable_concurrent_heartbeat());
839        config.set_enable_concurrent_heartbeat(true);
840        assert!(config.is_enable_concurrent_heartbeat());
841    }
842
843    #[test]
844    fn reset_client_config_copies_modern_java_fields() {
845        let mut source = ClientConfig::default();
846        source.set_use_tls(true);
847        source.set_enable_trace(true);
848        source.set_trace_topic(CheetahString::from("TraceTopicA"));
849        source.set_trace_msg_batch_num(32);
850        source.set_max_page_size_in_get_metadata(4096);
851        source.set_enable_concurrent_heartbeat(true);
852        source.set_concurrent_heartbeat_thread_pool_size(8);
853        source.set_pull_message_service_shards(4);
854
855        let mut target = ClientConfig::default();
856        target.set_use_tls(false);
857        target.set_enable_trace(false);
858        target.trace_topic = None;
859        target.set_trace_msg_batch_num(10);
860        target.set_max_page_size_in_get_metadata(2000);
861        target.set_enable_concurrent_heartbeat(false);
862        target.set_concurrent_heartbeat_thread_pool_size(1);
863        target.set_pull_message_service_shards(1);
864
865        target.reset_client_config(&source);
866
867        assert!(target.is_use_tls());
868        assert!(target.tls_config().enable);
869        assert!(target.is_enable_trace());
870        assert_eq!(
871            target.get_trace_topic().map(|topic| topic.as_str()),
872            Some("TraceTopicA")
873        );
874        assert_eq!(target.get_trace_msg_batch_num(), 32);
875        assert_eq!(target.get_max_page_size_in_get_metadata(), 4096);
876        assert!(target.is_enable_concurrent_heartbeat());
877        assert_eq!(target.get_concurrent_heartbeat_thread_pool_size(), 8);
878        assert_eq!(target.get_pull_message_service_shards(), 4);
879    }
880
881    #[test]
882    fn display_uses_java_access_channel_name() {
883        let mut config = ClientConfig::default();
884        config.set_access_channel(AccessChannel::Cloud);
885
886        let rendered = config.to_string();
887
888        assert!(rendered.contains("access_channel: CLOUD"));
889        assert!(!rendered.contains("access_channel: Cloud"));
890    }
891
892    #[test]
893    fn queues_with_namespace_wraps_each_queue_like_java_collection_helper() {
894        let mut config = ClientConfig::default();
895        let queues = vec![
896            MessageQueue::from_parts("topic_a", "broker-a", 0),
897            MessageQueue::from_parts("ns%topic_b", "broker-a", 1),
898        ];
899
900        let unchanged = config.queues_with_namespace(queues.clone());
901        assert_eq!(unchanged[0].topic_str(), "topic_a");
902        assert_eq!(unchanged[1].topic_str(), "ns%topic_b");
903
904        config.namespace = Some(CheetahString::from("ns"));
905        config.namespace_initialized.store(false, Ordering::Release);
906
907        let wrapped = config.queues_with_namespace(queues);
908        assert_eq!(wrapped[0].topic_str(), "ns%topic_a");
909        assert_eq!(wrapped[1].topic_str(), "ns%topic_b");
910    }
911
912    #[test]
913    fn immutable_namespace_resolution_wraps_queue_without_populating_cache() {
914        let mut config = ClientConfig {
915            namespace: Some(CheetahString::from("ns")),
916            ..Default::default()
917        };
918
919        let wrapped = config.queue_with_resolved_namespace(MessageQueue::from_parts("topic_a", "broker-a", 0));
920
921        assert_eq!(wrapped.topic_str(), "ns%topic_a");
922        assert!(!config.namespace_initialized.load(Ordering::Acquire));
923        assert_eq!(config.get_namespace().as_deref(), Some("ns"));
924        assert!(config.namespace_initialized.load(Ordering::Acquire));
925    }
926}
rocketmq_client_rust/base/client_config.rs

rocketmq_client_rust/base/
client_config.rs
