>,
state_fn: StateFn,
}
impl ProcessorMap
where
S: Clone + 'static,
{
/// Intialize a `ProcessorMap`
///
/// The state passed into this method will be passed to all jobs executed through this
/// ProcessorMap. The state argument could be useful for containing a hook into something like
/// r2d2, or the address of an actor in an actix-based system.
pub fn new(state_fn: StateFn) -> Self {
ProcessorMap {
inner: HashMap::new(),
state_fn,
}
}
/// Register a
/// [`Processor`](https://docs.rs/background-jobs/0.4.0/background_jobs/trait.Processor.html) with
/// this `ProcessorMap`.
///
/// `ProcessorMap`s are useless if no processors are registerd before workers are spawned, so
/// make sure to register all your processors up-front.
pub fn register_processor(&mut self, processor: P)
where
P: Processor + Send + 'static,
J: Job,
{
self.inner.insert(
P::NAME.to_owned(),
Box::new(move |value, state| processor.process(value, state)),
);
}
/// Process a given job
///
/// This should not be called from outside implementations of a backgoround-jobs runtime. It is
/// intended for internal use.
pub fn process_job(&self, job: JobInfo) -> impl Future- {
let opt = self
.inner
.get(job.processor())
.map(|processor| process(processor, (self.state_fn)(), job.clone()));
if let Some(fut) = opt {
Either::A(fut)
} else {
error!("Processor {} not present", job.processor());
Either::B(Ok(ReturnJobInfo::missing_processor(job.id())).into_future())
}
}
}
fn process
(
process_fn: &ProcessFn,
state: S,
job: JobInfo,
) -> impl Future- {
let args = job.args();
let id = job.id();
let processor = job.processor().to_owned();
process_fn(args, state).then(move |res| match res {
Ok(_) => {
info!("Job {} completed, {}", id, processor);
Ok(ReturnJobInfo::pass(id))
}
Err(e) => {
error!("Job {} errored, {}, {}", id, processor, e);
Ok(ReturnJobInfo::fail(id))
}
})
}