Add bb8-faktory connection pool

Author: fvaleye

Cargo.lock

@@ -1,6 +1,7 @@
 [workspace]
 members = [
     "bb8",
+    "faktory",
     "postgres",
     "redis",
 ]

Cargo.toml

@@ -0,0 +1,36 @@
+[package]
+name = "bb8-faktory"
+version = "0.1.0"
+edition = "2021"
+rust-version = "1.75"
+description = "Full-featured async (tokio-based) Faktory work server connection pool (like r2d2)"
+license = "MIT"
+repository = "https://github.com/djc/bb8"
+
+[features]
+default = []
+native-tls = ["faktory/native_tls", "dep:native-tls", "dep:tokio-native-tls", "dep:url"]
+rustls = ["faktory/rustls", "dep:tokio-rustls", "dep:rustls", "dep:url"]
+
+[dependencies]
+bb8 = { version = "0.9", path = "../bb8" }
+faktory = "0.13"
+tokio = { version = "1.0.0", features = ["io-util"] }
+url = { version = "2", optional = true }
+
+# Optional TLS dependencies
+native-tls = { version = "0.2", optional = true }
+tokio-native-tls = { version = "0.3", optional = true }
+tokio-rustls = { version = "0.25", optional = true }
+rustls = { version = "0.22", optional = true }
+
+[dev-dependencies]
+tokio = { version = "1.0.0", features = ["macros", "rt-multi-thread"] }
+
+[[example]]
+name = "native_tls"
+required-features = ["native-tls"]
+
+[[example]]
+name = "rustls"
+required-features = ["rustls"]

faktory/Cargo.toml

@@ -0,0 +1 @@
+../LICENSE

faktory/LICENSE

@@ -0,0 +1,23 @@
+use bb8::Pool;
+use bb8_faktory::FaktoryConnectionManager;
+use faktory::Job;
+
+// Enqueue a job to a Faktory work server
+//
+// Run with: FAKTORY_URL="tcp://localhost:7419" cargo run --example enqueue
+#[tokio::main]
+async fn main() {
+    let manager = FaktoryConnectionManager::from_env();
+
+    let pool = match Pool::builder().build(manager).await {
+        Ok(pool) => pool,
+        Err(e) => panic!("builder error: {e:?}"),
+    };
+
+    let mut conn = pool.get().await.unwrap();
+    conn.enqueue(Job::new("email", vec!["user@example.com"]))
+        .await
+        .unwrap();
+
+    println!("Job enqueued successfully!");
+}

faktory/examples/enqueue.rs

@@ -0,0 +1,28 @@
+use bb8::Pool;
+use bb8_faktory::FaktoryConnectionManager;
+use faktory::Job;
+
+// Enqueue multiple jobs to a Faktory work server
+//
+// Run with: FAKTORY_URL="tcp://localhost:7419" cargo run --example enqueue_many
+#[tokio::main]
+async fn main() {
+    let manager = FaktoryConnectionManager::from_env();
+
+    let pool = Pool::builder().build(manager).await.unwrap();
+
+    let mut conn = pool.get().await.unwrap();
+
+    let jobs = vec![
+        Job::new("email", vec!["user1@example.com"]),
+        Job::new("email", vec!["user2@example.com"]),
+        Job::new("email", vec!["user3@example.com"]),
+    ];
+
+    let (count, errors) = conn.enqueue_many(jobs).await.unwrap();
+
+    match errors {
+        Some(errs) => println!("Enqueued {count} jobs with errors: {errs:?}"),
+        None => println!("Successfully enqueued {count} jobs"),
+    }
+}

faktory/examples/enqueue_many.rs

@@ -0,0 +1,34 @@
+use bb8::Pool;
+use bb8_faktory::{FaktoryConnectionManager, TlsConnector};
+use faktory::Job;
+
+// Enqueue a job using native-tls for secure connections//
+// Start Faktory with TLS enabled (requires TLS configuration)
+#[tokio::main]
+async fn main() {
+    let manager = FaktoryConnectionManager::new("tcp+tls://localhost:7419")
+        .with_tls(TlsConnector::native_tls_default().unwrap());
+
+    // Or with custom connector (e.g., accepting invalid certs for testing):
+    // let connector = native_tls::TlsConnector::builder()
+    //     .danger_accept_invalid_certs(true)
+    //     .build()
+    //     .unwrap();
+    // let manager = FaktoryConnectionManager::new("tcp+tls://localhost:7419")
+    //     .with_tls(TlsConnector::NativeTls(connector));
+
+    // Or from environment (auto-selects TLS when feature enabled):
+    // let manager = FaktoryConnectionManager::from_env();
+
+    let pool = match Pool::builder().build(manager).await {
+        Ok(pool) => pool,
+        Err(e) => panic!("builder error: {e:?}"),
+    };
+
+    let mut conn = pool.get().await.unwrap();
+    conn.enqueue(Job::new("email", vec!["user@example.com"]))
+        .await
+        .unwrap();
+
+    println!("Job enqueued successfully over TLS (native-tls)!");
+}

