rocketmq_client_rust/consumer/
pop_callback.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::future::Future;
16use std::pin::Pin;
17use std::sync::Arc;
18
19use rocketmq_common::common::message::message_queue::MessageQueue;
20use rocketmq_common::common::mix_all;
21use rocketmq_error::RocketmqError;
22use rocketmq_remoting::code::response_code::ResponseCode;
23use rocketmq_remoting::protocol::heartbeat::subscription_data::SubscriptionData;
24use rocketmq_rust::ArcMut;
25use tracing::warn;
26
27use crate::consumer::consumer_impl::default_mq_push_consumer_impl::DefaultMQPushConsumerImpl;
28use crate::consumer::consumer_impl::default_mq_push_consumer_impl::PULL_TIME_DELAY_MILLS_WHEN_BROKER_FLOW_CONTROL;
29use crate::consumer::consumer_impl::pop_request::PopRequest;
30use crate::consumer::pop_result::PopResult;
31use crate::consumer::pop_status::PopStatus;
32
33/// Trait for handling the results of a pop operation.
34///
35/// This trait defines the methods for handling successful and error results of a pop operation.
36#[trait_variant::make(PopCallback: Send)]
37pub trait PopCallbackInner {
38    /// Called when the pop operation is successful.
39    ///
40    /// # Arguments
41    ///
42    /// * `pop_result` - The result of the pop operation.
43    async fn on_success(&mut self, pop_result: PopResult);
44
45    /// Called when the pop operation encounters an error.
46    ///
47    /// # Arguments
48    ///
49    /// * `e` - The error encountered during the pop operation.
50    fn on_error(&mut self, e: Box<dyn std::error::Error + Send>);
51}
52
53/*impl<F, Fut> PopCallback for F
54where
55    F: Fn(Option<PopResult>, Option<Box<dyn std::error::Error + Send>>) -> Fut + Send + Sync,
56    Fut: Future<Output = ()> + Send,
57{
58    /// Calls the function with the pop result when the pop operation is successful.
59    ///
60    /// # Arguments
61    ///
62    /// * `pop_result` - The result of the pop operation.
63    async fn on_success(&mut self, pop_result: PopResult) {
64        (*self)(Some(pop_result), None).await;
65    }
66
67    /// Does nothing when the pop operation encounters an error.
68    ///
69    /// # Arguments
70    ///
71    /// * `e` - The error encountered during the pop operation.
72    fn on_error(&self, e: Box<dyn std::error::Error + Send>) {
73        (*self)(None, Some(e));
74    }
75}*/
76
77/// Type alias for a callback function that handles the result of a pop operation.
78///
79/// This type alias defines a callback function that takes a `PopResult` and returns a boxed future.
80pub type PopCallbackFn = Arc<
81    dyn Fn(Option<PopResult>, Option<Box<dyn std::error::Error>>) -> Pin<Box<dyn Future<Output = ()> + Send + Sync>>
82        + Send
83        + Sync,
84>;
85
86pub struct DefaultPopCallback {
87    pub(crate) push_consumer_impl: ArcMut<DefaultMQPushConsumerImpl>,
88    pub(crate) message_queue_inner: Option<MessageQueue>,
89    pub(crate) subscription_data: Option<SubscriptionData>,
90    pub(crate) pop_request: Option<PopRequest>,
91}
92
93impl PopCallback for DefaultPopCallback {
94    async fn on_success(&mut self, pop_result: PopResult) {
95        let mut push_consumer_impl = self.push_consumer_impl.clone();
96
97        let Some(message_queue_inner) = self.message_queue_inner.take() else {
98            warn!("pop callback success ignored: message queue is missing");
99            return;
100        };
101        let Some(subscription_data) = self.subscription_data.take() else {
102            warn!(
103                "pop callback success ignored: subscription data is missing, mq={}",
104                message_queue_inner
105            );
106            if let Some(pop_request) = self.pop_request.take() {
107                let delay = push_consumer_impl.pull_time_delay_mills_when_exception;
108                push_consumer_impl.execute_pop_request_later(pop_request, delay);
109            }
110            return;
111        };
112        let Some(pop_request) = self.pop_request.take() else {
113            warn!(
114                "pop callback success ignored: pop request is missing, mq={}",
115                message_queue_inner
116            );
117            return;
118        };
119
120        let pop_result = push_consumer_impl
121            .process_pop_result(pop_result, &subscription_data)
122            .await;
123        match pop_result.pop_status {
124            PopStatus::Found => {
125                if pop_result.msg_found_list.as_ref().is_none_or(|value| value.is_empty()) {
126                    push_consumer_impl.execute_pop_request_immediately(pop_request).await;
127                } else {
128                    let Some(consume_message_pop_service) = push_consumer_impl.consume_message_pop_service.as_mut()
129                    else {
130                        warn!(
131                            "pop callback found messages but ConsumeMessagePopService is not initialized, mq={}",
132                            message_queue_inner
133                        );
134                        let delay = push_consumer_impl.pull_time_delay_mills_when_exception;
135                        push_consumer_impl.execute_pop_request_later(pop_request, delay);
136                        return;
137                    };
138                    consume_message_pop_service
139                        .submit_pop_consume_request(
140                            pop_result.msg_found_list.unwrap_or_default(),
141                            pop_request.get_pop_process_queue(),
142                            pop_request.get_message_queue(),
143                        )
144                        .await;
145                    let pull_interval = push_consumer_impl.consumer_config.pull_interval;
146                    if pull_interval > 0 {
147                        push_consumer_impl.execute_pop_request_later(pop_request, pull_interval);
148                    } else {
149                        push_consumer_impl.execute_pop_request_immediately(pop_request).await;
150                    }
151                }
152            }
153            PopStatus::NoNewMsg | PopStatus::PollingNotFound => {
154                push_consumer_impl.execute_pop_request_immediately(pop_request).await;
155            }
156            PopStatus::PollingFull => {
157                let pull_time_delay_mills_when_exception = push_consumer_impl.pull_time_delay_mills_when_exception;
158                push_consumer_impl.execute_pop_request_later(pop_request, pull_time_delay_mills_when_exception);
159            }
160        }
161    }
162
163    fn on_error(&mut self, err: Box<dyn std::error::Error + Send>) {
164        let mut push_consumer_impl = self.push_consumer_impl.clone();
165
166        let Some(message_queue_inner) = self.message_queue_inner.take() else {
167            warn!("pop callback exception ignored: message queue is missing, error={err}");
168            return;
169        };
170        let Some(pop_request) = self.pop_request.take() else {
171            warn!(
172                "pop callback exception ignored: pop request is missing, mq={}, error={}",
173                message_queue_inner, err
174            );
175            return;
176        };
177        let topic = message_queue_inner.topic_str();
178        if !topic.starts_with(mix_all::RETRY_GROUP_TOPIC_PREFIX) {
179            if let Some(er) = err.downcast_ref::<RocketmqError>() {
180                match er {
181                    RocketmqError::MQClientBrokerError(broker_error) => {
182                        if ResponseCode::from(broker_error.response_code()) == ResponseCode::SubscriptionNotLatest {
183                            warn!(
184                                "the subscription is not latest, group={}",
185                                push_consumer_impl.consumer_config.consumer_group,
186                            );
187                        } else {
188                            warn!(
189                                "execute the pop request exception, group={}",
190                                push_consumer_impl.consumer_config.consumer_group
191                            );
192                        }
193                    }
194                    _ => {
195                        warn!(
196                            "execute the pop request exception, group={}",
197                            push_consumer_impl.consumer_config.consumer_group
198                        );
199                    }
200                }
201            } else {
202                warn!(
203                    "execute the pull request exception, group={}",
204                    push_consumer_impl.consumer_config.consumer_group
205                );
206            }
207        }
208        let time_delay = if let Some(er) = err.downcast_ref::<RocketmqError>() {
209            match er {
210                RocketmqError::MQClientBrokerError(broker_error) => {
211                    if ResponseCode::from(broker_error.response_code()) == ResponseCode::FlowControl {
212                        PULL_TIME_DELAY_MILLS_WHEN_BROKER_FLOW_CONTROL
213                    } else {
214                        push_consumer_impl.pull_time_delay_mills_when_exception
215                    }
216                }
217                _ => push_consumer_impl.pull_time_delay_mills_when_exception,
218            }
219        } else {
220            push_consumer_impl.pull_time_delay_mills_when_exception
221        };
222
223        push_consumer_impl.execute_pop_request_later(pop_request, time_delay);
224    }
225}
226
227#[cfg(test)]
228mod tests {
229    use std::io;
230
231    use cheetah_string::CheetahString;
232    use rocketmq_common::common::message::message_ext::MessageExt;
233
234    use super::*;
235    use crate::base::client_config::ClientConfig;
236    use crate::consumer::consumer_impl::pop_process_queue::PopProcessQueue;
237    use crate::consumer::default_mq_push_consumer::ConsumerConfig;
238
239    fn message_queue() -> MessageQueue {
240        MessageQueue::from_parts("topic", "broker-a", 0)
241    }
242
243    fn pop_request() -> PopRequest {
244        PopRequest::new(
245            CheetahString::from_static_str("topic"),
246            CheetahString::from_static_str("group"),
247            message_queue(),
248            PopProcessQueue::new(),
249            0,
250        )
251    }
252
253    fn new_callback() -> DefaultPopCallback {
254        let consumer_config = ArcMut::new(ConsumerConfig::default());
255        let push_consumer_impl = ArcMut::new(DefaultMQPushConsumerImpl::new(
256            ClientConfig::default(),
257            consumer_config,
258            None,
259        ));
260        DefaultPopCallback {
261            push_consumer_impl,
262            message_queue_inner: None,
263            subscription_data: None,
264            pop_request: None,
265        }
266    }
267
268    #[tokio::test]
269    async fn success_without_message_queue_is_ignored_without_panic() {
270        let mut callback = new_callback();
271
272        PopCallback::on_success(&mut callback, PopResult::default()).await;
273    }
274
275    #[tokio::test]
276    async fn success_without_subscription_data_is_delayed_without_panic() {
277        let mut callback = new_callback();
278        callback.message_queue_inner = Some(message_queue());
279        callback.pop_request = Some(pop_request());
280
281        PopCallback::on_success(&mut callback, PopResult::default()).await;
282    }
283
284    #[tokio::test]
285    async fn success_without_pop_request_is_ignored_without_panic() {
286        let mut callback = new_callback();
287        callback.message_queue_inner = Some(message_queue());
288        callback.subscription_data = Some(SubscriptionData::default());
289
290        PopCallback::on_success(&mut callback, PopResult::default()).await;
291    }
292
293    #[tokio::test]
294    async fn found_messages_without_pop_service_are_delayed_without_panic() {
295        let mut callback = new_callback();
296        callback.message_queue_inner = Some(message_queue());
297        callback.subscription_data = Some(SubscriptionData::default());
298        callback.pop_request = Some(pop_request());
299        let pop_result = PopResult {
300            msg_found_list: Some(vec![MessageExt::default()]),
301            pop_status: PopStatus::Found,
302            ..Default::default()
303        };
304
305        PopCallback::on_success(&mut callback, pop_result).await;
306    }
307
308    #[test]
309    fn error_without_message_queue_is_ignored_without_panic() {
310        let mut callback = new_callback();
311
312        PopCallback::on_error(&mut callback, Box::new(io::Error::other("test error")));
313    }
314
315    #[test]
316    fn error_without_pop_request_is_ignored_without_panic() {
317        let mut callback = new_callback();
318        callback.message_queue_inner = Some(message_queue());
319
320        PopCallback::on_error(&mut callback, Box::new(io::Error::other("test error")));
321    }
322}
rocketmq_client_rust/consumer/pop_callback.rs

rocketmq_client_rust/consumer/
pop_callback.rs
