db73d56
ZhenYi
refactor: add events dependency and enhance event handling for improved clarity and organization
db73d56
use serde::Serialize;
use crate::{label, topic};
use config::rocketmq::RocketMqConfig;
use queue::{EVENT_PREFIX, Envelope, Labels, Publisher, QueueError, QueueResult, RocketMqConn};
pub const DURABLE_DOMAINS: &[&str] = &[
"auth",
"email",
"guild",
"chat",
"git",
"issue",
"notify",
"organization",
"pull",
"pull_request",
];
#[derive(Clone, Debug)]
pub struct EventHandler {
publisher: Publisher,
}
impl EventHandler {
pub fn new(mq: RocketMqConn) -> Self {
Self {
publisher: Publisher::new(mq),
}
}
pub async fn connect(config: &RocketMqConfig) -> QueueResult<Self> {
let conn = RocketMqConn::init(config).await?;
Ok(Self::new(conn))
}
pub async fn optional(config: Option<&RocketMqConfig>) -> QueueResult<Option<Self>> {
match config {
Some(config) => Self::connect(config).await.map(Some),
None => Ok(None),
}
}
pub fn publisher(&self) -> &Publisher {
&self.publisher
}
pub fn max_payload(&self) -> usize {
self.publisher.max_payload()
}
pub fn envelope<T>(&self, kind: &str, payload: T) -> Envelope<T> {
Envelope::new(kind, payload)
}
#[tracing::instrument(skip_all, fields(kind = kind, request_id = %request_id.unwrap_or_default()))]
pub async fn send<T>(&self, kind: &str, payload: T, request_id: Option<&str>) -> QueueResult<()>
where
T: Serialize,
{
self.send_labeled(kind, Labels::default(), payload, request_id)
.await
}
#[tracing::instrument(skip_all, fields(kind = kind, request_id = %request_id.unwrap_or_default()))]
pub async fn send_labeled<T>(
&self,
kind: &str,
labels: Labels,
payload: T,
request_id: Option<&str>,
) -> QueueResult<()>
where
T: Serialize,
{
self.send_keyed(kind, labels, None, payload, request_id)
.await
}
#[tracing::instrument(skip_all, fields(kind = kind, request_id = %request_id.unwrap_or_default()))]
pub async fn send_keyed<T>(
&self,
kind: &str,
labels: Labels,
idempotency_key: Option<String>,
payload: T,
request_id: Option<&str>,
) -> QueueResult<()>
where
T: Serialize,
{
let mut envelope = self.envelope(kind, payload).with_labels(labels);
if let Some(key) = idempotency_key {
envelope = envelope.with_idempotency_key(key);
}
self.send_envelope(&envelope, request_id).await
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
pub async fn send_envelope<T>(
&self,
envelope: &Envelope<T>,
request_id: Option<&str>,
) -> QueueResult<()>
where
T: Serialize,
{
let subject = format!("{EVENT_PREFIX}{}", envelope.kind);
envelope.verify_subject(EVENT_PREFIX, &subject)?;
self.dispatch(envelope, request_id).await
}
#[tracing::instrument(skip_all, fields(subject = %subject, request_id = %request_id.unwrap_or_default()))]
pub async fn publish_durable<T: Serialize>(
&self,
subject: &str,
envelope: &Envelope<T>,
request_id: Option<&str>,
) -> QueueResult<()> {
envelope.verify_event_subject(subject)?;
self.dispatch(envelope, request_id).await
}
pub async fn health(&self) -> QueueResult<()> {
let topic = self.publisher.topic(topic::HEALTH);
self.publisher.conn().health(&topic).await
}
pub async fn shutdown(&self) -> QueueResult<()> {
self.publisher.conn().shutdown().await;
Ok(())
}
#[tracing::instrument(skip_all, fields(kind = %envelope.kind, request_id = %request_id.unwrap_or_default()))]
async fn dispatch<T: Serialize>(
&self,
envelope: &Envelope<T>,
request_id: Option<&str>,
) -> QueueResult<()> {
ensure_kind(&envelope.kind)?;
let payload = serde_json::to_value(&envelope.payload).map_err(QueueError::from)?;
let area = topic::of(&envelope.kind);
let tag = label::tag(&envelope.kind, Some(&payload));
let request_id = envelope::context::request_id().or_else(|| request_id.map(str::to_owned));
let properties = envelope.properties(Some(&payload), request_id.as_deref());
let topic = self.publisher.topic(area);
let body = envelope.encode()?;
self.publisher
.publish(&topic, tag, vec![envelope.message_id()], properties, body)
.await
}
}
fn ensure_kind(kind: &str) -> QueueResult<()> {
let mut parts = kind.split('.');
let domain = parts.next().unwrap_or_default();
let name = parts.next().unwrap_or_default();
let operate = parts.next().unwrap_or_default();
if domain.is_empty() || name.is_empty() || operate.is_empty() {
return Err(QueueError::payload(format!(
"event kind `{kind}` is not a concrete event"
)));
}
Ok(())
}
const PUBLISH_ATTEMPTS: usize = 3;
const PUBLISH_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
const PUBLISH_BACKOFF: std::time::Duration = std::time::Duration::from_millis(50);
/// 统一的事件转发重试：单次尝试带超时，失败按退避重试，
/// 最终仍失败打 ERROR，便于告警发现事件链断了。
/// `context` 仅用于日志文案（如 "auth" / "email"）。
#[tracing::instrument(skip_all, fields(kind = kind, context = context, request_id = %_request_id.unwrap_or_default()))]
pub async fn retry_publish<T>(
events: &EventHandler,
kind: &str,
labels: Labels,
payload: T,
context: &str,
_request_id: Option<&str>,
) where
T: Serialize + Clone,
{
for attempt in 1..=PUBLISH_ATTEMPTS {
let published = events.send_labeled(kind, labels.clone(), payload.clone(), _request_id);
match tokio::time::timeout(PUBLISH_TIMEOUT, published).await {
Ok(Ok(())) => {
if attempt > 1 {
tracing::info!(kind, attempt, "{context} event forwarded after retry");
}
return;
}
Ok(Err(error)) => {
tracing::warn!(kind, attempt, %error, "failed to publish {context} event");
}
Err(_) => {
tracing::warn!(kind, attempt, "publishing {context} event timed out");
}
}
if attempt < PUBLISH_ATTEMPTS {
tokio::time::sleep(PUBLISH_BACKOFF * attempt as u32).await;
}
}
tracing::error!(
kind,
attempts = PUBLISH_ATTEMPTS,
"{context} event could not be forwarded to rocketmq"
);
}