db73d56
ZhenYi
refactor: add events dependency and enhance event handling for improved clarity and organization
db73d56
use std::sync::Arc;
use rocketmq_client_rust::consumer::default_mq_push_consumer::DefaultMQPushConsumer;
use rocketmq_client_rust::consumer::listener::{
ConsumeConcurrentlyContext, ConsumeConcurrentlyStatus,
};
use rocketmq_client_rust::consumer::mq_push_consumer::MQPushConsumer;
use rocketmq_common::common::message::message_ext::MessageExt;
use crate::error::{QueueError, QueueResult};
pub type MessageHandler = Arc<dyn Fn(&[&MessageExt]) -> bool + Send + Sync>;
#[derive(Clone, Debug)]
pub struct ConsumerConfig {
pub name_server: String,
pub consumer_group: String,
pub topic: String,
pub tag: String,
pub workers: u32,
pub max_reconsume: i32,
}
pub struct Consumer {
inner: DefaultMQPushConsumer,
}
impl Consumer {
pub async fn start(config: ConsumerConfig, handler: MessageHandler) -> QueueResult<Self> {
if config.name_server.trim().is_empty() {
return Err(QueueError::config("rocketmq.name_server must not be blank"));
}
if config.consumer_group.trim().is_empty() {
return Err(QueueError::config("consumer group must not be blank"));
}
if config.topic.trim().is_empty() {
return Err(QueueError::config("consumer topic must not be blank"));
}
let mut consumer = DefaultMQPushConsumer::builder()
.consumer_group(config.consumer_group)
.name_server_addr(config.name_server.replace(',', ";"))
.consume_thread_min(config.workers.max(1))
.consume_thread_max(config.workers.max(1))
.consume_message_batch_max_size(1)
.max_reconsume_times(config.max_reconsume.max(1))
.build();
consumer
.subscribe(config.topic, config.tag)
.await
.map_err(|error| {
QueueError::connect(format!("subscribing to rocketmq failed: {error}"))
})?;
let listener = move |messages: &[&MessageExt], _context: &ConsumeConcurrentlyContext| {
if handler(messages) {
Ok(ConsumeConcurrentlyStatus::ConsumeSuccess)
} else {
Ok(ConsumeConcurrentlyStatus::ReconsumeLater)
}
};
consumer.register_message_listener_concurrently(listener);
consumer.start().await.map_err(|error| {
QueueError::connect(format!("starting the rocketmq consumer failed: {error}"))
})?;
Ok(Self { inner: consumer })
}
pub async fn shutdown(&mut self) {
self.inner.shutdown().await;
}
}