Registry crate implementation

2026-01-23 18:44:11 +01:00
parent 2266634f36
commit acecd32c8e
43 changed files with 2258 additions and 912 deletions
--- a/crates/coordinator/src/dispatch.rs
+++ b/crates/coordinator/src/dispatch.rs
@@ -0,0 +1,97 @@
+/*
+ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
+ */
+
+#[cfg(feature = "redis")]
+use crate::PubSubStream;
+use crate::{Coordinator, Msg};
+
+#[allow(unused_variables)]
+impl Coordinator {
+    pub async fn publish(&self, topic: &'static str, message: Vec<u8>) -> trc::Result<()> {
+        match self {
+            #[cfg(feature = "redis")]
+            Coordinator::Redis(store) => {
+                crate::backend::redis::redis_publish(store, topic, message).await
+            }
+            #[cfg(feature = "nats")]
+            Coordinator::Nats(store) => store.publish(topic, message).await,
+            #[cfg(feature = "zenoh")]
+            Coordinator::Zenoh(store) => store.publish(topic, message).await,
+            #[cfg(feature = "kafka")]
+            Coordinator::Kafka(store) => store.publish(topic, message).await,
+            Coordinator::None => Err(trc::StoreEvent::NotSupported.into_err()),
+        }
+    }
+
+    pub async fn subscribe(&self, topic: &'static str) -> trc::Result<PubSubStream> {
+        match self {
+            #[cfg(feature = "redis")]
+            Coordinator::Redis(store) => crate::backend::redis::redis_subscribe(store, topic).await,
+            #[cfg(feature = "nats")]
+            Coordinator::Nats(store) => store.subscribe(topic).await,
+            #[cfg(feature = "zenoh")]
+            Coordinator::Zenoh(store) => store.subscribe(topic).await,
+            #[cfg(feature = "kafka")]
+            Coordinator::Kafka(store) => store.subscribe(topic).await,
+            Coordinator::None => Err(trc::StoreEvent::NotSupported.into_err()),
+        }
+    }
+
+    pub fn is_none(&self) -> bool {
+        matches!(self, Coordinator::None)
+    }
+}
+
+impl PubSubStream {
+    pub async fn next(&mut self) -> Option<Msg> {
+        match self {
+            #[cfg(feature = "redis")]
+            PubSubStream::Redis(stream) => stream.next().await,
+            #[cfg(feature = "redis")]
+            PubSubStream::RedisCluster(stream) => stream.next().await,
+            #[cfg(feature = "nats")]
+            PubSubStream::Nats(stream) => stream.next().await,
+            #[cfg(feature = "zenoh")]
+            PubSubStream::Zenoh(stream) => stream.next().await,
+            #[cfg(feature = "kafka")]
+            PubSubStream::Kafka(stream) => stream.next().await,
+            #[cfg(not(any(feature = "redis", feature = "nats")))]
+            PubSubStream::Unimplemented => None,
+        }
+    }
+}
+
+impl Msg {
+    pub fn payload(&self) -> &[u8] {
+        match self {
+            #[cfg(feature = "redis")]
+            Msg::Redis(msg) => msg.get_payload_bytes(),
+            #[cfg(feature = "nats")]
+            Msg::Nats(msg) => msg.payload.as_ref(),
+            #[cfg(feature = "zenoh")]
+            Msg::Zenoh(msg) => msg.as_slice(),
+            #[cfg(feature = "kafka")]
+            Msg::Kafka(msg) => msg.as_slice(),
+            #[cfg(not(any(feature = "redis", feature = "nats")))]
+            Msg::Unimplemented => &[],
+        }
+    }
+
+    pub fn topic(&self) -> &str {
+        match self {
+            #[cfg(feature = "redis")]
+            Msg::Redis(msg) => msg.get_channel_name(),
+            #[cfg(feature = "nats")]
+            Msg::Nats(msg) => msg.subject.as_str(),
+            #[cfg(feature = "zenoh")]
+            Msg::Zenoh(_) => "",
+            #[cfg(feature = "kafka")]
+            Msg::Kafka(_) => "",
+            #[cfg(not(any(feature = "redis", feature = "nats")))]
+            Msg::Unimplemented => "",
+        }
+    }
+}