faktory/examples/native_tls.rs

@@ -0,0 +1,33 @@
+use bb8::Pool;
+use bb8_faktory::{FaktoryConnectionManager, TlsConnector};
+use faktory::Job;
+
+// Enqueue a job using rustls for secure connections
+// Start Faktory with TLS enabled (requires TLS configuration)
+#[tokio::main]
+async fn main() {
+    let manager = FaktoryConnectionManager::new("tcp+tls://localhost:7419")
+        .with_tls(TlsConnector::rustls_default());
+
+    // Or with custom ClientConfig:
+    // let config = rustls::ClientConfig::builder()
+    //     .with_root_certificates(your_root_store)
+    //     .with_no_client_auth();
+    // let manager = FaktoryConnectionManager::new("tcp+tls://localhost:7419")
+    //     .with_tls(TlsConnector::rustls_with_config(config));
+
+    // Or from environment (auto-selects TLS when feature enabled):
+    // let manager = FaktoryConnectionManager::from_env();
+
+    let pool = match Pool::builder().build(manager).await {
+        Ok(pool) => pool,
+        Err(e) => panic!("builder error: {e:?}"),
+    };
+
+    let mut conn = pool.get().await.unwrap();
+    conn.enqueue(Job::new("email", vec!["user@example.com"]))
+        .await
+        .unwrap();
+
+    println!("Job enqueued successfully over TLS (rustls)!");
+}

faktory/examples/rustls.rs

