bump: to 1.0.0 alpha.2 (#13)

* remove excess category

* ease: types and add cloneable impementations

* fix: tests and update docs

Author: geofmureithi

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "apalis-postgres"
-version = "1.0.0-alpha.1"
+version = "1.0.0-alpha.2"
 authors = ["Njuguna Mureithi <mureithinjuguna@gmail.com>"]
 edition = "2024"
 repository = "https://github.com/apalis-dev/apalis-postgres"

README.md

@@ -25,7 +25,7 @@ The naming is designed to clearly indicate the storage mechanism and its capabil
 ```rust,no_run
 #[tokio::main]
 async fn main() {
-    let pool = PgPool::connect(env!("DATABASE_URL").unwrap()).await.unwrap();
+    let pool = PgPool::connect(env!("DATABASE_URL")).await.unwrap();
     PostgresStorage::setup(&pool).await.unwrap();
     let mut backend = PostgresStorage::new(&pool);
 
@@ -58,7 +58,7 @@ async fn main() {
 
 #[tokio::main]
 async fn main() {
-    let pool = PostgresPool::connect(env!("DATABASE_URL").unwrap()).await.unwrap();
+    let pool = PostgresPool::connect(env!("DATABASE_URL")).await.unwrap();
     PostgresStorage::setup(&pool).await.unwrap();
 
     let lazy_strategy = StrategyBuilder::new()
@@ -67,7 +67,7 @@ async fn main() {
     let config = Config::new("queue")
         .with_poll_interval(lazy_strategy)
         .set_buffer_size(5);
-    let backend = PostgresStorage::new_with_notify(&pool, &config).await;
+    let backend = PostgresStorage::new_with_notify(&pool, &config);
 
     tokio::spawn({
         let pool = pool.clone();

src/lib.rs

@@ -94,7 +94,7 @@
 //!     let config = Config::new("queue")
 //!         .with_poll_interval(lazy_strategy)
 //!         .set_buffer_size(5);
-//!     let backend = PostgresStorage::new_with_notify(&pool, &config).await;
+//!     let backend = PostgresStorage::new_with_notify(&pool, &config);
 //!
 //!     tokio::spawn({
 //!         let pool = pool.clone();
@@ -190,7 +190,7 @@
 //! ## License
 //!
 //! Licensed under either of Apache License, Version 2.0 or MIT license at your option.
-//! 
+//!
 //! [`PostgresStorageWithListener`]: crate::PostgresStorage
 //! [`SharedPostgresStorage`]: crate::shared::SharedPostgresStorage
 use std::{fmt::Debug, marker::PhantomData};
@@ -211,7 +211,7 @@ use futures::{
     stream::{self, BoxStream, select},
 };
 use serde::Deserialize;
-use sqlx::{PgPool, postgres::PgListener};
+pub use sqlx::{PgPool, postgres::PgConnectOptions, postgres::PgListener, postgres::Postgres};
 use ulid::Ulid;
 
 use crate::{
@@ -227,7 +227,7 @@ use crate::{
 };
 
 mod ack;
-mod config;
+pub mod config;
 mod fetcher;
 mod from_row {
     use chrono::{DateTime, Utc};
@@ -278,7 +278,7 @@ mod from_row {
         }
     }
 }
-mod context {
+pub mod context {
     pub type PgContext = apalis_sql::context::SqlContext;
 }
 mod queries;
@@ -289,6 +289,11 @@ pub type PgTask<Args> = Task<Args, PgContext, Ulid>;
 
 pub type CompactType = Vec<u8>;
 
+#[derive(Debug, Clone, Default)]
+pub struct PgNotify {
+    _private: PhantomData<()>,
+}
+
 #[pin_project::pin_project]
 pub struct PostgresStorage<
     Args,
@@ -354,20 +359,17 @@ impl<Args> PostgresStorage<Args> {
         }
     }
 
-    pub async fn new_with_notify(
+    pub fn new_with_notify(
         pool: &PgPool,
         config: &Config,
-    ) -> PostgresStorage<Args, CompactType, JsonCodec<CompactType>, PgListener> {
+    ) -> PostgresStorage<Args, CompactType, JsonCodec<CompactType>, PgNotify> {
         let sink = PgSink::new(pool, config);
-        let mut fetcher = PgListener::connect_with(pool)
-            .await
-            .expect("Failed to create listener");
-        fetcher.listen("apalis::job::insert").await.unwrap();
+
         PostgresStorage {
             _marker: PhantomData,
             pool: pool.clone(),
             config: config.clone(),
-            fetcher,
+            fetcher: PgNotify::default(),
             sink,
         }
     }
@@ -383,6 +385,18 @@ impl<Args> PostgresStorage<Args> {
     }
 }
 
+impl<Args, Compact, Codec, Fetcher> PostgresStorage<Args, Compact, Codec, Fetcher> {
+    pub fn with_codec<NewCodec>(self) -> PostgresStorage<Args, Compact, NewCodec, Fetcher> {
+        PostgresStorage {
+            _marker: PhantomData,
+            sink: PgSink::new(&self.pool, &self.config),
+            pool: self.pool,
+            config: self.config,
+            fetcher: self.fetcher,
+        }
+    }
+}
+
 impl<Args, Decode> Backend
     for PostgresStorage<Args, CompactType, Decode, PgFetcher<Args, CompactType, Decode>>
 where
@@ -448,7 +462,7 @@ where
     }
 }
 
-impl<Args, Decode> Backend for PostgresStorage<Args, CompactType, Decode, PgListener>
+impl<Args, Decode> Backend for PostgresStorage<Args, CompactType, Decode, PgNotify>
 where
     Args: Send + 'static + Unpin,
     Decode: Codec<Args, Compact = CompactType> + 'static + Send,
@@ -497,6 +511,15 @@ where
         let pool = self.pool.clone();
         let worker_id = worker.name().to_owned();
         let namespace = self.config.queue().to_string();
+        let listener = async move {
+            let mut fetcher = PgListener::connect_with(&pool)
+                .await
+                .expect("Failed to create listener");
+            fetcher.listen("apalis::job::insert").await.unwrap();
+            fetcher
+        };
+        let fetcher = stream::once(listener).flat_map(|f| f.into_stream());
+        let pool = self.pool.clone();
         let register_worker = initial_heartbeat(
             self.pool.clone(),
             self.config.clone(),
@@ -505,8 +528,7 @@ where
         )
         .map(|_| Ok(None));
         let register = stream::once(register_worker);
-        let lazy_fetcher = self
-            .fetcher
+        let lazy_fetcher = fetcher
             .into_stream()
             .filter_map(move |notification| {
                 let namespace = namespace.clone();
@@ -630,7 +652,7 @@ mod tests {
         .await
         .unwrap();
         let config = Config::new("test");
-        let mut backend = PostgresStorage::new_with_notify(&pool, &config).await;
+        let mut backend = PostgresStorage::new_with_notify(&pool, &config);
 
         let mut items = stream::repeat_with(|| {
             Task::builder(42u32)