rocketmq_client_rust/consumer/
default_lite_pull_consumer_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
15use std::sync::Arc;
16
17use crate::ClientResult;
18use cheetah_string::CheetahString;
19use rocketmq_model::common::consumer::consume_from_where::ConsumeFromWhere;
20use rocketmq_model::common::message::message_enum::MessageRequestMode;
21use rocketmq_model::common::mix_all;
22use rocketmq_protocol::protocol::heartbeat::message_model::MessageModel;
23use rocketmq_transport::api::RPCHook;
24
25use crate::base::client_options::ClientOptions;
26use crate::consumer::allocate_message_queue_strategy::AllocateMessageQueueStrategy;
27use crate::consumer::consumer_impl::default_lite_pull_consumer_impl::default_lite_pull_consume_timestamp;
28use crate::consumer::consumer_impl::default_lite_pull_consumer_impl::validate_lite_pull_consume_from_where;
29use crate::consumer::consumer_impl::default_lite_pull_consumer_impl::LitePullConsumerConfig;
30use crate::consumer::default_lite_pull_consumer::DefaultLitePullConsumer;
31use crate::consumer::rebalance_strategy::allocate_message_queue_averagely::AllocateMessageQueueAveragely;
32use crate::nameserver_discovery::NameServerDiscoveryConfig;
33use crate::runtime::ClientRuntime;
34use crate::trace::trace_dispatcher::TraceDispatcher;
35
36pub(crate) const MIN_AUTOCOMMIT_INTERVAL_MILLIS: u64 = 1000;
37
38/// Builder for creating a [`DefaultLitePullConsumer`] with customized configuration.
39///
40/// # Examples
41///
42/// ```rust,ignore
43/// use rocketmq_client::DefaultLitePullConsumer;
44///
45/// let consumer = DefaultLitePullConsumer::builder(client_runtime.clone())
46///     .consumer_group("my_consumer_group")
47///     .name_server_addr("127.0.0.1:9876")
48///     .pull_batch_size(32)
49///     .auto_commit(true)
50///     .build()?;
51/// ```
52pub struct DefaultLitePullConsumerBuilder {
53    client_runtime: Arc<ClientRuntime>,
54    client_options: Option<ClientOptions>,
55    // Client configuration
56    name_server_addr: Option<CheetahString>,
57    client_ip: Option<CheetahString>,
58    instance_name: Option<CheetahString>,
59    namespace: Option<CheetahString>,
60    access_channel: Option<CheetahString>,
61    use_tls: bool,
62
63    // Consumer configuration
64    consumer_group: Option<CheetahString>,
65    message_model: MessageModel,
66    consume_from_where: ConsumeFromWhere,
67    consume_timestamp: Option<CheetahString>,
68    allocate_message_queue_strategy: Arc<dyn AllocateMessageQueueStrategy + Send + Sync>,
69    unit_mode: bool,
70    classic_pull_manual_mode: bool,
71
72    // Pull configuration
73    pull_batch_size: i32,
74    pull_thread_nums: usize,
75    connect_broker_by_user: bool,
76    default_broker_id: u64,
77
78    // Flow control
79    pull_threshold_for_queue: i64,
80    pull_threshold_size_for_queue: i32,
81    pull_threshold_for_all: i64,
82    consume_max_span: i64,
83
84    // Backoff delays
85    pull_time_delay_millis_when_exception: u64,
86    pull_time_delay_millis_when_cache_flow_control: u64,
87    pull_time_delay_millis_when_broker_flow_control: u64,
88    broker_suspend_max_time_millis: u64,
89    consumer_timeout_millis_when_suspend: u64,
90    consumer_pull_timeout_millis: u64,
91
92    // Poll configuration
93    poll_timeout_millis: u64,
94
95    // Auto-commit
96    auto_commit: bool,
97    auto_commit_interval_millis: u64,
98
99    // Miscellaneous
100    topic_metadata_check_interval_millis: u64,
101    message_request_mode: MessageRequestMode,
102
103    // Advanced
104    rpc_hook: Option<Arc<dyn RPCHook>>,
105    trace_dispatcher: Option<Arc<dyn TraceDispatcher + Send + Sync>>,
106    enable_msg_trace: bool,
107    custom_trace_topic: Option<CheetahString>,
108}
109
110impl DefaultLitePullConsumerBuilder {
111    /// Creates a new builder with default configuration.
112    pub fn new(client_runtime: Arc<ClientRuntime>) -> Self {
113        Self {
114            client_runtime,
115            client_options: None,
116            name_server_addr: None,
117            client_ip: None,
118            instance_name: None,
119            namespace: None,
120            access_channel: None,
121            use_tls: false,
122            consumer_group: None,
123            message_model: MessageModel::Clustering,
124            consume_from_where: ConsumeFromWhere::ConsumeFromLastOffset,
125            consume_timestamp: Some(default_lite_pull_consume_timestamp()),
126            allocate_message_queue_strategy: Arc::new(AllocateMessageQueueAveragely),
127            unit_mode: false,
128            classic_pull_manual_mode: false,
129            pull_batch_size: 10,
130            pull_thread_nums: 20,
131            connect_broker_by_user: false,
132            default_broker_id: mix_all::MASTER_ID,
133            pull_threshold_for_queue: 1000,
134            pull_threshold_size_for_queue: 100,
135            pull_threshold_for_all: 10000,
136            consume_max_span: 2000,
137            pull_time_delay_millis_when_exception: 1000,
138            pull_time_delay_millis_when_cache_flow_control: 50,
139            pull_time_delay_millis_when_broker_flow_control: 20,
140            broker_suspend_max_time_millis: 20_000,
141            consumer_timeout_millis_when_suspend: 30_000,
142            consumer_pull_timeout_millis: 10_000,
143            poll_timeout_millis: 5000,
144            auto_commit: true,
145            auto_commit_interval_millis: 5000,
146            topic_metadata_check_interval_millis: 30000,
147            message_request_mode: MessageRequestMode::Pull,
148            rpc_hook: None,
149            trace_dispatcher: None,
150            enable_msg_trace: false,
151            custom_trace_topic: None,
152        }
153    }
154
155    /// Sets the name server address (required).
156    ///
157    /// # Examples
158    ///
159    /// ```rust,ignore
160    /// builder.name_server_addr("127.0.0.1:9876");
161    /// ```
162    pub fn name_server_addr(mut self, addr: impl Into<CheetahString>) -> Self {
163        self.name_server_addr = Some(addr.into());
164        self
165    }
166
167    /// Sets the complete client options, including typed NameServer discovery.
168    pub fn client_options(mut self, options: ClientOptions) -> Self {
169        self.client_options = Some(options);
170        self
171    }
172
173    /// Sets typed NameServer discovery on the current client options.
174    pub fn nameserver_discovery(mut self, discovery: NameServerDiscoveryConfig) -> Self {
175        let options = self.client_options.take().unwrap_or_default();
176        self.client_options = Some(options.with_nameserver_discovery(discovery));
177        self
178    }
179
180    /// Sets the consumer group name (required).
181    ///
182    /// # Examples
183    ///
184    /// ```rust,ignore
185    /// builder.consumer_group("my_consumer_group");
186    /// ```
187    pub fn consumer_group(mut self, group: impl Into<CheetahString>) -> Self {
188        self.consumer_group = Some(group.into());
189        self
190    }
191
192    /// Sets the client IP address (optional, auto-detected by default).
193    pub fn client_ip(mut self, ip: impl Into<CheetahString>) -> Self {
194        self.client_ip = Some(ip.into());
195        self
196    }
197
198    /// Sets the instance name (optional, auto-generated by default).
199    pub fn instance_name(mut self, name: impl Into<CheetahString>) -> Self {
200        self.instance_name = Some(name.into());
201        self
202    }
203
204    /// Sets the namespace (optional).
205    pub fn namespace(mut self, namespace: impl Into<CheetahString>) -> Self {
206        self.namespace = Some(namespace.into());
207        self
208    }
209
210    /// Sets whether client remoting connections should use TLS.
211    pub fn use_tls(mut self, use_tls: bool) -> Self {
212        self.use_tls = use_tls;
213        self
214    }
215
216    /// Sets the message model (default: Clustering).
217    pub fn message_model(mut self, model: MessageModel) -> Self {
218        self.message_model = model;
219        self
220    }
221
222    /// Sets where to start consuming from when no offset exists (default: LastOffset).
223    pub fn consume_from_where(mut self, consume_from: ConsumeFromWhere) -> Self {
224        self.consume_from_where = consume_from;
225        self
226    }
227
228    /// Sets the timestamp to consume from (for CONSUME_FROM_TIMESTAMP mode).
229    pub fn consume_timestamp(mut self, timestamp: impl Into<CheetahString>) -> Self {
230        self.consume_timestamp = Some(timestamp.into());
231        self
232    }
233
234    /// Sets the message queue allocation strategy.
235    pub fn allocate_message_queue_strategy(
236        mut self,
237        strategy: Arc<dyn AllocateMessageQueueStrategy + Send + Sync>,
238    ) -> Self {
239        self.allocate_message_queue_strategy = strategy;
240        self
241    }
242
243    /// Sets whether the subscription group runs in unit mode (default: false).
244    pub fn unit_mode(mut self, unit_mode: bool) -> Self {
245        self.unit_mode = unit_mode;
246        self
247    }
248
249    /// Sets the number of messages to pull in a single request (default: 10, range: 1-1024).
250    pub fn pull_batch_size(mut self, size: i32) -> Self {
251        self.pull_batch_size = size;
252        self
253    }
254
255    /// Sets the number of concurrent pull threads (default: 20).
256    pub fn pull_thread_nums(mut self, nums: usize) -> Self {
257        self.pull_thread_nums = nums;
258        self
259    }
260
261    /// Sets whether pulls should always use the configured default broker ID.
262    pub fn connect_broker_by_user(mut self, connect: bool) -> Self {
263        self.connect_broker_by_user = connect;
264        self
265    }
266
267    /// Sets the broker ID used when user-controlled broker selection is enabled.
268    pub fn default_broker_id(mut self, broker_id: u64) -> Self {
269        self.default_broker_id = broker_id;
270        self
271    }
272
273    /// Sets the maximum number of messages cached per queue (default: 1000).
274    pub fn pull_threshold_for_queue(mut self, threshold: i64) -> Self {
275        self.pull_threshold_for_queue = threshold;
276        self
277    }
278
279    /// Sets the maximum size in MiB of messages cached per queue (default: 100).
280    pub fn pull_threshold_size_for_queue(mut self, threshold: i32) -> Self {
281        self.pull_threshold_size_for_queue = threshold;
282        self
283    }
284
285    /// Sets the maximum total number of cached messages across all queues (default: 10000).
286    pub fn pull_threshold_for_all(mut self, threshold: i64) -> Self {
287        self.pull_threshold_for_all = threshold;
288        self
289    }
290
291    /// Sets the maximum offset span allowed in a process queue (default: 2000).
292    pub fn consume_max_span(mut self, span: i64) -> Self {
293        self.consume_max_span = span;
294        self
295    }
296
297    /// Sets the delay when pull encounters an exception (default: 1000ms).
298    pub fn pull_time_delay_millis_when_exception(mut self, delay: u64) -> Self {
299        self.pull_time_delay_millis_when_exception = delay;
300        self
301    }
302
303    /// Sets the delay when cache flow control is triggered (default: 50ms).
304    pub fn pull_time_delay_millis_when_cache_flow_control(mut self, delay: u64) -> Self {
305        self.pull_time_delay_millis_when_cache_flow_control = delay;
306        self
307    }
308
309    /// Sets the delay when broker flow control is triggered (default: 20ms).
310    pub fn pull_time_delay_millis_when_broker_flow_control(mut self, delay: u64) -> Self {
311        self.pull_time_delay_millis_when_broker_flow_control = delay;
312        self
313    }
314
315    /// Sets the maximum time that the broker may suspend a long-poll pull request (default:
316    /// 20000ms).
317    pub fn broker_suspend_max_time_millis(mut self, timeout: u64) -> Self {
318        self.broker_suspend_max_time_millis = timeout;
319        self
320    }
321
322    /// Sets the consumer-side timeout for suspended long-poll pull requests (default: 30000ms).
323    pub fn consumer_timeout_millis_when_suspend(mut self, timeout: u64) -> Self {
324        self.consumer_timeout_millis_when_suspend = timeout;
325        self
326    }
327
328    /// Sets the lite pull RPC timeout in milliseconds (default: 10000ms).
329    pub fn consumer_pull_timeout_millis(mut self, timeout: u64) -> Self {
330        self.consumer_pull_timeout_millis = timeout;
331        self
332    }
333
334    /// Sets the default poll timeout in milliseconds (default: 5000).
335    pub fn poll_timeout_millis(mut self, timeout: u64) -> Self {
336        self.poll_timeout_millis = timeout;
337        self
338    }
339
340    /// Sets whether to automatically commit offsets (default: true).
341    pub fn auto_commit(mut self, enable: bool) -> Self {
342        self.auto_commit = enable;
343        self
344    }
345
346    /// Sets the interval between automatic offset commits (default: 5000ms, minimum: 1000ms).
347    pub fn auto_commit_interval_millis(mut self, interval: u64) -> Self {
348        if interval >= MIN_AUTOCOMMIT_INTERVAL_MILLIS {
349            self.auto_commit_interval_millis = interval;
350        }
351        self
352    }
353
354    /// Sets the interval for checking topic metadata changes (default: 30000ms).
355    pub fn topic_metadata_check_interval_millis(mut self, interval: u64) -> Self {
356        self.topic_metadata_check_interval_millis = interval;
357        self
358    }
359
360    /// Sets the message request mode (default: Pull).
361    pub fn message_request_mode(mut self, mode: MessageRequestMode) -> Self {
362        self.message_request_mode = mode;
363        self
364    }
365
366    /// Sets the RPC hook for request/response interception.
367    pub fn rpc_hook(mut self, hook: Arc<dyn RPCHook>) -> Self {
368        self.rpc_hook = Some(hook);
369        self
370    }
371
372    /// Enables message trace with the default trace topic.
373    pub fn enable_msg_trace(mut self) -> Self {
374        self.enable_msg_trace = true;
375        self
376    }
377
378    /// Enables message trace with a custom trace topic.
379    pub fn enable_msg_trace_with_topic(mut self, trace_topic: impl Into<CheetahString>) -> Self {
380        self.enable_msg_trace = true;
381        self.custom_trace_topic = Some(trace_topic.into());
382        self
383    }
384
385    /// Sets a custom trace dispatcher.
386    pub fn trace_dispatcher(mut self, dispatcher: Arc<dyn TraceDispatcher + Send + Sync>) -> Self {
387        self.trace_dispatcher = Some(dispatcher);
388        self
389    }
390
391    pub(crate) fn classic_pull_manual_mode(mut self) -> Self {
392        self.classic_pull_manual_mode = true;
393        self
394    }
395
396    /// Builds the [`DefaultLitePullConsumer`].
397    pub fn build(self) -> ClientResult<DefaultLitePullConsumer> {
398        let Some(consumer_group) = self.consumer_group else {
399            return Err(mq_client_err!("consumer_group is required"));
400        };
401        validate_lite_pull_consume_from_where(self.consume_from_where)?;
402
403        let options = self.client_options.unwrap_or_default();
404        let mut client_config = options.client_config().clone();
405        let nameserver_discovery = options.nameserver_discovery().cloned();
406        client_config.set_enable_stream_request_type(!self.classic_pull_manual_mode);
407        if let Some(name_server_addr) = self.name_server_addr {
408            client_config.set_namesrv_addr(name_server_addr);
409        }
410
411        if let Some(ip) = self.client_ip {
412            client_config.set_client_ip(ip);
413        }
414
415        if let Some(name) = self.instance_name {
416            client_config.set_instance_name(name);
417        }
418
419        if let Some(namespace) = self.namespace {
420            client_config.set_namespace(namespace);
421        }
422        client_config.set_use_tls(self.use_tls);
423
424        let consumer_config = LitePullConsumerConfig {
425            consumer_group,
426            message_model: self.message_model,
427            consume_from_where: self.consume_from_where,
428            consume_timestamp: self.consume_timestamp,
429            allocate_message_queue_strategy: self.allocate_message_queue_strategy,
430            unit_mode: self.unit_mode,
431            classic_pull_manual_mode: self.classic_pull_manual_mode,
432            pull_batch_size: self.pull_batch_size,
433            pull_thread_nums: self.pull_thread_nums,
434            connect_broker_by_user: self.connect_broker_by_user,
435            default_broker_id: self.default_broker_id,
436            pull_threshold_for_queue: self.pull_threshold_for_queue,
437            pull_threshold_size_for_queue: self.pull_threshold_size_for_queue,
438            pull_threshold_for_all: self.pull_threshold_for_all,
439            consume_max_span: self.consume_max_span,
440            pull_time_delay_millis_when_exception: self.pull_time_delay_millis_when_exception,
441            pull_time_delay_millis_when_cache_flow_control: self.pull_time_delay_millis_when_cache_flow_control,
442            pull_time_delay_millis_when_broker_flow_control: self.pull_time_delay_millis_when_broker_flow_control,
443            broker_suspend_max_time_millis: self.broker_suspend_max_time_millis,
444            consumer_timeout_millis_when_suspend: self.consumer_timeout_millis_when_suspend,
445            consumer_pull_timeout_millis: self.consumer_pull_timeout_millis,
446            poll_timeout_millis: self.poll_timeout_millis,
447            auto_commit: self.auto_commit,
448            auto_commit_interval_millis: self.auto_commit_interval_millis,
449            topic_metadata_check_interval_millis: self.topic_metadata_check_interval_millis,
450            message_request_mode: self.message_request_mode,
451        };
452
453        Ok(DefaultLitePullConsumer::new_with_options(
454            self.client_runtime,
455            ClientOptions::from_parts(client_config, nameserver_discovery),
456            consumer_config,
457            self.rpc_hook,
458            self.trace_dispatcher,
459            self.enable_msg_trace,
460            self.custom_trace_topic,
461        ))
462    }
463}
464
465#[cfg(test)]
466mod tests {
467    use super::*;
468    use crate::test_support::error_assertions::client_exception;
469
470    fn test_runtime() -> Arc<ClientRuntime> {
471        crate::runtime::test_client_runtime("lite-pull-consumer-builder-test")
472    }
473
474    #[test]
475    fn build_without_consumer_group_returns_error() {
476        let result = DefaultLitePullConsumerBuilder::new(test_runtime()).build();
477        match result {
478            Ok(_) => panic!("builder should reject missing consumer group"),
479            Err(error) => assert!(client_exception(&error)
480                .to_string()
481                .contains("consumer_group is required")),
482        }
483    }
484
485    #[test]
486    fn build_without_name_server_addr_uses_default_client_config() {
487        let consumer = DefaultLitePullConsumerBuilder::new(test_runtime())
488            .consumer_group("lite_pull_group")
489            .build()
490            .expect("builder should use default client config");
491
492        assert_eq!(consumer.consumer_group().as_str(), "lite_pull_group");
493    }
494
495    #[test]
496    fn build_enables_stream_request_type_like_java_lite_pull_constructor() {
497        let consumer = DefaultLitePullConsumerBuilder::new(test_runtime())
498            .consumer_group("lite_pull_group")
499            .build()
500            .expect("builder should use Java LitePull client id mode");
501
502        assert!(consumer.client_config().is_enable_stream_request_type());
503        assert!(consumer.client_config().build_mq_client_id().ends_with("@STREAM"));
504    }
505
506    #[test]
507    fn build_use_tls_initializes_client_config_like_java_client_config() {
508        let consumer = DefaultLitePullConsumerBuilder::new(test_runtime())
509            .consumer_group("lite_pull_tls_group")
510            .use_tls(true)
511            .build()
512            .expect("builder should apply TLS flag");
513
514        assert!(consumer.is_use_tls());
515        assert!(consumer.client_config().is_use_tls());
516    }
517
518    #[test]
519    #[allow(deprecated)]
520    fn build_rejects_unsupported_consume_from_where_like_java_lite_pull() {
521        let result = DefaultLitePullConsumerBuilder::new(test_runtime())
522            .consumer_group("lite_pull_group")
523            .consume_from_where(ConsumeFromWhere::ConsumeFromMinOffset)
524            .build();
525
526        match result {
527            Ok(_) => panic!("Java LitePull rejects legacy consumeFromWhere values"),
528            Err(error) => assert!(client_exception(&error)
529                .to_string()
530                .contains("Invalid ConsumeFromWhere Value")),
531        }
532    }
533
534    #[test]
535    fn default_pull_threshold_for_all_matches_java() {
536        let consumer = DefaultLitePullConsumerBuilder::new(test_runtime())
537            .consumer_group("lite_pull_group")
538            .build()
539            .expect("builder should use Java default pull threshold");
540
541        assert_eq!(consumer.consumer_config().pull_threshold_for_all, 10000);
542        assert_eq!(LitePullConsumerConfig::default().pull_threshold_for_all, 10000);
543    }
544
545    #[test]
546    fn pull_thread_nums_defaults_and_overrides_match_java_lite_pull() {
547        let default_consumer = DefaultLitePullConsumerBuilder::new(test_runtime())
548            .consumer_group("lite_pull_group")
549            .build()
550            .expect("builder should use Java default pull thread count");
551
552        assert_eq!(default_consumer.consumer_config().pull_thread_nums, 20);
553        assert_eq!(LitePullConsumerConfig::default().pull_thread_nums, 20);
554
555        let custom_consumer = DefaultLitePullConsumerBuilder::new(test_runtime())
556            .consumer_group("lite_pull_group")
557            .pull_thread_nums(7)
558            .build()
559            .expect("builder should preserve configured pull thread count");
560
561        assert_eq!(custom_consumer.consumer_config().pull_thread_nums, 7);
562    }
563
564    #[tokio::test]
565    async fn broker_selection_defaults_and_overrides_match_java_lite_pull() {
566        let default_consumer = DefaultLitePullConsumerBuilder::new(test_runtime())
567            .consumer_group("lite_pull_group")
568            .build()
569            .expect("builder should use Java default broker selection");
570
571        assert!(!default_consumer.is_connect_broker_by_user().await);
572        assert_eq!(default_consumer.default_broker_id().await, mix_all::MASTER_ID);
573        assert!(!LitePullConsumerConfig::default().connect_broker_by_user);
574        assert_eq!(LitePullConsumerConfig::default().default_broker_id, mix_all::MASTER_ID);
575
576        let custom_consumer = DefaultLitePullConsumerBuilder::new(test_runtime())
577            .consumer_group("lite_pull_group")
578            .connect_broker_by_user(true)
579            .default_broker_id(2)
580            .build()
581            .expect("builder should preserve configured broker selection");
582
583        assert!(custom_consumer.is_connect_broker_by_user().await);
584        assert_eq!(custom_consumer.default_broker_id().await, 2);
585    }
586
587    #[tokio::test]
588    async fn unit_mode_defaults_and_overrides_match_java_lite_pull() {
589        let default_consumer = DefaultLitePullConsumerBuilder::new(test_runtime())
590            .consumer_group("lite_pull_group")
591            .build()
592            .expect("builder should use Java default unit mode");
593
594        assert!(!default_consumer.is_unit_mode().await);
595        assert!(!default_consumer.consumer_config().unit_mode);
596        assert!(!LitePullConsumerConfig::default().unit_mode);
597
598        let custom_consumer = DefaultLitePullConsumerBuilder::new(test_runtime())
599            .consumer_group("lite_pull_group")
600            .unit_mode(true)
601            .build()
602            .expect("builder should preserve configured unit mode");
603
604        assert!(custom_consumer.is_unit_mode().await);
605        assert!(custom_consumer.consumer_config().unit_mode);
606    }
607
608    #[test]
609    fn long_polling_timeout_defaults_and_overrides_match_java_lite_pull() {
610        let default_consumer = DefaultLitePullConsumerBuilder::new(test_runtime())
611            .consumer_group("lite_pull_group")
612            .build()
613            .expect("builder should use Java default long polling timeouts");
614
615        assert_eq!(
616            default_consumer.consumer_config().broker_suspend_max_time_millis,
617            20_000
618        );
619        assert_eq!(
620            default_consumer.consumer_config().consumer_timeout_millis_when_suspend,
621            30_000
622        );
623        assert_eq!(default_consumer.consumer_config().consumer_pull_timeout_millis, 10_000);
624        assert_eq!(LitePullConsumerConfig::default().broker_suspend_max_time_millis, 20_000);
625        assert_eq!(
626            LitePullConsumerConfig::default().consumer_timeout_millis_when_suspend,
627            30_000
628        );
629        assert_eq!(LitePullConsumerConfig::default().consumer_pull_timeout_millis, 10_000);
630
631        let custom_consumer = DefaultLitePullConsumerBuilder::new(test_runtime())
632            .consumer_group("lite_pull_group")
633            .broker_suspend_max_time_millis(12_345)
634            .consumer_timeout_millis_when_suspend(23_456)
635            .consumer_pull_timeout_millis(34_567)
636            .build()
637            .expect("builder should keep configured long polling timeouts");
638
639        assert_eq!(custom_consumer.consumer_config().broker_suspend_max_time_millis, 12_345);
640        assert_eq!(
641            custom_consumer.consumer_config().consumer_timeout_millis_when_suspend,
642            23_456
643        );
644        assert_eq!(custom_consumer.consumer_config().consumer_pull_timeout_millis, 34_567);
645    }
646
647    #[test]
648    fn default_consume_timestamp_matches_java_shape() {
649        let consumer = DefaultLitePullConsumerBuilder::new(test_runtime())
650            .consumer_group("lite_pull_group")
651            .build()
652            .expect("builder should use Java default consume timestamp");
653        let consumer_config = consumer.consumer_config();
654        let timestamp = consumer_config
655            .consume_timestamp
656            .as_ref()
657            .expect("Java LitePull default consume timestamp should be present");
658
659        assert!(rocketmq_runtime::common::time_utils::parse_date(
660            timestamp,
661            rocketmq_runtime::common::time_utils::YYYYMMDDHHMMSS
662        )
663        .is_some());
664        let config = LitePullConsumerConfig::default();
665        let config_timestamp = config
666            .consume_timestamp
667            .as_ref()
668            .expect("Java LitePull config default consume timestamp should be present");
669        assert!(rocketmq_runtime::common::time_utils::parse_date(
670            config_timestamp,
671            rocketmq_runtime::common::time_utils::YYYYMMDDHHMMSS
672        )
673        .is_some());
674    }
675
676    #[test]
677    fn auto_commit_interval_below_java_minimum_keeps_default() {
678        let consumer = DefaultLitePullConsumerBuilder::new(test_runtime())
679            .consumer_group("lite_pull_group")
680            .auto_commit_interval_millis(MIN_AUTOCOMMIT_INTERVAL_MILLIS - 1)
681            .build()
682            .expect("builder should ignore invalid auto commit interval");
683
684        assert_eq!(consumer.consumer_config().auto_commit_interval_millis, 5000);
685    }
686
687    #[test]
688    fn auto_commit_interval_below_java_minimum_keeps_previous_valid_value() {
689        let consumer = DefaultLitePullConsumerBuilder::new(test_runtime())
690            .consumer_group("lite_pull_group")
691            .auto_commit_interval_millis(2000)
692            .auto_commit_interval_millis(MIN_AUTOCOMMIT_INTERVAL_MILLIS - 1)
693            .build()
694            .expect("builder should keep previous valid auto commit interval");
695
696        assert_eq!(consumer.consumer_config().auto_commit_interval_millis, 2000);
697    }
698}
rocketmq_client_rust/consumer/default_lite_pull_consumer_builder.rs

rocketmq_client_rust/consumer/
default_lite_pull_consumer_builder.rs
