add: test utils that allow backend polling during tests (#374)

* add: test utils that allow backend polling during tests

* fix: introduce testwrapper and add more tests

* fix: add sample for testing

* fix: more fixes and actions fixes

* fix: more fixes on vacuuming

* tests: improve cleanup and generic testing

* fix: improve testing and fix some found bugs

* fix: postgres query and remove incompatible tests

* fix: remove redis incompatible check

* fix: minor fixes

* fix: postgres json elements

.github/workflows/mysql.yaml

@@ -1,11 +1,4 @@
 on:
- push:
- paths:
- - "packages/apalis-sql/src/lib.rs"
- - "packages/apalis-sql/mysql.rs"
- - "packages/apalis-sql/src/migrations/mysql/**"
- - "packages/apalis-sql/src/Cargo.toml"
- - ".github/workflows/mysql.yaml"
  pull_request:
  paths:
  - "packages/apalis-sql/src/lib.rs"

.github/workflows/postgres.yaml

@@ -1,11 +1,4 @@
 on:
- push:
- paths:
- - "packages/apalis-sql/src/lib.rs"
- - "packages/apalis-sql/postgres.rs"
- - "packages/apalis-sql/src/migrations/postgres/**"
- - "packages/apalis-sql/src/Cargo.toml"
- - ".github/workflows/postgres.yaml"
  pull_request:
  paths:
  - "packages/apalis-sql/src/lib.rs"
@@ -37,4 +30,4 @@ jobs:
  toolchain: stable
  override: true
  - run: cargo test --no-default-features --features postgres,migrate,tokio-comp -- --test-threads=1
- working-directory: packages/apalis-sql
+ working-directory: packages/apalis-sql

.github/workflows/redis.yaml

@@ -1,8 +1,4 @@
 on:
- push:
- paths:
- - "packages/apalis-redis/**"
- - ".github/workflows/redis.yaml"
  pull_request:
  paths:
  - "packages/apalis-redis/**"

.github/workflows/sqlite.yaml

@@ -1,11 +1,4 @@
 on:
- push:
- paths:
- - "packages/apalis-sql/src/lib.rs"
- - "packages/apalis-sql/src/sqlite.rs"
- - "packages/apalis-sql/src/migrations/sqlite/**"
- - "packages/apalis-sql/src/Cargo.toml"
- - ".github/workflows/sqlite.yaml"
  pull_request:
  paths:
  - "packages/apalis-sql/src/lib.rs"
@@ -28,4 +21,4 @@ jobs:
  toolchain: stable
  override: true
  - run: cargo test --no-default-features --features sqlite,migrate,tokio-comp -- --test-threads=1
- working-directory: packages/apalis-sql
+ working-directory: packages/apalis-sql

examples/email-service/Cargo.toml

@@ -8,4 +8,5 @@ apalis = { path = "../../", default-features = false }
 futures-util = "0.3.0"
 serde_json = "1.0"
 serde = { version = "1.0", features = ["derive"] }
-log = "0.4"
+log = "0.4"
+email_address = "0.2.5"

examples/email-service/src/lib.rs

@@ -1,3 +1,7 @@
+use std::{str::FromStr, sync::Arc};
+
+use apalis::prelude::*;
+use email_address::EmailAddress;
 use serde::{Deserialize, Serialize};
 
 #[derive(Debug, Deserialize, Serialize, Clone)]
@@ -7,8 +11,43 @@ pub struct Email {
  pub text: String,
 }
 
-pub async fn send_email(job: Email) {
- log::info!("Attempting to send email to {}", job.to);
+pub async fn send_email(job: Email) -> Result<(), Error> {
+ let validation = EmailAddress::from_str(&job.to);
+ match validation {
+ Ok(email) => {
+ log::info!("Attempting to send email to {}", email.as_str());
+ Ok(())
+ }
+ Err(email_address::Error::InvalidCharacter) => {
+ log::error!("Killed send email job. Invalid character {}", job.to);
+ Err(Error::Abort(String::from("Invalid character. Job killed")))
+ }
+ Err(e) => Err(Error::Failed(Arc::new(Box::new(e)))),
+ }
+}
+
+pub fn example_good_email() -> Email {
+ Email {
+ subject: "Test Subject".to_string(),
+ to: "example@gmail.com".to_string(),
+ text: "Some Text".to_string(),
+ }
+}
+
+pub fn example_killed_email() -> Email {
+ Email {
+ subject: "Test Subject".to_string(),
+ to: "example@©.com".to_string(), // killed because it has © which is invalid
+ text: "Some Text".to_string(),
+ }
+}
+
+pub fn example_retry_able_email() -> Email {
+ Email {
+ subject: "Test Subject".to_string(),
+ to: "example".to_string(),
+ text: "Some Text".to_string(),
+ }
 }
 
 pub const FORM_HTML: &str = r#"

examples/redis-with-msg-pack/src/main.rs

@@ -1,4 +1,4 @@
-use std::time::Duration;
+use std::{sync::Arc, time::Duration};
 
 use anyhow::Result;
 use apalis::prelude::*;
@@ -13,11 +13,11 @@ struct MessagePack;
 impl<T: Serialize + DeserializeOwned> Codec<T, Vec<u8>> for MessagePack {
  type Error = Error;
  fn encode(&self, input: &T) -> Result<Vec<u8>, Self::Error> {
- rmp_serde::to_vec(input).map_err(|e| Error::SourceError(Box::new(e)))
+ rmp_serde::to_vec(input).map_err(|e| Error::SourceError(Arc::new(Box::new(e))))
  }
 
  fn decode(&self, compact: &Vec<u8>) -> Result<T, Self::Error> {
- rmp_serde::from_slice(compact).map_err(|e| Error::SourceError(Box::new(e)))
+ rmp_serde::from_slice(compact).map_err(|e| Error::SourceError(Arc::new(Box::new(e))))
  }
 }
 

examples/redis/src/main.rs

@@ -47,7 +47,7 @@ async fn main() -> Result<()> {
  produce_jobs(storage.clone()).await?;
 
  let worker = WorkerBuilder::new("rango-tango")
- .chain(|svc| svc.map_err(Error::Failed))
+ .chain(|svc| svc.map_err(|e| Error::Failed(Arc::new(e))))
  .layer(RateLimitLayer::new(5, Duration::from_secs(1)))
  .layer(TimeoutLayer::new(Duration::from_millis(500)))
  .data(Count::default())

packages/apalis-core/Cargo.toml

@@ -29,10 +29,11 @@ optional = true
 
 
 [features]
-default = []
+default = ["test-utils"]
 docsrs = ["document-features"]
 sleep = ["futures-timer"]
 json = ["serde_json"]
+test-utils = []
 
 [package.metadata.docs.rs]
 # defines the configuration attribute `docsrs`

packages/apalis-core/src/codec/json.rs

@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use crate::{error::Error, Codec};
 use serde::{de::DeserializeOwned, Serialize};
 use serde_json::Value;
@@ -9,32 +11,33 @@ pub struct JsonCodec;
 impl<T: Serialize + DeserializeOwned> Codec<T, Vec<u8>> for JsonCodec {
  type Error = Error;
  fn encode(&self, input: &T) -> Result<Vec<u8>, Self::Error> {
- serde_json::to_vec(input).map_err(|e| Error::SourceError(Box::new(e)))
+ serde_json::to_vec(input).map_err(|e| Error::SourceError(Arc::new(Box::new(e))))
  }
 
  fn decode(&self, compact: &Vec<u8>) -> Result<T, Self::Error> {
- serde_json::from_slice(compact).map_err(|e| Error::SourceError(Box::new(e)))
+ serde_json::from_slice(compact).map_err(|e| Error::SourceError(Arc::new(Box::new(e))))
  }
 }
 
 impl<T: Serialize + DeserializeOwned> Codec<T, String> for JsonCodec {
  type Error = Error;
  fn encode(&self, input: &T) -> Result<String, Self::Error> {
- serde_json::to_string(input).map_err(|e| Error::SourceError(Box::new(e)))
+ serde_json::to_string(input).map_err(|e| Error::SourceError(Arc::new(Box::new(e))))
  }
 
  fn decode(&self, compact: &String) -> Result<T, Self::Error> {
- serde_json::from_str(compact).map_err(|e| Error::SourceError(Box::new(e)))
+ serde_json::from_str(compact).map_err(|e| Error::SourceError(Arc::new(Box::new(e))))
  }
 }
 
 impl<T: Serialize + DeserializeOwned> Codec<T, Value> for JsonCodec {
  type Error = Error;
  fn encode(&self, input: &T) -> Result<Value, Self::Error> {
- serde_json::to_value(input).map_err(|e| Error::SourceError(Box::new(e)))
+ serde_json::to_value(input).map_err(|e| Error::SourceError(Arc::new(Box::new(e))))
  }
 
  fn decode(&self, compact: &Value) -> Result<T, Self::Error> {
- serde_json::from_value(compact.clone()).map_err(|e| Error::SourceError(Box::new(e)))
+ serde_json::from_value(compact.clone())
+ .map_err(|e| Error::SourceError(Arc::new(Box::new(e))))
  }
 }

packages/apalis-core/src/error.rs

@@ -1,4 +1,4 @@
-use std::error::Error as StdError;
+use std::{error::Error as StdError, sync::Arc};
 use thiserror::Error;
 
 use crate::worker::WorkerError;
@@ -7,28 +7,24 @@ use crate::worker::WorkerError;
 pub type BoxDynError = Box<dyn StdError + 'static + Send + Sync>;
 
 /// Represents a general error returned by a task or by internals of the platform
-#[derive(Error, Debug)]
+#[derive(Error, Debug, Clone)]
 #[non_exhaustive]
 pub enum Error {
  /// An error occurred during execution.
  #[error("FailedError: {0}")]
- Failed(#[source] BoxDynError),
+ Failed(#[source] Arc<BoxDynError>),
 
  /// A generic IO error
  #[error("IoError: {0}")]
- Io(#[from] std::io::Error),
+ Io(#[from] Arc<std::io::Error>),
 
  /// Missing some context and yet it was requested during execution.
  #[error("MissingContextError: {0}")]
  MissingContext(String),
 
  /// Execution was aborted
- #[error("AbortError")]
- Abort,
-
- /// Execution failed and job will be retried
- #[error("RetryError: {0}")]
- Retry(#[source] BoxDynError),
+ #[error("AbortError: {0}")]
+ Abort(String),
 
  /// Encountered an error during worker execution
  #[error("WorkerError: {0}")]
@@ -38,11 +34,11 @@ pub enum Error {
  /// Encountered an error during service execution
  /// This should not be used inside a task function
  #[error("Encountered an error during service execution")]
- ServiceError(#[source] BoxDynError),
+ ServiceError(#[source] Arc<BoxDynError>),
 
  #[doc(hidden)]
  /// Encountered an error during service execution
  /// This should not be used inside a task function
  #[error("Encountered an error during streaming")]
- SourceError(#[source] BoxDynError),
+ SourceError(#[source] Arc<BoxDynError>),
 }