main
use futures_util::TryStreamExt;
use rocketmq_model::common::message::MessageTrait;
use rocketmq_model::common::message::message_ext::MessageExt;
use serde_json::Value;
use super::{EventProperties, gitdata, kenshell, platform_events};
use crate::gateway::platform::PlatformClient;
use crate::gateway::{ConsumeResult, Subscription, Subscriptions};
use actor::ActorHost;
use std::collections::HashMap;
pub struct EventContext<'a> {
pub host: &'a ActorHost,
pub platform: &'a PlatformClient,
pub subscriptions: &'a Subscriptions,
}
pub async fn action(message: MessageExt, context: &EventContext<'_>) -> ConsumeResult {
let topic = message.topic().to_string();
let tag = message
.tags_ref()
.map(ToString::to_string)
.unwrap_or_default();
let properties = EventProperties::from_message(&message);
match topic.split_once('-').map(|(prefix, _)| prefix) {
Some("gitdata") if topic == "gitdata-auth" || topic == "gitdata-org" => {
platform_events::route(&topic, &tag, message, properties, context).await
}
Some("kenshell") => kenshell::route(&topic, &tag, message, properties, context).await,
Some("gitdata") => gitdata::route(&topic, &tag, message, properties, context).await,
_ => unimplemented(message),
}
}
pub(super) async fn deliver_to_subscribers(
topic: &str,
tag: &str,
message: MessageExt,
properties: EventProperties,
context: &EventContext<'_>,
) -> ConsumeResult {
let event_user_id = user_id(&properties);
let body = body(&message);
let subscribers = match context
.subscriptions
.subscribers(topic, tag, &properties)
.await
{
Ok(subscribers) => subscribers,
Err(_) => return retry(message),
};
let mut recipients: HashMap<i64, Vec<Subscription>> = HashMap::new();
for subscription in subscribers {
if subscription.filter.matches(&properties) {
recipients
.entry(subscription.user_id)
.or_default()
.push(subscription);
}
}
let properties = &properties;
let body = &body;
let delivered = futures_util::stream::iter(recipients.into_iter().map(Ok::<_, ()>))
.try_for_each_concurrent(32, |(recipient, destinations)| async move {
let mut value = payload(topic, tag, event_user_id, recipient, properties, body);
value["destinations"] = serde_json::to_value(destinations).map_err(|_| ())?;
context
.host
.publish_event(recipient, value)
.await
.map_err(|_| ())
})
.await;
if delivered.is_err() {
return retry(message);
}
ConsumeResult::Ack { message }
}
pub(super) fn payload(
topic: &str,
tag: &str,
event_user_id: Option<i64>,
recipient_user_id: i64,
properties: &EventProperties,
body: &Value,
) -> Value {
serde_json::json!({
"topic": topic,
"tag": tag,
"user_id": event_user_id,
"recipient_user_id": recipient_user_id,
"properties": properties,
"body": body,
})
}
pub(super) fn user_id(properties: &EventProperties) -> Option<i64> {
properties
.user_id
.as_deref()?
.parse::<i64>()
.ok()
.filter(|id| *id > 0)
}
pub(super) fn body(message: &MessageExt) -> Value {
message
.body()
.and_then(|bytes| serde_json::from_slice(&bytes).ok())
.unwrap_or(Value::Null)
}
pub(super) fn retry(message: MessageExt) -> ConsumeResult {
ConsumeResult::Retry {
message,
delay_level: 0,
}
}
pub(super) fn unimplemented(message: MessageExt) -> ConsumeResult {
retry(message)
}