4d7cf67
pub mod consumer;
pub mod error;
pub mod metrics;
pub mod producer;
pub mod topic;
pub mod typed;
use std::sync::Arc;
use config::queue::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),
}),
}))
}
}