Remove thread pool, switch to tokio

Cargo.toml

@@ -10,19 +10,20 @@ readme = "README.md"
 keywords = ["sidekiq", "worker", "resque", "ruby"]
 
 [dependencies]
-crossbeam-channel = "0.5.5"
+anyhow = { version = "1.0.65", features = ["backtrace"] }
+async-channel = "1.7.1"
+async-trait = "0.1.57"
 signal-hook = "0.3.14"
 chrono = { version = "0.4.22", default-features = false, features = ["serde", "clock"] }
 env_logger = "0.9.0"
-error-chain = "0.12.4" # 2 years since last release. consider replacing with a different crate
+futures = "0.3.24"
+gethostname = "0.2.3"
 log = "0.4.17"
-r2d2 = "0.8.10"
 rand = "0.8.5"
-redis = { version = "0.21.5", features = ["r2d2"] }
+redis = { version = "0.21.5", features = ["connection-manager", "tokio-comp", "tokio-native-tls-comp"] }
 serde = "1.0.137"
 serde_json = "1.0.81"
-threadpool = "1.8.1" # 2 years since last release. there is a 2.0 branch currently in development
-gethostname = "0.2.3"
+tokio = { version = "1.21.1", features = ["full"] }
 
 [dev-dependencies]
 structopt = "0.3.26"

examples/main.rs

@@ -1,4 +1,7 @@
-use sidekiq_server::{error_handler, panic_handler, printer_handler, retry_middleware, SidekiqServer};
+use anyhow::anyhow;
+use async_trait::async_trait;
+use log::*;
+use sidekiq_server::{SidekiqServer, Job, JobHandler, JobHandlerResult, JobSuccessType::*};
 use structopt::StructOpt;
 
 #[derive(StructOpt, Debug, Clone)]
@@ -36,22 +39,54 @@ fn main() {
         })
         .collect();
 
-    let mut server = SidekiqServer::new(&params.redis, params.concurrency).unwrap();
+    let runtime = {
+        let mut builder = tokio::runtime::Builder::new_multi_thread();
+        builder.worker_threads(params.concurrency + 3);
+        builder.enable_all();
+        builder.build().unwrap()
+    };
+    runtime.handle().clone().block_on(async {
+        let mut server = SidekiqServer::new(&params.redis, params.concurrency).await.unwrap();
+        server.namespace = params.namespace;
+        server.force_quite_timeout = params.timeout;
 
-    server.attach_handler("Printer", printer_handler);
-    server.attach_handler("Error", error_handler);
-    server.attach_handler("Panic", panic_handler);
+        server.attach_handler("DefaultJob", DefaultJob());
+        server.attach_handler("FailingJob", FailingJob());
+        server.attach_handler("PanickingJob", PanickingJob());
 
-    server.attach_middleware(retry_middleware);
-    for (name, weight) in queues {
-        server.new_queue(&name, weight);
+        for (name, weight) in queues {
+            server.new_queue(&name, weight);
+        }
+
+        server.start().await;
+        runtime.shutdown_background();
+    });
+}
+
+pub struct DefaultJob();
+
+#[async_trait]
+impl JobHandler for DefaultJob {
+    async fn perform(&self, job: &Job) -> JobHandlerResult {
+        info!("handling {:?}", job);
+        Ok(Success)
     }
+}
 
-    server.namespace = params.namespace;
-    server.force_quite_timeout = params.timeout;
-    start(server)
+pub struct FailingJob();
+
+#[async_trait]
+impl JobHandler for FailingJob {
+    async fn perform(&self, _job: &Job) -> JobHandlerResult {
+        Err(anyhow!("oh no"))
+    }
 }
 
-fn start(mut server: SidekiqServer) {
-    server.start();
-}
+pub struct PanickingJob();
+
+#[async_trait]
+impl JobHandler for PanickingJob {
+    async fn perform(&self, _job: &Job) -> JobHandlerResult {
+        panic!("oh no")
+    }
+}

src/job_handler.rs

@@ -1,35 +1,12 @@
+use anyhow::Result;
+use async_trait::async_trait;
+
 use crate::job::Job;
 use crate::JobSuccessType;
-use crate::JobSuccessType::*;
-use crate::errors::{ErrorKind, Result};
 
 pub type JobHandlerResult = Result<JobSuccessType>;
 
-pub trait JobHandler: Send {
-    fn handle(&mut self, job: &Job) -> JobHandlerResult;
-    fn cloned(&mut self) -> Box<dyn JobHandler>;
-}
-
-impl<F> JobHandler for F
-    where F: FnMut(&Job) -> JobHandlerResult + Copy + Send + 'static
-{
-    fn handle(&mut self, job: &Job) -> JobHandlerResult {
-        self(job)
-    }
-    fn cloned(&mut self) -> Box<dyn JobHandler> {
-        Box::new(*self)
-    }
-}
-
-pub fn printer_handler(job: &Job) -> JobHandlerResult {
-    info!("handling {:?}", job);
-    Ok(Success)
-}
-
-pub fn error_handler(_: &Job) -> JobHandlerResult {
-    Err(ErrorKind::JobHandlerError(Box::new("a".parse::<i8>().unwrap_err())).into())
+#[async_trait]
+pub trait JobHandler: Send + Sync {
+    async fn perform(&self, job: &Job) -> JobHandlerResult;
 }
-
-pub fn panic_handler(_: &Job) -> JobHandlerResult {
-    panic!("yeah, I do it deliberately")
-}

src/lib.rs

@@ -1,33 +1,18 @@
 #![forbid(unsafe_code)]
 
-extern crate serde;
 #[macro_use]
 extern crate serde_json;
 #[macro_use]
 extern crate log;
-extern crate env_logger;
-#[macro_use]
-extern crate error_chain;
-extern crate threadpool;
-extern crate redis;
-extern crate rand;
-extern crate chrono;
-#[macro_use]
-extern crate crossbeam_channel;
 
 mod server;
 mod job_handler;
-pub mod errors;
 mod job;
 mod worker;
-mod middleware;
 
 pub use server::SidekiqServer;
-pub use job_handler::{JobHandler, JobHandlerResult, printer_handler, error_handler, panic_handler};
-pub use middleware::{MiddleWare, MiddleWareResult, peek_middleware, retry_middleware,
-                     time_elapse_middleware, NextFunc};
+pub use job_handler::{JobHandler, JobHandlerResult};
 pub use job::{Job, RetryInfo};
-pub type RedisPool = r2d2::Pool<redis::Client>;
 
 #[derive(Debug, Clone)]
 pub enum JobSuccessType {