rocketmq_client_rust/base/
client_config_builder.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//! ClientConfig builder module
16//!
17//! This module provides a fluent builder API for constructing ClientConfig instances,
18//! making configuration more readable and allowing for validation before building.
19
20use cheetah_string::CheetahString;
21use rocketmq_common::common::tls_config::TlsConfig;
22use rocketmq_error::RocketMQResult;
23use rocketmq_remoting::protocol::LanguageCode;
24
25use super::access_channel::AccessChannel;
26use super::client_config::ClientConfig;
27use super::client_config_validation::ClientConfigValidator;
28
29/// Builder for creating [`ClientConfig`] instances with a fluent API
30///
31/// # Example
32///
33/// ```rust
34/// use rocketmq_client_rust::base::client_config::ClientConfig;
35///
36/// let config = ClientConfig::builder()
37///     .namesrv_addr("localhost:9876")
38///     .instance_name("my_producer")
39///     .enable_tls(true)
40///     .poll_name_server_interval(60_000)
41///     .build()?;
42/// # Ok::<(), rocketmq_error::RocketMQError>(())
43/// ```
44pub struct ClientConfigBuilder {
45    config: ClientConfig,
46}
47
48impl Default for ClientConfigBuilder {
49    fn default() -> Self {
50        Self::new()
51    }
52}
53
54impl ClientConfigBuilder {
55    /// Creates a new builder with default configuration
56    pub fn new() -> Self {
57        Self {
58            config: ClientConfig::new(),
59        }
60    }
61
62    // ========================================================================
63    // Name server configuration
64    // ========================================================================
65
66    /// Sets the name server address
67    pub fn namesrv_addr(mut self, addr: impl Into<CheetahString>) -> Self {
68        self.config.set_namesrv_addr(addr.into());
69        self
70    }
71
72    // ========================================================================
73    // Client identification
74    // ========================================================================
75
76    /// Sets the client IP address
77    pub fn client_ip(mut self, ip: impl Into<CheetahString>) -> Self {
78        self.config.set_client_ip(ip.into());
79        self
80    }
81
82    /// Sets the instance name
83    pub fn instance_name(mut self, name: impl Into<CheetahString>) -> Self {
84        self.config.set_instance_name(name.into());
85        self
86    }
87
88    // ========================================================================
89    // Thread pool configuration--only java client has these options,
90    // but we keep them for compatibility and future use
91    // ========================================================================
92
93    /// Sets the number of client callback executor threads
94    pub fn client_callback_executor_threads(mut self, threads: usize) -> Self {
95        self.config.set_client_callback_executor_threads(threads);
96        self
97    }
98
99    /// Sets the concurrent heartbeat thread pool size
100    pub fn concurrent_heartbeat_thread_pool_size(mut self, size: usize) -> Self {
101        self.config.set_concurrent_heartbeat_thread_pool_size(size);
102        self
103    }
104
105    // ========================================================================
106    // Namespace configuration
107    // ========================================================================
108
109    /// Sets the namespace (deprecated in Java, kept for compatibility)
110    pub fn namespace(mut self, namespace: impl Into<CheetahString>) -> Self {
111        self.config.set_namespace(namespace.into());
112        self
113    }
114
115    /// Sets the namespace v2
116    pub fn namespace_v2(mut self, namespace: impl Into<CheetahString>) -> Self {
117        self.config.set_namespace_v2(namespace.into());
118        self
119    }
120
121    // ========================================================================
122    // Access channel
123    // ========================================================================
124
125    /// Sets the access channel (Local or Cloud)
126    pub fn access_channel(mut self, channel: AccessChannel) -> Self {
127        self.config.set_access_channel(channel);
128        self
129    }
130
131    // ========================================================================
132    // Interval configuration
133    // ========================================================================
134
135    /// Sets the poll name server interval in milliseconds
136    pub fn poll_name_server_interval(mut self, millis: u32) -> Self {
137        self.config.set_poll_name_server_interval(millis);
138        self
139    }
140
141    /// Sets the heartbeat broker interval in milliseconds
142    pub fn heartbeat_broker_interval(mut self, millis: u32) -> Self {
143        self.config.set_heartbeat_broker_interval(millis);
144        self
145    }
146
147    /// Sets the persist consumer offset interval in milliseconds
148    pub fn persist_consumer_offset_interval(mut self, millis: u32) -> Self {
149        self.config.set_persist_consumer_offset_interval(millis);
150        self
151    }
152
153    // ========================================================================
154    // Timeout configuration
155    // ========================================================================
156
157    /// Sets the pull time delay in milliseconds when exception occurs
158    pub fn pull_time_delay_millis_when_exception(mut self, millis: u32) -> Self {
159        self.config.set_pull_time_delay_millis_when_exception(millis);
160        self
161    }
162
163    /// Sets the MQ client API timeout in milliseconds
164    pub fn mq_client_api_timeout(mut self, millis: u64) -> Self {
165        self.config.set_mq_client_api_timeout(millis);
166        self
167    }
168
169    /// Sets the detect timeout in milliseconds
170    pub fn detect_timeout(mut self, millis: u32) -> Self {
171        self.config.set_detect_timeout(millis);
172        self
173    }
174
175    /// Sets the detect interval in milliseconds
176    pub fn detect_interval(mut self, millis: u32) -> Self {
177        self.config.set_detect_interval(millis);
178        self
179    }
180
181    // ========================================================================
182    // Boolean flags
183    // ========================================================================
184
185    /// Sets unit mode enabled
186    pub fn enable_unit_mode(mut self, enabled: bool) -> Self {
187        self.config.set_unit_mode(enabled);
188        self
189    }
190
191    /// Sets VIP channel enabled
192    pub fn enable_vip_channel(mut self, enabled: bool) -> Self {
193        self.config.set_vip_channel_enabled(enabled);
194        self
195    }
196
197    /// Sets heartbeat v2 enabled
198    pub fn enable_heartbeat_v2(mut self, enabled: bool) -> Self {
199        self.config.set_use_heartbeat_v2(enabled);
200        self
201    }
202
203    /// Sets concurrent heartbeat enabled
204    pub fn enable_concurrent_heartbeat(mut self, enabled: bool) -> Self {
205        self.config.enable_concurrent_heartbeat = enabled;
206        self
207    }
208
209    /// Sets TLS enabled
210    pub fn enable_tls(mut self, enabled: bool) -> Self {
211        self.config.set_use_tls(enabled);
212        self
213    }
214
215    /// Sets the complete TLS configuration while preserving the builder's enable_tls flag.
216    pub fn tls_config(mut self, tls_config: TlsConfig) -> Self {
217        self.config.set_tls_config(tls_config);
218        self
219    }
220
221    /// Enables RocketMQ Java-compatible TLS test mode.
222    pub fn tls_test_mode_enable(mut self, enabled: bool) -> Self {
223        self.config.set_tls_test_mode_enable(enabled);
224        self
225    }
226
227    /// Controls whether the client verifies the server certificate.
228    pub fn tls_client_auth_server(mut self, enabled: bool) -> Self {
229        self.config.set_tls_client_auth_server(enabled);
230        self
231    }
232
233    /// Sets the PEM trust certificate path for server verification.
234    pub fn tls_client_trust_cert_path(mut self, path: impl Into<String>) -> Self {
235        self.config.set_tls_client_trust_cert_path(path);
236        self
237    }
238
239    /// Sets the PEM client certificate path for mTLS.
240    pub fn tls_client_cert_path(mut self, path: impl Into<String>) -> Self {
241        self.config.set_tls_client_cert_path(path);
242        self
243    }
244
245    /// Sets the PEM client private key path for mTLS.
246    pub fn tls_client_key_path(mut self, path: impl Into<String>) -> Self {
247        self.config.set_tls_client_key_path(path);
248        self
249    }
250
251    /// Sets decode read body enabled
252    pub fn enable_decode_read_body(mut self, enabled: bool) -> Self {
253        self.config.set_decode_read_body(enabled);
254        self
255    }
256
257    /// Sets decode decompress body enabled
258    pub fn enable_decode_decompress_body(mut self, enabled: bool) -> Self {
259        self.config.set_decode_decompress_body(enabled);
260        self
261    }
262
263    /// Sets stream request type enabled
264    pub fn enable_stream_request_type(mut self, enabled: bool) -> Self {
265        self.config.set_enable_stream_request_type(enabled);
266        self
267    }
268
269    /// Sets send latency enabled
270    pub fn enable_send_latency(mut self, enabled: bool) -> Self {
271        self.config.set_send_latency_enable(enabled);
272        self
273    }
274
275    /// Sets start detector enabled
276    pub fn enable_start_detector(mut self, enabled: bool) -> Self {
277        self.config.set_start_detector_enable(enabled);
278        self
279    }
280
281    /// Sets heartbeat channel event listener enabled
282    pub fn enable_heartbeat_channel_event_listener(mut self, enabled: bool) -> Self {
283        self.config.set_enable_heartbeat_channel_event_listener(enabled);
284        self
285    }
286
287    /// Sets trace enabled
288    pub fn enable_trace(mut self, enabled: bool) -> Self {
289        self.config.set_enable_trace(enabled);
290        self
291    }
292
293    // ========================================================================
294    // Other configuration
295    // ========================================================================
296
297    /// Sets the unit name
298    pub fn unit_name(mut self, name: impl Into<CheetahString>) -> Self {
299        self.config.set_unit_name(name.into());
300        self
301    }
302
303    /// Sets the SOCKS proxy config
304    pub fn socks_proxy_config(mut self, config: impl Into<CheetahString>) -> Self {
305        self.config.set_socks_proxy_config(config.into());
306        self
307    }
308
309    /// Sets the language code
310    pub fn language(mut self, language: LanguageCode) -> Self {
311        self.config.set_language(language);
312        self
313    }
314
315    /// Sets the trace topic
316    pub fn trace_topic(mut self, topic: impl Into<CheetahString>) -> Self {
317        self.config.set_trace_topic(topic.into());
318        self
319    }
320
321    /// Sets the trace message batch number
322    pub fn trace_msg_batch_num(mut self, num: usize) -> Self {
323        self.config.set_trace_msg_batch_num(num);
324        self
325    }
326
327    /// Sets the max page size in get metadata
328    pub fn max_page_size_in_get_metadata(mut self, size: usize) -> Self {
329        self.config.set_max_page_size_in_get_metadata(size);
330        self
331    }
332
333    // ========================================================================
334    // Build methods
335    // ========================================================================
336
337    /// Builds the ClientConfig with validation
338    ///
339    /// # Errors
340    ///
341    /// Returns an error if any configuration value is invalid.
342    pub fn build(self) -> RocketMQResult<ClientConfig> {
343        // Validate configuration before returning
344        self.validate()?;
345        Ok(self.config)
346    }
347
348    /// Builds the ClientConfig without validation (internal use only)
349    ///
350    /// # Safety
351    ///
352    /// This method skips validation and should only be used when you're
353    /// certain all values are correct.
354    #[doc(hidden)]
355    pub fn build_unvalidated(self) -> ClientConfig {
356        self.config
357    }
358
359    /// Validates the current configuration
360    fn validate(&self) -> RocketMQResult<()> {
361        // Validate intervals
362        ClientConfigValidator::validate_poll_name_server_interval(self.config.poll_name_server_interval)?;
363        ClientConfigValidator::validate_heartbeat_broker_interval(self.config.heartbeat_broker_interval)?;
364        ClientConfigValidator::validate_persist_consumer_offset_interval(self.config.persist_consumer_offset_interval)?;
365
366        // Validate timeouts
367        ClientConfigValidator::validate_mq_client_api_timeout(self.config.mq_client_api_timeout)?;
368
369        // Validate numeric limits
370        ClientConfigValidator::validate_trace_msg_batch_num(self.config.trace_msg_batch_num)?;
371        ClientConfigValidator::validate_max_page_size_in_get_metadata(self.config.max_page_size_in_get_metadata)?;
372
373        // Validate thread pool sizes
374        ClientConfigValidator::validate_client_callback_executor_threads(self.config.client_callback_executor_threads)?;
375
376        // Only validate concurrent heartbeat thread pool size if concurrent heartbeat is enabled
377        if self.config.enable_concurrent_heartbeat {
378            ClientConfigValidator::validate_concurrent_heartbeat_thread_pool_size(
379                self.config.concurrent_heartbeat_thread_pool_size,
380            )?;
381        }
382
383        Ok(())
384    }
385}
386
387#[cfg(test)]
388mod tests {
389    use super::*;
390
391    #[test]
392    fn test_builder_basic() {
393        let config = ClientConfig::builder()
394            .namesrv_addr("localhost:9876")
395            .instance_name("test_instance")
396            .build()
397            .unwrap();
398
399        assert_eq!(config.get_namesrv_addr().unwrap().as_str(), "localhost:9876");
400        assert_eq!(config.get_instance_name().as_str(), "test_instance");
401    }
402
403    #[test]
404    fn test_builder_with_all_options() {
405        let config = ClientConfig::builder()
406            .namesrv_addr("localhost:9876")
407            .instance_name("test")
408            .client_ip("127.0.0.1")
409            .poll_name_server_interval(60_000)
410            .heartbeat_broker_interval(30_000)
411            .enable_tls(true)
412            .enable_concurrent_heartbeat(true)
413            .concurrent_heartbeat_thread_pool_size(4)
414            .build()
415            .unwrap();
416
417        assert_eq!(config.get_namesrv_addr().unwrap().as_str(), "localhost:9876");
418        assert_eq!(config.get_instance_name().as_str(), "test");
419        assert_eq!(config.get_client_ip().unwrap().as_str(), "127.0.0.1");
420        assert_eq!(config.poll_name_server_interval, 60_000);
421        assert_eq!(config.heartbeat_broker_interval, 30_000);
422        assert!(config.use_tls);
423        assert!(config.tls_config.enable);
424        assert!(config.enable_concurrent_heartbeat);
425        assert_eq!(config.concurrent_heartbeat_thread_pool_size, 4);
426    }
427
428    #[test]
429    fn builder_sets_tls_certificate_fields() {
430        let config = ClientConfig::builder()
431            .enable_tls(true)
432            .tls_client_auth_server(false)
433            .tls_client_trust_cert_path("/certs/ca.pem")
434            .tls_client_cert_path("/certs/client.pem")
435            .tls_client_key_path("/certs/client.key")
436            .build()
437            .unwrap();
438
439        assert!(config.use_tls);
440        assert!(config.tls_config.enable);
441        assert!(!config.tls_config.client.auth_server);
442        assert_eq!(
443            config.tls_config.client.trust_cert_path.as_deref(),
444            Some("/certs/ca.pem")
445        );
446        assert_eq!(config.tls_config.client.cert_path.as_deref(), Some("/certs/client.pem"));
447        assert_eq!(config.tls_config.client.key_path.as_deref(), Some("/certs/client.key"));
448    }
449
450    #[test]
451    fn test_builder_validation_invalid_poll_interval() {
452        let result = ClientConfig::builder()
453            .poll_name_server_interval(100) // Too small
454            .build();
455
456        assert!(result.is_err());
457    }
458
459    #[test]
460    fn test_builder_validation_invalid_heartbeat_interval() {
461        let result = ClientConfig::builder()
462            .heartbeat_broker_interval(100) // Too small
463            .build();
464
465        assert!(result.is_err());
466    }
467
468    #[test]
469    fn test_builder_validation_invalid_thread_pool_size() {
470        let result = ClientConfig::builder()
471            .enable_concurrent_heartbeat(true)
472            .concurrent_heartbeat_thread_pool_size(0) // Invalid
473            .build();
474
475        assert!(result.is_err());
476    }
477
478    #[test]
479    fn test_builder_validation_invalid_trace_msg_batch_num() {
480        let result = ClientConfig::builder().trace_msg_batch_num(0).build();
481
482        assert!(result.is_err());
483    }
484
485    #[test]
486    fn test_builder_unvalidated() {
487        // This should not panic even with invalid values
488        let config = ClientConfig::builder()
489            .poll_name_server_interval(100) // Invalid
490            .build_unvalidated();
491
492        assert_eq!(config.poll_name_server_interval, 100);
493    }
494}
rocketmq_client_rust/base/client_config_builder.rs

rocketmq_client_rust/base/
client_config_builder.rs
