1a43c72
pub mod config;
pub mod consumer;
pub mod error;
pub mod producer;
pub mod topic;
pub mod typed;
use std::sync::Arc;
pub use config::AppQueueSettings;
pub use consumer::{AutoOffsetReset, RawMessage, SubscribeHandle, SubscriptionMode};
pub use error::QueueError;
pub use producer::KafkaPublisher;
pub use typed::{TypedMessage, TypedMessageMeta, TypedQueue};
#[derive(Clone)]
pub struct Queue {
inner: Arc<QueueInner>,
}
struct QueueInner {
publisher: KafkaPublisher,
cfg: Arc<AppQueueSettings>,
}
impl Queue {
pub fn publisher(&self) -> &KafkaPublisher {
&self.inner.publisher
}
pub fn config(&self) -> &AppQueueSettings {
&self.inner.cfg
}
pub(crate) fn config_arc(&self) -> Arc<AppQueueSettings> {
self.inner.cfg.clone()
}
pub(crate) fn topic_prefix(&self) -> String {
self.inner.cfg.topic_prefix.clone()
}
pub async fn publish_raw(
&self,
topic: &str,
key: Option<&[u8]>,
payload: &[u8],
) -> Result<(), QueueError> {
self.inner.publisher.publish(topic, key, payload).await
}
pub fn subscribe_raw<F, Fut>(
&self,
mode: SubscriptionMode,
topics: &[&str],
handler: F,
) -> Result<SubscribeHandle, QueueError>
where
F: Fn(RawMessage) -> Fut + Send + Sync + 'static,
Fut: std::future::Future<Output = Result<(), QueueError>> + Send + 'static,
{
if topics.is_empty() {
return Err(QueueError::Topic(
"subscribe requires at least one topic".to_string(),
));
}
let cfg = self.config_arc();
let (consumer, group_id) = consumer::build_consumer(&cfg, &mode, topics)?;
let handler: consumer::RawHandler = Arc::new(move |raw| Box::pin(handler(raw)));
Ok(consumer::spawn_raw_loop(cfg, consumer, group_id, handler))
}
pub fn typed<T>(&self) -> TypedQueue<'_, T>
where
T: serde::Serialize + serde::de::DeserializeOwned + Send + Sync + 'static,
{
TypedQueue::new(self)
}
pub async fn shutdown(&self) -> Result<(), QueueError> {
Ok(())
}
}
pub struct QueueBuilder {
cfg: AppQueueSettings,
}
impl QueueBuilder {
pub fn from_config(cfg: AppQueueSettings) -> Self {
Self { cfg }
}
pub async fn build(self) -> Result<Option<Queue>, QueueError> {
if !self.cfg.enabled {
return Ok(None);
}
self.cfg.is_valid().map_err(QueueError::Config)?;
let publisher = KafkaPublisher::from_config(&self.cfg)?;
Ok(Some(Queue {
inner: Arc::new(QueueInner {
publisher,
cfg: Arc::new(self.cfg),
}),
}))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn disabled_returns_none() {
let cfg = AppQueueSettings {
enabled: false,
..Default::default()
};
let queue = QueueBuilder::from_config(cfg).build().await.unwrap();
assert!(queue.is_none());
}
#[test]
fn invalid_config_rejected() {
let cfg = AppQueueSettings {
enabled: true,
brokers: "".into(),
..Default::default()
};
assert!(cfg.is_valid().is_err());
}
#[test]
fn invalid_offset_reset_rejected() {
let cfg = AppQueueSettings {
enabled: true,
auto_offset_reset: "middle".into(),
..Default::default()
};
assert!(cfg.is_valid().is_err());
}
#[test]
fn invalid_compression_rejected() {
let cfg = AppQueueSettings {
enabled: true,
producer_compression: "rar".into(),
..Default::default()
};
assert!(cfg.is_valid().is_err());
}
#[test]
fn invalid_acks_rejected() {
let cfg = AppQueueSettings {
enabled: true,
producer_acks: "2".into(),
..Default::default()
};
assert!(cfg.is_valid().is_err());
}
#[test]
fn subscribe_requires_topic() {
let cfg = AppQueueSettings::default();
let publisher = KafkaPublisher::from_config(&cfg).unwrap();
let queue = Queue {
inner: Arc::new(QueueInner {
publisher,
cfg: Arc::new(cfg),
}),
};
let err = queue
.subscribe_raw::<_, std::future::Ready<Result<(), QueueError>>>(
SubscriptionMode::Group {
group_id: "g".into(),
auto_offset_reset: AutoOffsetReset::Earliest,
},
&[],
|_| std::future::ready(Ok(())),
)
.unwrap_err();
assert!(matches!(err, QueueError::Topic(_)));
}
}