b771a06
use rocketmq_common::common::message::message_ext::MessageExt;
use super::EventProperties;
use super::common::{EventContext, deliver_to_subscribers, unimplemented};
use crate::gateway::ConsumeResult;
pub(super) async fn route(
topic: &str,
tag: &str,
message: MessageExt,
properties: EventProperties,
context: &EventContext<'_>,
) -> ConsumeResult {
if !is_supported(topic, tag) {
return unimplemented(message);
}
deliver_to_subscribers(topic, tag, message, properties, context).await
}
pub(super) fn is_supported(topic: &str, tag: &str) -> bool {
let Some((prefix, area)) = super::split_topic(topic) else {
return false;
};
prefix == "gitdata"
&& !super::identity_area(area)
&& super::catalog().any(|(_, known_area, known_tag)| known_area == area && known_tag == tag)
}