apalis_core/task/
task_fn.rs

1//! Utilities for adapting async functions into a task handler.
2//!
3//! The [`task_fn`] helper and the [`TaskFn`] struct in this module allow you to wrap
4//! async functions or closures into a [`TaskFn`] implementation, which can then be
5//! used in service middleware pipelines or other components expecting a [`TaskFn`].
6//!
7//! This is particularly useful when building lightweight, composable services from plain
8//! functions, including those with extracted arguments via [`FromRequest`].
9//!
10//! ## Features
11//!
12//! - Supports functions with up to 16 additional arguments beyond the core request.
13//! - Automatically applies argument extraction using the [`FromRequest`] trait.
14//! - Converts output to responses using the [`IntoResponse`] trait.
15//! - Captures function argument types at compile time via generics for static dispatch.
16//!
17//!
18//! ## Introduction
19//!
20//! The first argument of any task function is the type `Args` which is tied to the backend's task type.
21//! Eg if you are writing a task for an email service, `Args` might be a struct `Email` that includes fields like `user_id`, `subject`, and `message`.
22//!
23//! A rule of thumb is to never store database models as task arguments.
24//!
25//! Instead of doing this:
26//! ```rust
27//! struct User {
28//!     id: String,
29//!     // other fields...
30//! }
31//! struct Email {
32//!     user: User,
33//!     subject: String,
34//!     message: String,
35//! }
36//! ```
37//! Do this:
38//! ```
39//! struct Email {
40//!     user_id: String,
41//!     subject: String,
42//!     message: String,
43//! }
44//! ```
45//!
46//! All the primitive types (e.g. `String`, `u32`) can be used directly as task arguments.
47//!
48//! **Note:**
49//!
50//! > *Some backends like `apalis-cron` offer a specific `Args` type (Tick) for cron jobs while most others like `postgres` use a more generic `Args` type.*
51//!
52//! A guide for extracting complex information from tasks using [`FromRequest`] is available in [step 3](#3-implementing-custom-argument-extraction-with-fromrequest).
53//!
54//! ## Getting started
55//!
56//! Task handlers are async functions that process a task. You can use the [`task_fn`] helper
57//! to wrap your handler into a service.
58//!
59//! ```rust
60//! # use apalis_core::task::data::Data;
61//! #[derive(Clone)]
62//! struct State;
63//!
64//! // A simple handler that takes an id and injected state
65//! async fn handler(id: u32, state: Data<State>) -> String {
66//!     format!("Got id {} with state", id)
67//! }
68//! ```
69//! You would need to inject the state in your worker builder:
70//!
71//! ```rs
72//! let worker = WorkerBuilder::new()
73//!     .backend(in_memory)
74//!     .data(State)
75//!     .build(handler);
76//! ```
77//!
78//! ## Dependency Injection
79//!
80//! `apalis-core` supports default injection for common types in your handler arguments, such as:
81//! - [`WorkerContext`]: Worker context
82//! - [`TaskContext`]: The tasks context including the execution context
83//! - [`Attempt`]: Information about the current attempt
84//! - [`Data<T>`]: Injected data/state
85//! - [`TaskId`]: The unique ID of the task
86//!
87//! Example:
88//! ```rust
89//! # use apalis_core::task::{attempt::Attempt, data::Data, task_id::TaskId, context::TaskContext};
90//! #[derive(Clone)]
91//! struct State;
92//!
93//! async fn process_task(_: u32, attempt: Attempt, ctx: TaskContext, id: TaskId) -> String {
94//!     format!("Attempt {} for task {} with elapsed: {:?}", attempt.current(), id, ctx.elapsed())
95//! }
96//! ```
97//!
98//!
99//! ## Custom argument extraction with [`FromRequest`]
100//!
101//! You can extract custom types from the request by implementing [`FromRequest`].
102//!
103//! Suppose you have a task to send emails, and you want to automatically extract a `User` from the task's `user_id`:
104//!
105//! ```rust
106//! struct Email {
107//!     user_id: String,
108//!     subject: String,
109//!     message: String,
110//! }
111//!
112//! // Implement FromRequest for User
113//! # use apalis_core::task::from_request::FromRequest;
114//! # use apalis_core::task::Task;
115//! # use apalis_core::error::BoxDynError;
116//! # struct User {
117//! #    id: String,
118//! #    // other fields...
119//! # }
120//!
121//! impl FromRequest<Task<Email>> for User {
122//!     type Error = BoxDynError;
123//!     async fn from_request(req: &Task<Email>) -> Result<Self, Self::Error> {
124//!         let user_id = req.args.user_id.clone();
125//!         // Simulate fetching user from DB
126//!         Ok(User { id: user_id })
127//!     }
128//! }
129//!
130//! // Now your handler can take User directly
131//! async fn send_email(email: Email, user: User) -> Result<(), BoxDynError> {
132//!     // Use email and user
133//!     Ok(())
134//! }
135//! ```
136//!
137//! ## How It Works
138//!
139//! - [`task_fn`] wraps your handler into a [`TaskFn`] service.
140//! - Arguments are extracted using [`FromRequest`].
141//! - DI types are injected automatically.
142//! - The handler's output is converted to a response using [`IntoResponse`].
143//!
144//! [`task_fn`]: crate::task::task_fn::task_fn
145//! [`TaskFn`]: crate::task::task_fn::TaskFn
146//! [`FromRequest`]: crate::task::from_request::FromRequest
147//! [`IntoResponse`]: crate::task::into_response::IntoResponse
148//! [`Attempt`]: crate::task::attempt::Attempt
149//! [`Data<T>`]: crate::task::data::Data
150//! [`WorkerContext`]: crate::worker::context::WorkerContext
151//! [`TaskContext`]: crate::task::context::TaskContext
152//! [`TaskId`]: crate::task::task_id::TaskId
153
154use crate::backend::finalize::FinalizeBackend;
155use crate::backend::{Backend, BackendConfig};
156use crate::error::BoxDynError;
157use crate::task::Task;
158use crate::worker::service::{IntoWorkerService, WorkerService};
159use futures_util::FutureExt;
160use futures_util::future::Map;
161use std::fmt;
162use std::future::Future;
163use std::marker::PhantomData;
164use std::task::{Context, Poll};
165use tower_service::Service;
166
167use crate::task::{from_request::FromRequest, into_response::IntoResponse};
168
169/// A helper method to build a [`TaskFn`] from an async function or closure.
170///
171/// # Example
172/// ```rust
173/// # use apalis_core::task::data::Data;
174/// #[derive(Clone)]
175/// struct State {
176///     // db: Arc<DatabaseConnection>,
177/// }
178/// async fn handler(id: u32, state: Data<State>) -> String {
179///     format!("Got id {} with state", id)
180/// }   
181///```
182/// This method can take functions with up to 16 additional arguments beyond the core request.
183///
184/// See Also:
185///
186/// - [`FromRequest`]
187/// - [`IntoResponse`]
188pub fn task_fn<F, Args, FnArgs>(f: F) -> TaskFn<F, Args, FnArgs> {
189    TaskFn {
190        f,
191        req: PhantomData,
192        fn_args: PhantomData,
193    }
194}
195
196/// An executable service implemented by a closure.
197///
198/// See [`task_fn`] for more details.
199pub struct TaskFn<F, Args, FnArgs> {
200    f: F,
201    req: PhantomData<Args>,
202    fn_args: PhantomData<FnArgs>,
203}
204
205impl<T: Copy, Args, FnArgs> Copy for TaskFn<T, Args, FnArgs> {}
206
207impl<T: Clone, Args, FnArgs> Clone for TaskFn<T, Args, FnArgs> {
208    fn clone(&self) -> Self {
209        Self {
210            f: self.f.clone(),
211            req: PhantomData,
212            fn_args: PhantomData,
213        }
214    }
215}
216
217impl<T, Args, FnArgs> fmt::Debug for TaskFn<T, Args, FnArgs> {
218    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
219        f.debug_struct("TaskFn")
220            .field("f", &std::any::type_name::<T>())
221            .field(
222                "req",
223                &format_args!("PhantomData<Task<{}>>", std::any::type_name::<Args>(),),
224            )
225            .field(
226                "fn_args",
227                &format_args!("PhantomData<{}>", std::any::type_name::<FnArgs>()),
228            )
229            .finish()
230    }
231}
232
233/// The Future returned from [`TaskFn`] service.
234type FnFuture<F, O, R, E> = Map<F, fn(O) -> std::result::Result<R, E>>;
235
236macro_rules! impl_service_fn {
237    ($($K:ident),+) => {
238        #[allow(unused_parens)]
239        impl<T, F, Args: Send + 'static, R, $($K),+> Service<Task<Args>> for TaskFn<T, Args,  ($($K),+)>
240        where
241            T: FnMut(Args, $($K),+) -> F + Send + Clone + 'static,
242            F: Future + Send,
243            F::Output: IntoResponse<Output = R>,
244            $(
245                $K: FromRequest<Task<Args>> + Send,
246                < $K as FromRequest<Task<Args>> >::Error: std::error::Error + 'static + Send + Sync,
247            )+
248        {
249            type Response = R;
250            type Error = BoxDynError;
251            type Future = futures_util::future::BoxFuture<'static, Result<R, BoxDynError>>;
252
253            fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
254                Poll::Ready(Ok(()))
255            }
256
257            fn call(&mut self, task: Task<Args>) -> Self::Future {
258                let mut svc = self.f.clone();
259                #[allow(non_snake_case)]
260                let fut = async move {
261                    #[allow(clippy::double_parens)]
262                    let results: Result<($($K),+), BoxDynError> = { Ok(($($K::from_request(&task).await.map_err(|e| Box::new(e) as BoxDynError)?),+)) };
263                    match results {
264                        Ok(($($K),+)) => {
265                            let req = task.args;
266                            (svc)(req, $($K),+).map(F::Output::into_response).await
267                        }
268                        Err(e) => Err(e),
269                    }
270                };
271                fut.boxed()
272            }
273        }
274
275        #[allow(unused_parens)]
276        impl<T, Args,  F, R, B, O, $($K),+>
277            IntoWorkerService<B, TaskFn<T, Args,  ($($K),+)>> for T
278        where
279            B: Backend + BackendConfig<Args = Args>,
280            B::Kind: FinalizeBackend<B, Args, Backend = O>,
281            O: Backend<Task = Task<Args>>,
282            T: FnMut(Args, $($K),+) -> F + Send + Clone + 'static,
283            F: Future + Send,
284            Args: Send + 'static,
285            F::Output: IntoResponse<Output = R>,
286            TaskFn<T, Args,  ($($K),+)>: Service<O::Task>,
287
288            $(
289                $K: FromRequest<Task<Args>> + Send,
290                < $K as FromRequest<Task<Args>> >::Error: std::error::Error + 'static + Send + Sync,
291            )+
292        {
293            type Backend = O;
294            type Task = Task<Args>;
295
296            fn into_service(self, backend: B) -> WorkerService<O, TaskFn<T, Args,  ($($K),+)>> {
297                let backend = B::Kind::finalize(backend);
298                WorkerService {
299                    backend,
300                    service: task_fn(self)
301                }
302            }
303        }
304    };
305}
306
307impl<T, F, Args, R> Service<Task<Args>> for TaskFn<T, Args, ()>
308where
309    T: FnMut(Args) -> F,
310    F: Future,
311    F::Output: IntoResponse<Output = R>,
312{
313    type Response = R;
314    type Error = BoxDynError;
315    type Future = FnFuture<F, F::Output, R, BoxDynError>;
316
317    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
318        Poll::Ready(Ok(()))
319    }
320
321    fn call(&mut self, task: Task<Args>) -> Self::Future {
322        let fut = (self.f)(task.args);
323
324        fut.map(F::Output::into_response)
325    }
326}
327
328impl<T, Args, F, R, B, O> IntoWorkerService<B, TaskFn<T, Args, ()>> for T
329where
330    T: FnMut(Args) -> F,
331    F: Future,
332    F::Output: IntoResponse<Output = R>,
333    B: Backend + BackendConfig<Args = Args>,
334    B::Kind: FinalizeBackend<B, Args, Backend = O>,
335    O: Backend<Task = Task<Args>>,
336    Args: Send,
337    TaskFn<T, Args, ()>: Service<O::Task>,
338{
339    type Backend = O;
340    type Task = Task<Args>;
341
342    fn into_service(self, backend: B) -> WorkerService<O, TaskFn<T, Args, ()>> {
343        let backend = B::Kind::finalize(backend);
344        WorkerService {
345            backend,
346            service: task_fn(self),
347        }
348    }
349}
350
351impl<Args, S, B, O> IntoWorkerService<B, S> for S
352where
353    B: Backend + BackendConfig<Args = Args>,
354    B::Kind: FinalizeBackend<B, Args, Backend = O>,
355    O: Backend<Task = Task<Args>>,
356    S: Service<O::Task>,
357{
358    type Backend = O;
359    type Task = Task<Args>;
360    fn into_service(self, backend: B) -> WorkerService<O, S> {
361        let backend = B::Kind::finalize(backend);
362        WorkerService {
363            backend,
364            service: self,
365        }
366    }
367}
368
369impl_service_fn!(A);
370impl_service_fn!(A1, A2);
371impl_service_fn!(A1, A2, A3);
372impl_service_fn!(A1, A2, A3, A4);
373impl_service_fn!(A1, A2, A3, A4, A5);
374impl_service_fn!(A1, A2, A3, A4, A5, A6);
375impl_service_fn!(A1, A2, A3, A4, A5, A6, A7);
376impl_service_fn!(A1, A2, A3, A4, A5, A6, A7, A8);
377impl_service_fn!(A1, A2, A3, A4, A5, A6, A7, A8, A9);
378impl_service_fn!(A1, A2, A3, A4, A5, A6, A7, A8, A9, A10);
379impl_service_fn!(A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11);
380impl_service_fn!(A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12);
381impl_service_fn!(A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13);
382impl_service_fn!(A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13, A14);
383impl_service_fn!(
384    A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13, A14, A15
385);
386impl_service_fn!(
387    A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13, A14, A15, A16
388);
apalis_core/task/task_fn.rs

apalis_core/task/
task_fn.rs
