mirrors/background-jobs
1
0
Fork 0
mirror of https://git.asonix.dog/asonix/background-jobs.git synced 2024-11-21 19:40:59 +00:00
Code Issues Projects Releases Packages Wiki Activity

Remove lingering .enter calls

Browse source
This commit is contained in:
Aode (lion) 2021-09-17 17:09:55 -05:00
parent ba60a5cbe0
commit 68a80085de
4 changed files with 52 additions and 55 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
examples/actix-example/src/main.rs
View file

@ -119,6 +119,6 @@ impl Job for PanickingJob {
const MAX_RETRIES: MaxRetries = MaxRetries::Count(0);
fn run(self, _: MyState) -> Self::Future {
panic!("A panicking job does not stop others from running")
panic!("boom")
}
}

1
jobs-actix/src/worker.rs
View file

@ -53,7 +53,6 @@ pub(crate) fn local_worker<State>(
let id = Uuid::new_v4();
let span = tracing::info_span!(
parent: None,
"Worker",
worker.id = tracing::field::display(&id),
worker.queue = tracing::field::display(&queue),

15
jobs-core/src/actix_job.rs
View file

@ -107,20 +107,17 @@ where
fn run(self, state: Self::State) -> Self::Future {
let (tx, rx) = oneshot::channel();
let span = Span::current();
let handle = actix_rt::spawn(async move {
let entered = span.enter();
let fut = ActixJob::run(self, state);
drop(entered);
let result = fut.instrument(span.clone()).await;
let handle = actix_rt::spawn(
async move {
let result = fut.await;
if tx.send(result).is_err() {
let entered = span.enter();
error!("Job dropped");
drop(entered);
}
});
}
.instrument(Span::current()),
);
Box::pin(async move {
handle.await.unwrap();

33
jobs-core/src/processor_map.rs
View file

@ -79,16 +79,17 @@ where
/// intended for internal use.
pub async fn process(&self, job: JobInfo) -> ReturnJobInfo {
let span = job_span(&job);
let opt = self.inner.get(job.name()).map(|name| {
let entered = span.enter();
let fut = process(Arc::clone(name), (self.state_fn)(), job.clone());
drop(entered);
fut
});
let fut = async move {
let opt = self
.inner
.get(job.name())
.map(|name| process(Arc::clone(name), (self.state_fn)(), job.clone()));
let res = if let Some(fut) = opt {
fut.instrument(span.clone()).await
fut.await
} else {
let span = Span::current();
span.record(
"exception.message",
&tracing::field::display("Not registered"),
@ -97,13 +98,14 @@ where
"exception.details",
&tracing::field::display("Not registered"),
);
let entered = span.enter();
error!("Not registered");
drop(entered);
ReturnJobInfo::unregistered(job.id())
};
res
};
fut.instrument(span).await
}
}
@ -118,14 +120,11 @@ where
pub async fn process(&self, job: JobInfo) -> ReturnJobInfo {
let span = job_span(&job);
let fut = async move {
let res = if let Some(name) = self.inner.get(job.name()) {
let entered = span.enter();
let fut = process(Arc::clone(name), self.state.clone(), job);
drop(entered);
fut.instrument(span.clone()).await
process(Arc::clone(name), self.state.clone(), job).await
} else {
let entered = span.enter();
let span = Span::current();
span.record(
"exception.message",
&tracing::field::display("Not registered"),
@ -135,11 +134,13 @@ where
&tracing::field::display("Not registered"),
);
error!("Not registered");
drop(entered);
ReturnJobInfo::unregistered(job.id())
};
res
};
fut.instrument(span).await
}
}