fix: minor updates

examples/email-service/src/lib.rs

@@ -20,7 +20,9 @@ pub async fn send_email(job: Email) -> Result<(), Error> {
         }
         Err(email_address::Error::InvalidCharacter) => {
             log::error!("Killed send email job. Invalid character {}", job.to);
-            Err(Error::Abort(Arc::new(Box::new(email_address::Error::InvalidCharacter))))
+            Err(Error::Abort(Arc::new(Box::new(
+                email_address::Error::InvalidCharacter,
+            ))))
         }
         Err(e) => Err(Error::Failed(Arc::new(Box::new(e)))),
     }

examples/fn-args/src/main.rs

@@ -19,13 +19,13 @@ struct SimpleJob {}
 
 // A task can have up to 16 arguments
 async fn simple_job(
-    _: SimpleJob, // Required, must be of the type of the job/message
-    worker_id: WorkerId, // The worker running the job, added by worker
+    _: SimpleJob,                        // Required, must be of the type of the job/message
+    worker_id: WorkerId,                 // The worker running the job, added by worker
     _worker_ctx: Context<TokioExecutor>, // The worker context, added by worker
     _sqlite: Data<SqliteStorage<SimpleJob>>, // The source, added by storage
-    task_id: Data<TaskId>, // The task id, added by storage
-    ctx: Data<SqlContext>, // The task context, added by storage
-    count: Data<Count>, // Our custom data added via layer
+    task_id: Data<TaskId>,               // The task id, added by storage
+    ctx: Data<SqlContext>,               // The task context, added by storage
+    count: Data<Count>,                  // Our custom data added via layer
 ) {
     // increment the counter
     let current = count.fetch_add(1, Ordering::Relaxed);

examples/graceful-shutdown/src/main.rs

@@ -8,7 +8,7 @@ use tracing::info;
 #[derive(Debug, Serialize, Deserialize)]
 struct LongRunningJob {}
 
-async fn long_running_task(task: LongRunningJob, worker_ctx: Context<TokioExecutor>) {
+async fn long_running_task(_task: LongRunningJob, worker_ctx: Context<TokioExecutor>) {
     loop {
         tokio::time::sleep(Duration::from_secs(1)).await; // Do some hard thing
         info!("is_shutting_down: {}", worker_ctx.is_shutting_down(),);

examples/redis-mq-example/src/main.rs

@@ -29,7 +29,7 @@ impl<T, C> Clone for RedisMq<T, C> {
             conn: self.conn.clone(),
             msg_type: PhantomData,
             config: self.config.clone(),
-            codec: self.codec.clone(),
+            codec: self.codec,
         }
     }
 }
@@ -86,7 +86,7 @@ where
         _res: &Result<Res, apalis_core::error::Error>,
     ) -> Result<(), Self::AckError> {
         self.conn
-            .delete_message(self.config.get_namespace(), &ctx)
+            .delete_message(self.config.get_namespace(), ctx)
             .await?;
         Ok(())
     }

packages/apalis-sql/src/context.rs

@@ -117,8 +117,8 @@ impl SqlContext {
     }
 
     /// Set the last error
-    pub fn set_last_error(&mut self, error: String) {
-        self.last_error = Some(error);
+    pub fn set_last_error(&mut self, error: Option<String>) {
+        self.last_error = error;
     }
 
     /// Record an attempt to execute the request

packages/apalis-sql/src/mysql.rs

@@ -74,7 +74,7 @@ where
             job_type: PhantomData,
             controller: self.controller.clone(),
             config: self.config.clone(),
-            codec: self.codec.clone(),
+            codec: self.codec,
             ack_notify: self.ack_notify.clone(),
         }
     }

packages/apalis-sql/src/postgres.rs

@@ -112,7 +112,7 @@ impl<T, C: Codec> fmt::Debug for PostgresStorage<T, C> {
             .field("controller", &self.controller)
             .field("config", &self.config)
             .field("codec", &std::any::type_name::<C>())
-            .field("ack_notify", &std::any::type_name_of_val(&self.ack_notify))
+            // .field("ack_notify", &std::any::type_name_of_val(&self.ack_notify))
             .finish()
     }
 }

packages/apalis-sql/src/sqlite.rs

@@ -56,7 +56,7 @@ impl<T> Clone for SqliteStorage<T> {
             job_type: PhantomData,
             controller: self.controller.clone(),
             config: self.config.clone(),
-            codec: self.codec.clone(),
+            codec: self.codec,
         }
     }
 }
@@ -484,7 +484,7 @@ impl<T: Sync + Send, Res: Serialize + Sync> Ack<T, Res> for SqliteStorage<T> {
             .bind(ctx.id().to_string())
             .bind(ctx.lock_by().as_ref().unwrap().to_string())
             .bind(result)
-            .bind(calculate_status(&res).to_string())
+            .bind(calculate_status(res).to_string())
             .bind(ctx.attempts().current() as i64 + 1)
             .execute(&pool)
             .await?;