filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Bootstrap from registry - part 7

Browse Source
This commit is contained in:
mdecimus
2026-02-03 12:23:05 +01:00
parent 262aed19af
commit 7c8be27fcf
93 changed files with 1206 additions and 9398 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
crates/coordinator/src/backend/kafka/mod.rs
View File

@@ -4,14 +4,16 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::sync::Arc;
use crate::Coordinator;
use rdkafka::{
ClientConfig, ClientContext, TopicPartitionList,
consumer::{BaseConsumer, ConsumerContext, Rebalance, StreamConsumer},
error::KafkaResult,
producer::FutureProducer,
};
use std::{fmt::Debug, time::Duration};
use utils::config::{Config, utils::AsKey};
use registry::schema::structs::KafkaCoordinator;
pub mod pubsub;
@@ -23,70 +25,41 @@ pub struct KafkaPubSub {
}
impl KafkaPubSub {
pub async fn open(config: &mut Config, prefix: impl AsKey) -> Option<Self> {
let prefix = prefix.as_key();
let brokers = config
.values((&prefix, "brokers"))
.map(|(_, v)| v.to_string())
.collect::<Vec<_>>();
if brokers.is_empty() {
config.new_build_error((&prefix, "brokers"), "No Kafka brokers specified");
return None;
pub async fn open(config: KafkaCoordinator) -> Result<Coordinator, String> {
if config.brokers.is_empty() {
return Err("No Kafka brokers specified".to_string());
}
let brokers = config.brokers.join(",");
let mut consumer_builder = ClientConfig::new();
consumer_builder
.set(
"group.id",
config.value_require_non_empty((&prefix, "group-id"))?,
)
.set(
"bootstrap.servers",
config.value_require_non_empty((&prefix, "brokers"))?,
)
.set("group.id", config.group_id)
.set("bootstrap.servers", &brokers)
.set("enable.partition.eof", "false")
.set(
"session.timeout.ms",
config
.property_or_default((&prefix, "timeout.session"), "5s")
.unwrap_or(Duration::from_secs(5))
.as_millis()
.to_string(),
config.timeout_session.as_millis().to_string(),
)
.set("enable.auto.commit", "true");
let producer = ClientConfig::new()
.set(
"bootstrap.servers",
config.value_require_non_empty((&prefix, "brokers"))?,
)
.set("bootstrap.servers", brokers)
.set(
"message.timeout.ms",
config
.property_or_default((&prefix, "timeout.message"), "5s")
.unwrap_or(Duration::from_secs(5))
.as_millis()
.to_string(),
config.timeout_message.as_millis().to_string(),
)
.create()
.map_err(|err| {
config.new_build_error(
(&prefix, "config"),
format!("Failed to create Kafka producer: {}", err),
);
})
.ok()?;
.map_err(|err| format!("Failed to create Kafka producer: {}", err))?;
KafkaPubSub {
Ok(Coordinator::Kafka(Arc::new(KafkaPubSub {
consumer_builder,
producer,
}
.into()
})))
}
}
impl Debug for KafkaPubSub {
impl std::fmt::Debug for KafkaPubSub {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("KafkaPubSub").finish()
}

5
crates/coordinator/src/backend/kafka/pubsub.rs
View File

@@ -4,15 +4,14 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::time::Duration;
use super::{CustomContext, KafkaPubSub, LoggingConsumer};
use crate::dispatch::pubsub::{Msg, PubSubStream};
use crate::{Msg, PubSubStream};
use rdkafka::{
Message,
consumer::{CommitMode, Consumer, StreamConsumer},
producer::FutureRecord,
};
use std::time::Duration;
use trc::{ClusterEvent, Error, EventType};
pub struct KafkaPubSubStream {

99
crates/coordinator/src/backend/nats/mod.rs
View File

@@ -4,10 +4,11 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::time::Duration;
use std::sync::Arc;
use crate::Coordinator;
use async_nats::Client;
use utils::config::{Config, utils::AsKey};
use registry::schema::structs::NatsCoordinator;
pub mod pubsub;
@@ -17,92 +18,36 @@ pub struct NatsPubSub {
}
impl NatsPubSub {
pub async fn open(config: &mut Config, prefix: impl AsKey) -> Option<Self> {
let prefix = prefix.as_key();
let urls = config
.values((&prefix, "address"))
.map(|(_, v)| v.to_string())
.collect::<Vec<_>>();
if urls.is_empty() {
config.new_build_error((&prefix, "address"), "No Nats addresses specified");
return None;
pub async fn open(config: NatsCoordinator) -> Result<Coordinator, String> {
if config.addresses.is_empty() {
return Err("No Nats addresses specified".to_string());
}
let mut opts = async_nats::ConnectOptions::new()
.max_reconnects(
config
.property_or_default::<Option<usize>>((&prefix, "max-reconnects"), "false")
.unwrap_or_default(),
)
.connection_timeout(
config
.property_or_default((&prefix, "timeout.connection"), "5s")
.unwrap_or_else(|| Duration::from_secs(5)),
)
.request_timeout(
config
.property_or_default::<Option<Duration>>((&prefix, "timeout.request"), "10s")
.unwrap_or_else(|| Some(Duration::from_secs(10))),
)
.ping_interval(
config
.property_or_default((&prefix, "ping-interval"), "60s")
.unwrap_or_else(|| Duration::from_secs(5)),
)
.client_capacity(
config
.property_or_default((&prefix, "capacity.client"), "2048")
.unwrap_or(2048),
)
.subscription_capacity(
config
.property_or_default((&prefix, "capacity.subscription"), "65536")
.unwrap_or(65536),
)
.read_buffer_capacity(
config
.property_or_default((&prefix, "capacity.read-buffer"), "65535")
.unwrap_or(65535),
)
.require_tls(
config
.property_or_default((&prefix, "tls.enable"), "false")
.unwrap_or_default(),
);
.max_reconnects(config.max_reconnects.map(|v| v as usize))
.connection_timeout(config.timeout_connection.into_inner())
.request_timeout(config.timeout_request.into_inner().into())
.ping_interval(config.ping_interval.into_inner())
.client_capacity(config.capacity_client as usize)
.subscription_capacity(config.capacity_subscription as usize)
.read_buffer_capacity(config.capacity_read_buffer as u16)
.require_tls(config.use_tls);
if config
.property_or_default((&prefix, "no-echo"), "true")
.unwrap_or(true)
{
if config.no_echo {
opts = opts.no_echo();
}
if let (Some(user), Some(pass)) = (
config.value((&prefix, "user")),
config.value((&prefix, "password")),
) {
if let (Some(user), Some(pass)) = (config.auth_username, config.auth_secret) {
opts = opts.user_and_password(user.to_string(), pass.to_string());
} else if let Some(credentials) = config.value((&prefix, "credentials")) {
} else if let Some(credentials) = config.credentials {
opts = opts
.credentials(credentials)
.map_err(|err| {
config.new_build_error(
(&prefix, "credentials"),
format!("Failed to parse Nats credentials: {}", err),
);
})
.ok()?;
.credentials(&credentials)
.map_err(|err| format!("Failed to parse Nats credentials: {}", err))?;
}
async_nats::connect_with_options(urls, opts)
async_nats::connect_with_options(config.addresses, opts)
.await
.map_err(|err| {
config.new_build_error(
(&prefix, "urls"),
format!("Failed to connect to Nats: {}", err),
);
})
.map(|client| NatsPubSub { client })
.ok()
.map(|client| Coordinator::Nats(Arc::new(NatsPubSub { client })))
.map_err(|err| format!("Failed to connect to Nats: {}", err))
}
}

2
crates/coordinator/src/backend/nats/pubsub.rs
View File

@@ -5,7 +5,7 @@
*/
use super::NatsPubSub;
use crate::dispatch::pubsub::{Msg, PubSubStream};
use crate::{Msg, PubSubStream};
use futures::StreamExt;
use trc::{ClusterEvent, Error, EventType};

100
crates/coordinator/src/backend/redis/mod.rs
View File

@@ -4,102 +4,4 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::{Msg, PubSubStream};
use futures::StreamExt;
use redis::{AsyncCommands, PushInfo, cluster::ClusterConfig, cluster_async::ClusterConnection};
use std::fmt::Display;
use store::backend::redis::{RedisPool, RedisStore};
use tokio::sync::mpsc::UnboundedReceiver;
pub struct RedisPubSubStream {
stream: redis::aio::PubSubStream,
}
pub struct RedisClusterPubSubStream {
_conn: ClusterConnection,
rx: UnboundedReceiver<PushInfo>,
}
pub(crate) async fn redis_publish(
redis: &RedisStore,
topic: &'static str,
message: Vec<u8>,
) -> trc::Result<()> {
match &redis.pool {
RedisPool::Single(pool) => pool
.get()
.await
.map_err(into_error)?
.as_mut()
.publish(topic, message)
.await
.map_err(into_error),
RedisPool::Cluster(pool) => pool
.get()
.await
.map_err(into_error)?
.as_mut()
.publish(topic, message)
.await
.map_err(into_error),
}
}
pub(crate) async fn redis_subscribe(
redis: &RedisStore,
topic: &'static str,
) -> trc::Result<PubSubStream> {
match &redis.pool {
RedisPool::Single(pool) => {
let mut pubsub = pool
.manager()
.client
.get_async_pubsub()
.await
.map_err(into_error)?;
pubsub.subscribe(topic).await.map_err(into_error)?;
Ok(PubSubStream::Redis(RedisPubSubStream {
stream: pubsub.into_on_message(),
}))
}
RedisPool::Cluster(pool) => {
let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
let mut _conn = pool
.manager()
.client
.get_async_connection_with_config(ClusterConfig::default().set_push_sender(tx))
.await
.map_err(into_error)?;
_conn.subscribe(topic).await.map_err(into_error)?;
Ok(PubSubStream::RedisCluster(RedisClusterPubSubStream {
_conn,
rx,
}))
}
}
}
impl RedisPubSubStream {
pub async fn next(&mut self) -> Option<Msg> {
self.stream.next().await.map(Msg::Redis)
}
}
impl RedisClusterPubSubStream {
pub async fn next(&mut self) -> Option<Msg> {
loop {
if let Some(msg) = redis::Msg::from_push_info(self.rx.recv().await?) {
return Some(Msg::Redis(msg));
}
}
}
}
#[inline(always)]
fn into_error(err: impl Display) -> trc::Error {
trc::StoreEvent::RedisError.reason(err)
}
pub mod pubsub;

105
crates/coordinator/src/backend/redis/pubsub.rs Normal file
View File

@@ -0,0 +1,105 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::{Msg, PubSubStream};
use futures::StreamExt;
use redis::{AsyncCommands, PushInfo, cluster::ClusterConfig, cluster_async::ClusterConnection};
use std::fmt::Display;
use store::backend::redis::{RedisPool, RedisStore};
use tokio::sync::mpsc::UnboundedReceiver;
pub struct RedisPubSubStream {
stream: redis::aio::PubSubStream,
}
pub struct RedisClusterPubSubStream {
_conn: ClusterConnection,
rx: UnboundedReceiver<PushInfo>,
}
pub(crate) async fn redis_publish(
redis: &RedisStore,
topic: &'static str,
message: Vec<u8>,
) -> trc::Result<()> {
match &redis.pool {
RedisPool::Single(pool) => pool
.get()
.await
.map_err(into_error)?
.as_mut()
.publish(topic, message)
.await
.map_err(into_error),
RedisPool::Cluster(pool) => pool
.get()
.await
.map_err(into_error)?
.as_mut()
.publish(topic, message)
.await
.map_err(into_error),
}
}
pub(crate) async fn redis_subscribe(
redis: &RedisStore,
topic: &'static str,
) -> trc::Result<PubSubStream> {
match &redis.pool {
RedisPool::Single(pool) => {
let mut pubsub = pool
.manager()
.client
.get_async_pubsub()
.await
.map_err(into_error)?;
pubsub.subscribe(topic).await.map_err(into_error)?;
Ok(PubSubStream::Redis(RedisPubSubStream {
stream: pubsub.into_on_message(),
}))
}
RedisPool::Cluster(pool) => {
let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
let mut _conn = pool
.manager()
.client
.get_async_connection_with_config(ClusterConfig::default().set_push_sender(tx))
.await
.map_err(into_error)?;
_conn.subscribe(topic).await.map_err(into_error)?;
Ok(PubSubStream::RedisCluster(RedisClusterPubSubStream {
_conn,
rx,
}))
}
}
}
impl RedisPubSubStream {
pub async fn next(&mut self) -> Option<Msg> {
self.stream.next().await.map(Msg::Redis)
}
}
impl RedisClusterPubSubStream {
pub async fn next(&mut self) -> Option<Msg> {
loop {
if let Some(msg) = redis::Msg::from_push_info(self.rx.recv().await?) {
return Some(Msg::Redis(msg));
}
}
}
}
#[inline(always)]
fn into_error(err: impl Display) -> trc::Error {
trc::StoreEvent::RedisError.reason(err)
}

27
crates/coordinator/src/backend/zenoh/mod.rs
View File

@@ -4,7 +4,9 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use utils::config::{Config, utils::AsKey};
use registry::schema::structs::ZenohCoordinator;
use crate::Coordinator;
pub mod pubsub;
#[derive(Debug)]
@@ -13,26 +15,13 @@ pub struct ZenohPubSub {
}
impl ZenohPubSub {
pub async fn open(config: &mut Config, prefix: impl AsKey) -> Option<Self> {
let prefix = prefix.as_key();
let zenoh_config =
zenoh::Config::from_json5(config.value_require_non_empty((&prefix, "config"))?)
.map_err(|err| {
config.new_build_error(
(&prefix, "config"),
format!("Invalid zenoh config: {}", err),
);
})
.ok()?;
pub async fn open(config: ZenohCoordinator) -> Result<Coordinator, String> {
let zenoh_config = zenoh::Config::from_json5(&config.config)
.map_err(|err| format!("Invalid Zenoh config: {}", err))?;
zenoh::open(zenoh_config)
.await
.map_err(|err| {
config.new_build_error(
(&prefix, "config"),
format!("Failed to create zenoh session: {}", err),
);
})
.map_err(|err| format!("Failed to create Zenoh session: {}", err))
.map(|session| ZenohPubSub { session })
.ok()
.map(|store| Coordinator::Zenoh(std::sync::Arc::new(store)))
}
}

2
crates/coordinator/src/backend/zenoh/pubsub.rs
View File

@@ -5,7 +5,7 @@
*/
use super::ZenohPubSub;
use crate::dispatch::pubsub::{Msg, PubSubStream};
use crate::{Msg, PubSubStream};
use trc::{ClusterEvent, Error, EventType};
pub struct ZenohPubSubStream {

70
crates/coordinator/src/bootstrap.rs Normal file
View File

@@ -0,0 +1,70 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::Coordinator;
use registry::schema::{prelude::Object, structs};
use store::{InMemoryStore, registry::bootstrap::Bootstrap};
#[allow(unreachable_patterns)]
impl Coordinator {
pub async fn build(bp: &mut Bootstrap, in_memory: &InMemoryStore) -> Option<Self> {
let result = match bp.setting_infallible::<structs::Coordinator>().await {
structs::Coordinator::Disabled => Ok(Coordinator::None),
#[cfg(feature = "redis")]
structs::Coordinator::Default => {
if let InMemoryStore::Redis(redis) = &in_memory {
Ok(Coordinator::Redis(redis.clone()))
} else {
Err(
"Default coordinator requires Redis or Redis Cluster in-memory backend"
.to_string(),
)
}
}
#[cfg(feature = "kafka")]
structs::Coordinator::Kafka(kafka_coordinator) => {
crate::backend::kafka::KafkaPubSub::open(kafka_coordinator).await
}
#[cfg(feature = "nats")]
structs::Coordinator::Nats(nats_coordinator) => {
crate::backend::nats::NatsPubSub::open(nats_coordinator).await
}
#[cfg(feature = "zenoh")]
structs::Coordinator::Zenoh(zenoh_coordinator) => {
crate::backend::zenoh::ZenohPubSub::open(zenoh_coordinator).await
}
#[cfg(feature = "redis")]
structs::Coordinator::Redis(redis_store) => {
store::backend::redis::RedisStore::open_single(redis_store)
.await
.map(unwrap_redis)
}
#[cfg(feature = "redis")]
structs::Coordinator::RedisCluster(redis_cluster_store) => {
store::backend::redis::RedisStore::open_cluster(redis_cluster_store)
.await
.map(unwrap_redis)
}
_ => Err("Binary was not compiled with the selected coordinator backend".to_string()),
};
match result {
Ok(store) => Some(store),
Err(err) => {
bp.build_error(Object::Coordinator.singleton(), err);
None
}
}
}
}
fn unwrap_redis(store: InMemoryStore) -> Coordinator {
if let InMemoryStore::Redis(redis) = store {
Coordinator::Redis(redis)
} else {
unreachable!()
}
}

6
crates/coordinator/src/dispatch.rs
View File

@@ -12,7 +12,7 @@ impl Coordinator {
match self {
#[cfg(feature = "redis")]
Coordinator::Redis(store) => {
crate::backend::redis::redis_publish(store, topic, message).await
crate::backend::redis::pubsub::redis_publish(store, topic, message).await
}
#[cfg(feature = "nats")]
Coordinator::Nats(store) => store.publish(topic, message).await,
@@ -27,7 +27,9 @@ impl Coordinator {
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,
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")]

5
crates/coordinator/src/lib.rs
View File

@@ -8,6 +8,7 @@
use std::sync::Arc;
pub mod backend;
pub mod bootstrap;
pub mod dispatch;
#[derive(Clone, Default)]
@@ -26,9 +27,9 @@ pub enum Coordinator {
pub enum PubSubStream {
#[cfg(feature = "redis")]
Redis(crate::backend::redis::RedisPubSubStream),
Redis(crate::backend::redis::pubsub::RedisPubSubStream),
#[cfg(feature = "redis")]
RedisCluster(crate::backend::redis::RedisClusterPubSubStream),
RedisCluster(crate::backend::redis::pubsub::RedisClusterPubSubStream),
#[cfg(feature = "nats")]
Nats(crate::backend::nats::pubsub::NatsPubSubStream),
#[cfg(feature = "zenoh")]