File structure

crates/buqueue-core/src/delivery.rs → 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;

crates/buqueue-core/src/message.rs → 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
 ///

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;

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,
+        }
+    }
+}

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;

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;

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;

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<Producer = Self::Producer, Consumer = Self::Consumer>;
+
+    /// 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<Self> {
+        DynamicBuilder(self)
+    }
+
+    /// Build both a procuder and consumer
+    fn build_pair(
+        self,
+    ) -> impl Future<Output = BuqueueResult<(Self::Producer, Self::Consumer)>> + Send;
+
+    /// Build only a producer
+    fn build_producer(self) -> impl Future<Output = BuqueueResult<Self::Producer>> + Send;
+
+    /// Build only a consumer
+    fn build_consumer(self) -> impl Future<Output = BuqueueResult<Self::Consumer>> + Send;
+}
+
+/// Wraps a `BackendBuilder` to return `DynProducer` / `DynConsumer`
+/// instead of concrete backend types.
+///
+/// Produced by `BackendBuilder::make_dynamic()`
+pub struct DynamicBuilder<B>(B);
+
+impl<B: BackendBuilder + Send> BackendBuilder for DynamicBuilder<B>
+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<Self::Producer> {
+        Ok(self.0.build_producer().await?.into_dyn())
+    }
+
+    async fn build_consumer(self) -> BuqueueResult<Self::Consumer> {
+        Ok(self.0.build_consumer().await?.into_dyn())
+    }
+}

crates/buqueue-core/src/consumer.rs → 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 ------------------------------
 

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;

crates/buqueue-core/src/producer.rs → 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};