27fffe3
use std::sync::Arc;
use config::rocketmq::RocketMqConfig;
use rocketmq_client_rust::producer::default_mq_producer::DefaultMQProducer;
use rocketmq_common::common::message::message_single::Message;
use tokio::sync::Mutex;
use crate::error::{QueueError, QueueResult};
#[derive(Clone)]
pub struct RocketMqConn {
producer: Arc<Mutex<DefaultMQProducer>>,
config: RocketMqConfig,
}
impl std::fmt::Debug for RocketMqConn {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("RocketMqConn")
.field("name_server", &self.config.name_server)
.field("producer_group", &self.config.producer_group)
.field("topic_prefix", &self.config.topic_prefix)
.finish()
}
}
impl RocketMqConn {
pub async fn init(config: &RocketMqConfig) -> QueueResult<Self> {
config.is_valid().map_err(QueueError::config)?;
let mut producer = DefaultMQProducer::builder()
.producer_group(config.producer_group.clone())
.name_server_addr(config.name_server.replace(',', ";"))
.build();
producer.start().await.map_err(|error| {
QueueError::connect(format!("failed to start the rocketmq producer: {error}"))
})?;
Ok(Self {
producer: Arc::new(Mutex::new(producer)),
config: config.clone(),
})
}
pub fn config(&self) -> &RocketMqConfig {
&self.config
}
pub fn max_payload(&self) -> usize {
self.config.max_message_bytes
}
pub async fn send(&self, topic: &str, message: Message) -> QueueResult<()> {
let mut producer = self.producer.lock().await;
let result = producer
.send_with_timeout(message, self.config.send_timeout_ms)
.await
.map_err(|error| {
QueueError::publish(format!("failed to publish to `{topic}`: {error}"))
})?;
if result.is_none() {
return Err(QueueError::publish(format!(
"producer returned no send result for `{topic}`"
)));
}
Ok(())
}
pub async fn shutdown(&self) {
let mut producer = self.producer.lock().await;
producer.shutdown().await;
}
}