rocketmq_client_rust/consumer/
default_mq_pull_consumer_builder.rs

1// Copyright 2026 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(deprecated)]
16
17use std::sync::Arc;
18use std::time::Duration;
19
20use crate::ClientResult;
21use cheetah_string::CheetahString;
22use rocketmq_protocol::protocol::heartbeat::message_model::MessageModel;
23use rocketmq_transport::api::RPCHook;
24
25use crate::base::client_options::ClientOptions;
26use crate::consumer::default_lite_pull_consumer::DefaultLitePullConsumer;
27use crate::consumer::default_mq_pull_consumer::DefaultMQPullConsumer;
28use crate::nameserver_discovery::NameServerDiscoveryConfig;
29use crate::runtime::ClientRuntime;
30
31/// Builds a Classic Pull compatibility facade on the shared client runtime.
32pub struct DefaultMQPullConsumerBuilder {
33    client_runtime: Arc<ClientRuntime>,
34    client_options: Option<ClientOptions>,
35    name_server_addr: Option<CheetahString>,
36    consumer_group: Option<CheetahString>,
37    message_model: MessageModel,
38    unit_mode: bool,
39    broker_suspend_max_time: Duration,
40    consumer_timeout_when_suspend: Duration,
41    consumer_pull_timeout: Duration,
42    queue_refresh_interval: Duration,
43    rpc_hook: Option<Arc<dyn RPCHook>>,
44}
45
46impl DefaultMQPullConsumerBuilder {
47    /// Creates a builder backed by an application-owned client runtime.
48    pub fn new(client_runtime: Arc<ClientRuntime>) -> Self {
49        Self {
50            client_runtime,
51            client_options: None,
52            name_server_addr: None,
53            consumer_group: None,
54            message_model: MessageModel::Clustering,
55            unit_mode: false,
56            broker_suspend_max_time: Duration::from_secs(20),
57            consumer_timeout_when_suspend: Duration::from_secs(30),
58            consumer_pull_timeout: Duration::from_secs(10),
59            queue_refresh_interval: Duration::from_secs(30),
60            rpc_hook: None,
61        }
62    }
63
64    /// Sets the consumer group.
65    pub fn consumer_group(mut self, consumer_group: impl Into<CheetahString>) -> Self {
66        self.consumer_group = Some(consumer_group.into());
67        self
68    }
69
70    /// Sets the complete client options, including typed NameServer discovery.
71    pub fn client_options(mut self, options: ClientOptions) -> Self {
72        self.client_options = Some(options);
73        self
74    }
75
76    /// Sets typed NameServer discovery.
77    pub fn nameserver_discovery(mut self, discovery: NameServerDiscoveryConfig) -> Self {
78        let options = self.client_options.take().unwrap_or_default();
79        self.client_options = Some(options.with_nameserver_discovery(discovery));
80        self
81    }
82
83    /// Sets a static NameServer address list.
84    pub fn name_server_addr(mut self, name_server_addr: impl Into<CheetahString>) -> Self {
85        self.name_server_addr = Some(name_server_addr.into());
86        self
87    }
88
89    /// Sets the consumer message model.
90    pub fn message_model(mut self, message_model: MessageModel) -> Self {
91        self.message_model = message_model;
92        self
93    }
94
95    /// Sets unit mode.
96    pub fn unit_mode(mut self, unit_mode: bool) -> Self {
97        self.unit_mode = unit_mode;
98        self
99    }
100
101    /// Sets the broker-side long-poll suspension limit.
102    pub fn broker_suspend_max_time(mut self, timeout: Duration) -> Self {
103        self.broker_suspend_max_time = timeout;
104        self
105    }
106
107    /// Sets the client timeout used by block-if-not-found pulls.
108    pub fn consumer_timeout_when_suspend(mut self, timeout: Duration) -> Self {
109        self.consumer_timeout_when_suspend = timeout;
110        self
111    }
112
113    /// Sets the default timeout used by ordinary pulls.
114    pub fn consumer_pull_timeout(mut self, timeout: Duration) -> Self {
115        self.consumer_pull_timeout = timeout;
116        self
117    }
118
119    /// Sets how frequently registered queue listeners refresh topic routes.
120    pub fn queue_refresh_interval(mut self, interval: Duration) -> Self {
121        self.queue_refresh_interval = interval;
122        self
123    }
124
125    /// Sets the request hook used by the shared client instance.
126    pub fn rpc_hook(mut self, rpc_hook: Arc<dyn RPCHook>) -> Self {
127        self.rpc_hook = Some(rpc_hook);
128        self
129    }
130
131    /// Builds the Classic Pull facade.
132    ///
133    /// # Errors
134    ///
135    /// Returns an error when the consumer group is missing, a duration is zero or outside the
136    /// supported millisecond range, the suspended-request timeout is not greater than the broker
137    /// suspension limit, or the underlying consumer configuration is invalid.
138    pub fn build(self) -> ClientResult<DefaultMQPullConsumer> {
139        let consumer_group = self
140            .consumer_group
141            .ok_or_else(|| crate::mq_client_err!("consumer_group is required"))?;
142        if self.broker_suspend_max_time.is_zero() {
143            return Err(crate::mq_client_err!("broker suspend timeout must be positive"));
144        }
145        if self.consumer_timeout_when_suspend <= self.broker_suspend_max_time {
146            return Err(crate::mq_client_err!(
147                "consumer timeout when suspended must exceed broker suspend timeout"
148            ));
149        }
150        if self.consumer_pull_timeout.is_zero() {
151            return Err(crate::mq_client_err!("consumer pull timeout must be positive"));
152        }
153        if self.queue_refresh_interval.is_zero() {
154            return Err(crate::mq_client_err!("queue refresh interval must be positive"));
155        }
156
157        let broker_suspend_max_time_millis = duration_millis("broker suspend timeout", self.broker_suspend_max_time)?;
158        let consumer_timeout_when_suspend_millis =
159            duration_millis("consumer timeout when suspended", self.consumer_timeout_when_suspend)?;
160        let consumer_pull_timeout_millis = duration_millis("consumer pull timeout", self.consumer_pull_timeout)?;
161        let queue_refresh_interval_millis = duration_millis("queue refresh interval", self.queue_refresh_interval)?;
162
163        let mut lite_builder = DefaultLitePullConsumer::builder(self.client_runtime.clone())
164            .consumer_group(consumer_group.clone())
165            .message_model(self.message_model)
166            .unit_mode(self.unit_mode)
167            .classic_pull_manual_mode()
168            .auto_commit(false)
169            .broker_suspend_max_time_millis(broker_suspend_max_time_millis)
170            .consumer_timeout_millis_when_suspend(consumer_timeout_when_suspend_millis)
171            .consumer_pull_timeout_millis(consumer_pull_timeout_millis)
172            .topic_metadata_check_interval_millis(queue_refresh_interval_millis);
173        if let Some(options) = self.client_options {
174            lite_builder = lite_builder.client_options(options);
175        }
176        if let Some(name_server_addr) = self.name_server_addr {
177            lite_builder = lite_builder.name_server_addr(name_server_addr);
178        }
179        if let Some(rpc_hook) = self.rpc_hook {
180            lite_builder = lite_builder.rpc_hook(rpc_hook);
181        }
182
183        DefaultMQPullConsumer::from_lite_consumer(
184            consumer_group,
185            lite_builder.build()?,
186            self.consumer_pull_timeout,
187            self.broker_suspend_max_time,
188            self.consumer_timeout_when_suspend,
189        )
190    }
191}
192
193fn duration_millis(_name: &str, duration: Duration) -> ClientResult<u64> {
194    u64::try_from(duration.as_millis()).map_err(crate::ClientError::illegal_argument_source)
195}
rocketmq_client_rust/consumer/default_mq_pull_consumer_builder.rs

rocketmq_client_rust/consumer/
default_mq_pull_consumer_builder.rs
