27fffe3
use bytes::Bytes;
use rocketmq_common::common::message::message_single::Message;
use crate::conn::RocketMqConn;
use crate::error::{QueueError, QueueResult};
#[derive(Clone, Debug)]
pub struct Publisher {
conn: RocketMqConn,
}
impl Publisher {
pub fn new(conn: RocketMqConn) -> Self {
Self { conn }
}
pub fn conn(&self) -> &RocketMqConn {
&self.conn
}
pub fn max_payload(&self) -> usize {
self.conn.max_payload()
}
pub fn topic(&self, area: &str) -> String {
self.conn.config().topic(area)
}
pub async fn publish(
&self,
topic: &str,
tag: &str,
keys: Vec<String>,
properties: Vec<(String, String)>,
body: Bytes,
) -> QueueResult<()> {
let max_payload = self.max_payload();
if body.len() > max_payload {
return Err(QueueError::payload(format!(
"event body for `{topic}` is {} bytes, exceeding the {max_payload}-byte limit",
body.len()
)));
}
let mut builder = Message::builder()
.topic(topic)
.tags(tag)
.keys(keys)
.body(body.to_vec());
for (key, value) in properties {
builder = builder.raw_property(key.clone(), value).map_err(|error| {
QueueError::publish(format!(
"invalid message property `{key}` for `{topic}`: {error}"
))
})?;
}
let message = builder.build().map_err(|error| {
QueueError::publish(format!(
"failed to build the message for `{topic}`: {error}"
))
})?;
self.conn.send(topic, message).await
}
}