@@ -0,0 +1,188 @@
+//! Faktory support for the `bb8` connection pool.
+#![deny(missing_docs, missing_debug_implementations)]
+
+pub use bb8;
+pub use faktory;
+
+use faktory::{Client, Error};
+
+#[cfg(any(feature = "native-tls", feature = "rustls"))]
+use url::Url;
+
+#[cfg(feature = "rustls")]
+use std::sync::Arc;
+
+/// TLS connector configuration.
+#[derive(Clone, Default)]
+pub enum TlsConnector {
+    /// Plain TCP connection.
+    #[default]
+    NoTls,
+    /// TLS using native-tls.
+    #[cfg(feature = "native-tls")]
+    NativeTls(native_tls::TlsConnector),
+    /// TLS using rustls.
+    #[cfg(feature = "rustls")]
+    Rustls(tokio_rustls::TlsConnector),
+}
+
+impl std::fmt::Debug for TlsConnector {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::NoTls => f.debug_tuple("NoTls").finish(),
+            #[cfg(feature = "native-tls")]
+            Self::NativeTls(connector) => f.debug_tuple("NativeTls").field(connector).finish(),
+            #[cfg(feature = "rustls")]
+            Self::Rustls(_) => f.debug_tuple("Rustls").field(&"..").finish(),
+        }
+    }
+}
+
+impl TlsConnector {
+    /// Create a native-tls connector with default settings.
+    #[cfg(feature = "native-tls")]
+    pub fn native_tls_default() -> Result<Self, native_tls::Error> {
+        let connector = native_tls::TlsConnector::builder().build()?;
+        Ok(Self::NativeTls(connector))
+    }
+
+    /// Create a rustls connector with default settings (empty root store).
+    ///
+    /// Note: This uses an empty certificate store. For production use,
+    /// consider using `rustls_default_with_native_certs()` or providing
+    /// your own `ClientConfig`.
+    #[cfg(feature = "rustls")]
+    pub fn rustls_default() -> Self {
+        let config = rustls::ClientConfig::builder()
+            .with_root_certificates(rustls::RootCertStore::empty())
+            .with_no_client_auth();
+        Self::Rustls(tokio_rustls::TlsConnector::from(Arc::new(config)))
+    }
+
+    /// Create a rustls connector from a custom ClientConfig.
+    #[cfg(feature = "rustls")]
+    pub fn rustls_with_config(config: rustls::ClientConfig) -> Self {
+        Self::Rustls(tokio_rustls::TlsConnector::from(Arc::new(config)))
+    }
+}
+
+/// A `bb8::ManageConnection` for `faktory::Client`.
+#[derive(Debug, Clone)]
+pub struct FaktoryConnectionManager {
+    url: String,
+    tls: TlsConnector,
+}
+
+impl FaktoryConnectionManager {
+    /// Create a new `FaktoryConnectionManager` with the specified URL.
+    ///
+    /// The URL should be in the format: `protocol://[:password@]hostname[:port]`
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use bb8_faktory::FaktoryConnectionManager;
+    ///
+    /// let manager = FaktoryConnectionManager::new("tcp://localhost:7419");
+    /// ```
+    pub fn new(url: impl Into<String>) -> Self {
+        Self {
+            url: url.into(),
+            tls: TlsConnector::default(),
+        }
+    }
+
+    /// Create a new `FaktoryConnectionManager` using environment variables.
+    ///
+    /// This reads `FAKTORY_PROVIDER` to get the env var name (defaults to `FAKTORY_URL`),
+    /// then reads that env var for the URL (defaults to `tcp://localhost:7419`).
+    ///
+    /// If a TLS feature is enabled, it will automatically use that TLS implementation
+    /// with default settings.
+    ///
+    /// # Example
+    ///
+    /// ```
+    /// use bb8_faktory::FaktoryConnectionManager;
+    ///
+    /// let manager = FaktoryConnectionManager::from_env();
+    /// ```
+    pub fn from_env() -> Self {
+        let url = std::env::var("FAKTORY_PROVIDER")
+            .ok()
+            .and_then(|provider| std::env::var(provider).ok())
+            .or_else(|| std::env::var("FAKTORY_URL").ok())
+            .unwrap_or_else(|| "tcp://localhost:7419".to_string());
+
+        #[cfg(feature = "rustls")]
+        let tls = TlsConnector::rustls_default();
+
+        #[cfg(all(feature = "native-tls", not(feature = "rustls")))]
+        let tls = TlsConnector::native_tls_default().expect("Failed to build native-tls connector");
+
+        #[cfg(not(any(feature = "native-tls", feature = "rustls")))]
+        let tls = TlsConnector::NoTls;
+
+        Self { url, tls }
+    }
+
+    /// Set the TLS connector for the connection.
+    ///
+    /// # Example
+    ///
+    /// ```ignore
+    /// use bb8_faktory::{FaktoryConnectionManager, TlsConnector};
+    ///
+    /// let manager = FaktoryConnectionManager::new("tcp+tls://localhost:7419")
+    ///     .with_tls(TlsConnector::native_tls_default().unwrap());
+    /// ```
+    pub fn with_tls(mut self, tls: TlsConnector) -> Self {
+        self.tls = tls;
+        self
+    }
+}
+
+impl bb8::ManageConnection for FaktoryConnectionManager {
+    type Connection = Client;
+    type Error = Error;
+
+    async fn connect(&self) -> Result<Self::Connection, Self::Error> {
+        match &self.tls {
+            TlsConnector::NoTls => Client::connect_to(&self.url).await,
+            #[cfg(feature = "native-tls")]
+            TlsConnector::NativeTls(connector) => {
+                let password = Url::parse(&self.url)
+                    .ok()
+                    .and_then(|url| url.password().map(|p| p.to_string()));
+                let stream = faktory::native_tls::TlsStream::with_connector(
+                    connector.clone(),
+                    Some(self.url.as_str()),
+                )
+                .await?;
+                let buffered = tokio::io::BufStream::new(stream);
+                Client::connect_with(buffered, password).await
+            }
+            #[cfg(feature = "rustls")]
+            TlsConnector::Rustls(connector) => {
+                let password = Url::parse(&self.url)
+                    .ok()
+                    .and_then(|url| url.password().map(|p| p.to_string()));
+                let stream = faktory::rustls::TlsStream::with_connector(
+                    connector.clone(),
+                    Some(self.url.as_str()),
+                )
+                .await?;
+                let buffered = tokio::io::BufStream::new(stream);
+                Client::connect_with(buffered, password).await
+            }
+        }
+    }
+
+    async fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
+        conn.current_info().await.map(|_| ())
+    }
+
+    fn has_broken(&self, _: &mut Self::Connection) -> bool {
+        false
+    }
+}