apalis_core/worker/builder.rs
1//! Builder types for composing and building workers.
2//!
3//! The `WorkerBuilder` component is the recommended
4//! way to construct [`Worker`] instances in a flexible and
5//! composable manner.
6//!
7//! The builder pattern enables customization of various parts of a worker,
8//! in the following order:
9//!
10//! 1. Setting a backend that implements the [`Backend`] trait
11//! 2. Adding application state via [`Data`](crate::task::data)
12//! 3. Decorating the service pipeline with middleware
13//! 4. Handling lifecycle events with `on_event`
14//! 5. Providing task processing logic using [`build`](WorkerBuilder::build) that implements [`IntoWorkerService`].
15//!
16//! The [`IntoWorkerService`] trait can be used to convert a function or a service into a worker service. The following implementations are provided:
17//! - For async functions via [`task_fn`](crate::task::task_fn::task_fn)
18//! - For any type that implements the [`Service`] trait for `T: Task`
19//! - For workflows via [`apalis-workflow`](https://docs.rs/apalis-workflow)
20//!
21//! ## Basic usage
22//!
23//! ```rust,no_run
24//! # use apalis_core::worker::builder::WorkerBuilder;
25//! # use apalis_core::backend::memory::MemoryStorage;
26//! # use apalis_core::worker::context::WorkerContext;
27//! # use apalis_core::task::data::Data;
28//! # use apalis_core::backend::TaskSink;
29//! # use apalis_core::worker::ext::event_listener::EventListenerExt;
30//!
31//! # #[tokio::main]
32//! # async fn main() {
33//! # let mut in_memory = MemoryStorage::new();
34//! # in_memory.push(24).await.unwrap();
35//! async fn task(job: u32, count: Data<usize>, worker: WorkerContext) {
36//! println!("Received job: {job:?}");
37//! worker.stop().unwrap();
38//! }
39//!
40//! let worker = WorkerBuilder::new("rango-tango")
41//! .backend(in_memory)
42//! .data(0usize)
43//! .on_event(|worker, ev| {
44//! println!("On Event = {:?}", ev);
45//! })
46//! .build(task);
47//!
48//! worker.run().await.unwrap();
49//! # }
50//! ```
51//! ## Order
52//!
53//! The order in which you add layers affects how tasks are processed. Layers added earlier are wrapped by those added later.
54//!
55//! ### Why does order matter?
56//! Each layer wraps the previous one, so the outermost layer is applied last. This means that middleware added later can observe or modify the effects of earlier layers. For example, tracing added before retry will see all retries as a single operation, while tracing added after retry will log each retry attempt separately.
57//!
58//! For example:
59//! ```ignore
60//! WorkerBuilder::new()
61//! .enable_tracing()
62//! .retry(RetryPolicy::retries(3))
63//! .build(task);
64//! ```
65//! In this case, tracing is applied before retry. The tracing span may not reflect the correct attempt count.
66//!
67//! Reversing the order:
68//! ```ignore
69//! WorkerBuilder::new()
70//! .retry(RetryPolicy::retries(3))
71//! .enable_tracing()
72//! .build(task);
73//! ```
74//! Now, retry is applied first, and tracing wraps around it. The tracing span will correctly capture retries.
75//!
76//! **Tip:** Add layers in the order you want them to wrap task processing.
77use std::marker::PhantomData;
78use tower_layer::{Identity, Stack};
79use tower_service::Service;
80
81use crate::{
82 backend::{Backend, BackendConfig},
83 monitor::shutdown::Shutdown,
84 task::data::Data,
85 worker::{
86 Worker, context::WorkerContext, event::EventHandlerBuilder, service::IntoWorkerService,
87 },
88};
89
90/// Declaratively builds a [`Worker`]
91pub struct WorkerBuilder<Args, Source, Middleware> {
92 pub(crate) context: WorkerContext,
93 pub(crate) request: PhantomData<Args>,
94 pub(crate) layer: Middleware,
95 pub(crate) source: Source,
96 pub(crate) event_handler: EventHandlerBuilder,
97 pub(crate) shutdown: Option<Shutdown>,
98}
99
100impl<Args, Source, Middleware> std::fmt::Debug for WorkerBuilder<Args, Source, Middleware> {
101 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
102 f.debug_struct("WorkerBuilder")
103 .field("id", &self.context.name())
104 .field("job", &std::any::type_name::<Args>())
105 .field("layer", &std::any::type_name::<Middleware>())
106 .field("source", &std::any::type_name::<Source>())
107 .finish()
108 }
109}
110
111impl WorkerBuilder<(), (), Identity> {
112 /// Build a new [`WorkerBuilder`] instance with a name for the worker to build
113 pub fn new<T: Into<WorkerContext>>(name: T) -> Self {
114 Self {
115 request: PhantomData,
116 layer: Identity::new(),
117 source: (),
118 context: name.into(),
119 event_handler: EventHandlerBuilder::default(),
120 shutdown: None,
121 }
122 }
123}
124
125impl WorkerBuilder<(), (), Identity> {
126 /// Set the source to a backend that implements [Backend]
127 pub fn backend<NB, NJ>(self, backend: NB) -> WorkerBuilder<NJ, NB, Identity>
128 where
129 NB: Backend + BackendConfig<Args = NJ>,
130 {
131 WorkerBuilder {
132 request: PhantomData,
133 layer: self.layer,
134 source: backend,
135 context: self.context,
136 shutdown: self.shutdown,
137 event_handler: self.event_handler,
138 }
139 }
140}
141
142impl<Args, M, B> WorkerBuilder<Args, B, M>
143where
144 B: Backend,
145{
146 /// Allows of decorating the service that consumes jobs.
147 /// Allows adding multiple middleware in one call
148 pub fn chain<NewLayer>(
149 self,
150 f: impl FnOnce(M) -> NewLayer,
151 ) -> WorkerBuilder<Args, B, NewLayer> {
152 let middleware = f(self.layer);
153
154 WorkerBuilder {
155 request: self.request,
156 layer: middleware,
157 context: self.context,
158 source: self.source,
159 shutdown: self.shutdown,
160 event_handler: self.event_handler,
161 }
162 }
163 /// Allows adding middleware to the layer stack
164 pub fn layer<U>(self, layer: U) -> WorkerBuilder<Args, B, Stack<U, M>> {
165 WorkerBuilder {
166 request: self.request,
167 source: self.source,
168 layer: Stack::new(layer, self.layer),
169 context: self.context,
170 shutdown: self.shutdown,
171 event_handler: self.event_handler,
172 }
173 }
174
175 /// Adds data to the context
176 ///
177 /// This will be shared by all requests
178 pub fn data<D>(self, data: D) -> WorkerBuilder<Args, B, Stack<Data<D>, M>> {
179 WorkerBuilder {
180 request: self.request,
181 source: self.source,
182 layer: Stack::new(Data::new(data), self.layer),
183 context: self.context,
184 shutdown: self.shutdown,
185 event_handler: self.event_handler,
186 }
187 }
188
189 /// Map the backend decorating and composing a new backend
190 ///
191 /// **NOTE**
192 /// This method mutates the backend but not the sink
193 /// This means that you may need to confirm the sink changes in the decorations
194 #[doc(hidden)]
195 pub fn map_backend<F, NB>(self, map: F) -> WorkerBuilder<Args, NB, M>
196 where
197 NB: Backend<Task = B::Task>,
198 F: FnOnce(B) -> NB,
199 {
200 WorkerBuilder {
201 request: self.request,
202 source: map(self.source),
203 layer: self.layer,
204 context: self.context,
205 shutdown: self.shutdown,
206 event_handler: self.event_handler,
207 }
208 }
209}
210
211/// Finalizes the builder and constructs a [`Worker`] with the provided service
212impl<Args, B, M> WorkerBuilder<Args, B, M> {
213 /// Consumes the builder and a service to construct the final worker
214 pub fn build<W, Svc, NB>(self, service: W) -> Worker<Args, W::Backend, Svc, M>
215 where
216 W: IntoWorkerService<B, Svc, Backend = NB>,
217 B: Backend + BackendConfig,
218 NB: Backend + BackendConfig + Send + Unpin,
219 Svc: Service<NB::Task>,
220 Args: Send + 'static,
221 {
222 let svc = service.into_service(self.source);
223 let mut worker = Worker::new(self.context, svc.backend, svc.service, self.layer);
224 worker.event_handler = self
225 .event_handler
226 .write()
227 .map(|mut d| d.take())
228 .unwrap()
229 .unwrap_or(Box::new(|_, _e| {
230 debug!("[>] {_e}");
231 }));
232 worker.shutdown = self.shutdown;
233 worker
234 }
235}