95 lines
3.3 KiB
Rust
95 lines
3.3 KiB
Rust
/*
|
|
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
|
|
*
|
|
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
|
*/
|
|
|
|
use crate::{Coordinator, Msg, PubSubStream};
|
|
|
|
#[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::pubsub::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::pubsub::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,
|
|
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(),
|
|
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(_) => "",
|
|
Msg::Unimplemented => "",
|
|
}
|
|
}
|
|
}
|