diff --git a/crates/buqueue-core/src/delivery.rs b/crates/buqueue-core/src/core/delivery.rs similarity index 99% rename from crates/buqueue-core/src/delivery.rs rename to crates/buqueue-core/src/core/delivery.rs index 6af1230..773966c 100644 --- a/crates/buqueue-core/src/delivery.rs +++ b/crates/buqueue-core/src/core/delivery.rs @@ -22,7 +22,7 @@ //! intentional crash safety, but always call `nack()` explicitly so the //! delivery count increments correctly -use crate::error::{BuqueueError, BuqueueResult, ErrorKind}; +use crate::prelude::{BuqueueError, BuqueueResult, ErrorKind}; use bytes::Bytes; use chrono::{DateTime, Utc}; use serde::de::DeserializeOwned; diff --git a/crates/buqueue-core/src/error.rs b/crates/buqueue-core/src/core/error.rs similarity index 100% rename from crates/buqueue-core/src/error.rs rename to crates/buqueue-core/src/core/error.rs diff --git a/crates/buqueue-core/src/message.rs b/crates/buqueue-core/src/core/message.rs similarity index 99% rename from crates/buqueue-core/src/message.rs rename to crates/buqueue-core/src/core/message.rs index 1c748bd..652f8dd 100644 --- a/crates/buqueue-core/src/message.rs +++ b/crates/buqueue-core/src/core/message.rs @@ -33,7 +33,7 @@ use bytes::Bytes; use serde::Serialize; use std::collections::HashMap; -use crate::error::{BuqueueError, BuqueueResult, ErrorKind}; +use crate::prelude::{BuqueueError, BuqueueResult, ErrorKind}; /// A message to be sent into a queue /// diff --git a/crates/buqueue-core/src/core/mod.rs b/crates/buqueue-core/src/core/mod.rs new file mode 100644 index 0000000..fd79db9 --- /dev/null +++ b/crates/buqueue-core/src/core/mod.rs @@ -0,0 +1,5 @@ +//! Fundamental types, building blocks everything else depends on + +pub mod delivery; +pub mod error; +pub mod message; diff --git a/crates/buqueue-core/src/feature/dlq.rs b/crates/buqueue-core/src/feature/dlq.rs new file mode 100644 index 0000000..67bb5c5 --- /dev/null +++ b/crates/buqueue-core/src/feature/dlq.rs @@ -0,0 +1,62 @@ +//! Dead Letter Queue configuration +//! +//! Pass a `DlqConfig` to any backend builder bia `.dead_letter_queue(config)` +//! to enable automatic routing of failed messages to a dead letter destination + +/// Configuration for Dlq behavior +/// +/// When a message has been delivered and nack'd `max_receive_count` times, +/// buqeueu routes it to the `destination` instead of requeueing it. +/// +/// ## Usage +/// +/// ```rust,ignore +/// let (producer, cosumer) = SqsBackend::builder(config) +/// .dead_letter_queue(DlqConfig{ +/// destination: "https://sqs.../orders-dlq".to_string(), +/// max_receive_count: 5, +/// }) +/// .build_pair() +/// .await?; +/// ``` +/// +/// ## Backend mapping +/// +/// SQS -> The DLQ queue URL +/// NATS -> The `JetStream` stream to route to +/// `RabbitMQ` -> The dead letter exchange name +/// Redis -> The dead letter stream key +/// Kafka -> The dead letter topic name +#[derive(Debug, Clone)] +pub struct DlqConfig { + /// Where failed message are sent + /// + /// The format depends on the backend + pub destination: String, + + /// How many total delivery attempts are made before routing to the DLQ + /// + /// A value of `5` means: one original delivery + 4 redeliveries + /// On the 5th nack, the message goes to `destination` + /// + /// Must be at least 1. The recommended value for most workloads is 3-5 + pub max_receive_count: u32, +} + +impl DlqConfig { + /// Creates a new `DlqConfig`. + /// + /// # Panics + /// if `max_receive_count` is less than 1 + #[must_use] + pub fn new(destination: String, max_receive_count: u32) -> Self { + assert!( + max_receive_count >= 1, + "max_receive_count must be at least 1" + ); + Self { + destination, + max_receive_count, + } + } +} diff --git a/crates/buqueue-core/src/feature/mod.rs b/crates/buqueue-core/src/feature/mod.rs new file mode 100644 index 0000000..1039a4d --- /dev/null +++ b/crates/buqueue-core/src/feature/mod.rs @@ -0,0 +1,4 @@ +//! Traits that backends implement to plug into buqueue + +pub mod dlq; +pub mod shutdown; diff --git a/crates/buqueue-core/src/shutdown.rs b/crates/buqueue-core/src/feature/shutdown.rs similarity index 100% rename from crates/buqueue-core/src/shutdown.rs rename to crates/buqueue-core/src/feature/shutdown.rs diff --git a/crates/buqueue-core/src/lib.rs b/crates/buqueue-core/src/lib.rs index ca47ddc..e3c1a60 100644 --- a/crates/buqueue-core/src/lib.rs +++ b/crates/buqueue-core/src/lib.rs @@ -10,15 +10,12 @@ //! All traits use **native async function in traits** (AFIT) //! The `async-trait` proc-macro create is not used anywhere in buqueue //! -//! ## What lives here +//! ## Module structure //! -//! - `Message`: the type you send into a queue -//! - `Delivery`: the type you receive from a queue -//! - `QueueProducer` - the trait for sending messages -//! - `QueueConsumer` - the trait for receiving messages -//! - `QueueBackend` - the trait that ties a backend's config to its producer and consumer types -//! - `DlqConfig` - dead letter queue configuration -//! - `BuqueueError` and `ErrorKind` - structured error types +//! - [`core`]: fundamental types: `Message`, `Delivery`, errors +//! - [`traits`]: abstractions backends implement: `QueueProducer`, `QueueConsumer`, `QueueBackend` +//! - [`feature`]: opt-in behaviours: `DlqConfig`, `ShutdownHandle` +//! - [`prelude`]: everything a typical user needs in one glob import //! //! ## For backend implementors //! @@ -31,11 +28,19 @@ //! //! Then implement `QueueProducer`, `QueueConsumer` and `QueueBackend` for your backend types -#![warn(unsafe_code)] +#![warn(missing_docs)] +#![forbid(unsafe_code)] -pub mod consumer; -pub mod delivery; -pub mod error; -pub mod message; -pub mod producer; -pub mod shutdown; +pub mod core; +pub mod feature; +pub mod prelude; +pub mod traits; + +pub use core::delivery; +pub use core::error; +pub use core::message; +pub use feature::dlq; +pub use feature::shutdown; +pub use traits::backend; +pub use traits::consumer; +pub use traits::producer; diff --git a/crates/buqueue-core/src/prelude.rs b/crates/buqueue-core/src/prelude.rs new file mode 100644 index 0000000..f2196b5 --- /dev/null +++ b/crates/buqueue-core/src/prelude.rs @@ -0,0 +1,21 @@ +//! Common imports for buqueue-core +//! +//! Instead of importing multiple modules, you can do: +//! +//! ```rust +//! use buqueue_core::prelude::*; +//! ``` + +// Core types +pub use crate::core::delivery::Delivery; +pub use crate::core::error::{BuqueueError, BuqueueResult, ErrorKind}; +pub use crate::core::message::{Message, MessageBulder}; + +// Traits +pub use crate::traits::backend::{BackendBuilder, QueueBackend}; +pub use crate::traits::consumer::{DynConsumer, QueueConsumer}; +pub use crate::traits::producer::{DynProducer, MessageId, QueueProducer}; + +// Features +pub use crate::feature::dlq::DlqConfig; +pub use crate::feature::shutdown::ShutdownHandle; diff --git a/crates/buqueue-core/src/traits/backend.rs b/crates/buqueue-core/src/traits/backend.rs new file mode 100644 index 0000000..6256471 --- /dev/null +++ b/crates/buqueue-core/src/traits/backend.rs @@ -0,0 +1,101 @@ +//! The `QueueBackend` and `BackendBuilder` traits + +use crate::prelude::{ + BuqueueResult, DlqConfig, DynConsumer, DynProducer, QueueConsumer, QueueProducer, +}; + +/// A backend that can produce and consume messages +/// +/// Every buqueue backend implements this trait. The pattern is: +/// +/// ```rust,ignore +/// let(producer, consumer) = MyBackend::builder(config) +/// .dead_letter_queue(dlq_config) +/// .build_pair() +/// .await?; +/// ``` +pub trait QueueBackend: Sized { + /// Backend-specific configuration type (e.g. `KafkaConfig`, `SqsConfig`) + type Config; + /// Concrete producer type + type Producer: QueueProducer + 'static; + /// Concrete consumer type + type Consumer: QueueConsumer + 'static; + /// Builder type returned by `builder()` + type Builder: BackendBuilder; + + /// Start configuring the backend + fn builder(config: Self::Config) -> Self::Builder; +} + +/// Fluent builder returned by `QueueBackend::builder()` +pub trait BackendBuilder: Sized + Send { + /// Concrete producer type producer by this builder + type Producer: Send + 'static; + /// Concrete consumer type produced by this builder + type Consumer: Send + 'static; + + /// Configure a dead letter queue + #[must_use] + fn dead_letter_queue(self, config: DlqConfig) -> Self; + + /// Erase the producer and consumer types, returning a `DynProducer` and + /// `DynConsumer` pair. Use this when you need to select a backend at + /// runtime without generic parameters propagation through your application: + /// + /// ```rust,ignore + /// let (producer, consumer): (DynProducer, DynConsumer) = + /// match std::env::var("QUEUE_BACKEND").as_deref() { + /// Ok("sqs") => SqsBackend::builder(cfg).make_dynamic().build_pair().await?, + /// Ok("nats") => NatsBackend::builder(cfg).make_dynamic().build_pair().await?, + /// _ => MemoryBackend::builder().make_dynamic().build_pair().await?, + /// }; + /// ``` + fn make_dynamic(self) -> DynamicBuilder { + DynamicBuilder(self) + } + + /// Build both a procuder and consumer + fn build_pair( + self, + ) -> impl Future> + Send; + + /// Build only a producer + fn build_producer(self) -> impl Future> + Send; + + /// Build only a consumer + fn build_consumer(self) -> impl Future> + Send; +} + +/// Wraps a `BackendBuilder` to return `DynProducer` / `DynConsumer` +/// instead of concrete backend types. +/// +/// Produced by `BackendBuilder::make_dynamic()` +pub struct DynamicBuilder(B); + +impl BackendBuilder for DynamicBuilder +where + B::Producer: QueueProducer + 'static, + B::Consumer: QueueConsumer + 'static, +{ + type Producer = DynProducer; + type Consumer = DynConsumer; + + fn dead_letter_queue(mut self, config: DlqConfig) -> Self { + self.0 = self.0.dead_letter_queue(config); + self + } + + async fn build_pair(self) -> BuqueueResult<(Self::Producer, Self::Consumer)> { + let (p, c) = self.0.build_pair().await?; + Ok((p.into_dyn(), c.into_dyn())) + } + + async fn build_producer(self) -> BuqueueResult { + Ok(self.0.build_producer().await?.into_dyn()) + } + + async fn build_consumer(self) -> BuqueueResult { + Ok(self.0.build_consumer().await?.into_dyn()) + } +} diff --git a/crates/buqueue-core/src/consumer.rs b/crates/buqueue-core/src/traits/consumer.rs similarity index 99% rename from crates/buqueue-core/src/consumer.rs rename to crates/buqueue-core/src/traits/consumer.rs index fa43036..b12f1f9 100644 --- a/crates/buqueue-core/src/consumer.rs +++ b/crates/buqueue-core/src/traits/consumer.rs @@ -27,7 +27,7 @@ use std::pin::Pin; use futures::{Stream, stream}; -use crate::{delivery::Delivery, error::BuqueueResult, shutdown::ShutdownHandle}; +use crate::prelude::{BuqueueResult, Delivery, ShutdownHandle}; // -------- QueueConsumer ------------------------------ diff --git a/crates/buqueue-core/src/traits/mod.rs b/crates/buqueue-core/src/traits/mod.rs new file mode 100644 index 0000000..7df5add --- /dev/null +++ b/crates/buqueue-core/src/traits/mod.rs @@ -0,0 +1,5 @@ +//! Optional behavious, dead letter queues and graceful shutdown + +pub mod backend; +pub mod consumer; +pub mod producer; diff --git a/crates/buqueue-core/src/producer.rs b/crates/buqueue-core/src/traits/producer.rs similarity index 99% rename from crates/buqueue-core/src/producer.rs rename to crates/buqueue-core/src/traits/producer.rs index ce089ad..9db4975 100644 --- a/crates/buqueue-core/src/producer.rs +++ b/crates/buqueue-core/src/traits/producer.rs @@ -4,7 +4,7 @@ //! //! Uses native async function in traits, no `#[async_trait]` proc-macro -use crate::{error::BuqueueResult, message::Message}; +use crate::prelude::{BuqueueResult, Message}; use chrono::{DateTime, Utc}; use std::{future::Future, pin::Pin, sync::Arc};