Skip to content

Commit eb503d1

Browse files
stav121stav121
committed
update(core): v0.2.3
* adds task step statuses
1 parent f957382 commit eb503d1

10 files changed

+116
-85
lines changed

‎Cargo.toml

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
[package]
22
name = "tasklet"
3-
version = "0.2.2"
3+
version = "0.2.3"
44
authors = ["Stavros Grigoriou <[email protected]>"]
55
edition = "2021"
66
rust-version = "1.75.0"

‎README.md

+9-8
Original file line numberDiff line numberDiff line change
@@ -33,16 +33,18 @@ In your `Cargo.toml` add:
3333

3434
```
3535
[dependencies]
36-
tasklet = "0.2.2"
36+
tasklet = "0.2.3"
3737
```
3838

3939
## Example
4040

4141
Find more examples in the [examples](/examples) folder.
4242

4343
```rust
44-
use log::{error, info};
44+
use log::info;
4545
use simple_logger::SimpleLogger;
46+
use tasklet::task::TaskStepStatusErr::Error;
47+
use tasklet::task::TaskStepStatusOk::Success;
4648
use tasklet::{TaskBuilder, TaskScheduler};
4749

4850
/// A simple example of a task with two steps,
@@ -65,19 +67,18 @@ async fn main() {
6567
TaskBuilder::new(chrono::Local)
6668
.every("1 * * * * * *")
6769
.description("A simple task")
68-
.add_step_default(|| {
70+
.add_step("Step 1", || {
6971
info!("Hello from step 1");
70-
Ok(()) // Let the scheduler know this step was a success.
72+
Ok(Success) // Let the scheduler know this step was a success.
7173
})
72-
.add_step_default(move || {
74+
.add_step("Step 2", move || {
7375
if exec_count % 2 == 0 {
74-
error!("Oh no this step failed!");
7576
exec_count += 1;
76-
Err(()) // Indicate that this step was a fail.
77+
Err(Error(Some("Oh no this task failed".into()))) // Indicate that this step was a fail.
7778
} else {
7879
info!("Hello from step 2");
7980
exec_count += 1;
80-
Ok(()) // Indicate that this step was a success.
81+
Ok(Success) // Indicate that this step was a success.
8182
}
8283
})
8384
.build(),

‎examples/generator_example.rs

+3-2
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
use chrono::Utc;
22
use log::info;
33
use simple_logger::SimpleLogger;
4+
use tasklet::task::TaskStepStatusOk::Success;
45
use tasklet::{TaskBuilder, TaskGenerator, TaskScheduler};
56

67
/// This examples shows how to use a (not so usefully) `TaskGenerator`
@@ -26,11 +27,11 @@ async fn main() {
2627
.repeat(2)
2728
.add_step_default(|| {
2829
info!("[Step 1] This is a generated task!");
29-
Ok(())
30+
Ok(Success)
3031
})
3132
.add_step_default(|| {
3233
info!("[Step 2] This is generated task!");
33-
Ok(())
34+
Ok(Success)
3435
})
3536
.build(),
3637
)

‎examples/one_task_example.rs

+6-5
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
1-
use log::{error, info};
1+
use log::info;
22
use simple_logger::SimpleLogger;
3+
use tasklet::task::TaskStepStatusErr::Error;
4+
use tasklet::task::TaskStepStatusOk::Success;
35
use tasklet::{TaskBuilder, TaskScheduler};
46

57
/// A simple example of a task with two steps,
@@ -24,17 +26,16 @@ async fn main() {
2426
.description("A simple task")
2527
.add_step_default(|| {
2628
info!("Hello from step 1");
27-
Ok(()) // Let the scheduler know this step was a success.
29+
Ok(Success) // Let the scheduler know this step was a success.
2830
})
2931
.add_step_default(move || {
3032
if exec_count % 2 == 0 {
31-
error!("Oh no this step failed!");
3233
exec_count += 1;
33-
return Err(()); // Indicate that this step was a fail.
34+
return Err(Error(Some("Oh no this step failed".into()))); // Indicate that this step was a fail.
3435
}
3536
info!("Hello from step 2");
3637
exec_count += 1;
37-
Ok(()) // Indicate that this step was a success.
38+
Ok(Success) // Indicate that this step was a success.
3839
})
3940
.build(),
4041
);

‎examples/read_me_example.rs

+8-7
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,11 @@
1-
use log::{error, info};
1+
use log::info;
22
use simple_logger::SimpleLogger;
3+
use tasklet::task::TaskStepStatusErr::Error;
4+
use tasklet::task::TaskStepStatusOk::Success;
35
use tasklet::{TaskBuilder, TaskScheduler};
46

5-
/// A simple example of a task with two step,
6-
/// that might work or fail some times.
7+
/// A simple example of a task with two steps,
8+
/// that might work or fail sometimes.
79
#[tokio::main]
810
async fn main() {
911
// Init the logger.
@@ -24,17 +26,16 @@ async fn main() {
2426
.description("A simple task")
2527
.add_step("Step 1", || {
2628
info!("Hello from step 1");
27-
Ok(()) // Let the scheduler know this step was a success.
29+
Ok(Success) // Let the scheduler know this step was a success.
2830
})
2931
.add_step("Step 2", move || {
3032
if exec_count % 2 == 0 {
31-
error!("Oh no this step failed!");
3233
exec_count += 1;
33-
Err(()) // Indicate that this step was a fail.
34+
Err(Error(Some("Oh no this task failed".into()))) // Indicate that this step was a fail.
3435
} else {
3536
info!("Hello from step 2");
3637
exec_count += 1;
37-
Ok(()) // Indicate that this step was a success.
38+
Ok(Success) // Indicate that this step was a success.
3839
}
3940
})
4041
.build(),

‎examples/simple_example.rs

+3-2
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
use log::info;
22
use simple_logger::SimpleLogger;
3+
use tasklet::task::TaskStepStatusOk::Success;
34
use tasklet::{TaskBuilder, TaskScheduler};
45

56
/// An example of a `TaskScheduler` instance with 2 `Task` instances.
@@ -27,7 +28,7 @@ async fn main() {
2728
.add_step_default(move || {
2829
count = count - 1;
2930
info!("I have {} more executions left!", count);
30-
Ok(())
31+
Ok(Success)
3132
})
3233
.build(),
3334
)
@@ -37,7 +38,7 @@ async fn main() {
3738
.description("Just another task")
3839
.add_step_default(|| {
3940
info!("I will run forever!");
40-
Ok(())
41+
Ok(Success)
4142
})
4243
.build(),
4344
);

‎examples/task_builder_example.rs

+3-2
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
use chrono::Utc;
22
use simple_logger::SimpleLogger;
3+
use tasklet::task::TaskStepStatusOk::Success;
34
use tasklet::{TaskBuilder, TaskScheduler};
45

56
/// An example of a `TaskScheduler` instance with one`Task` instance
@@ -18,8 +19,8 @@ async fn main() {
1819
.every("* * * * * *")
1920
.description("Some description")
2021
.repeat(5)
21-
.add_step("First step", || Ok(()))
22-
.add_step("Second step", || Ok(()))
22+
.add_step("First step", || Ok(Success))
23+
.add_step("Second step", || Ok(Success))
2324
.build(),
2425
);
2526

‎src/builders.rs

+12-9
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
use crate::task::{Task, TaskStep};
1+
use crate::task::{Task, TaskStep, TaskStepStatusErr, TaskStepStatusOk};
22
use chrono::TimeZone;
33
use cron::Schedule;
44

@@ -107,12 +107,13 @@ where
107107
/// # Examples
108108
///
109109
/// ```rust
110+
/// # use tasklet::task::TaskStepStatusErr::Error;
110111
/// # use tasklet::TaskBuilder;
111-
/// let _ = TaskBuilder::new(chrono::Utc).add_step("A step that fails.", || Err(()));
112+
/// let _ = TaskBuilder::new(chrono::Utc).add_step("A step that fails.", || Err(Error(None)));
112113
/// ```
113114
pub fn add_step<F>(mut self, description: &str, function: F) -> TaskBuilder<T>
114115
where
115-
F: (FnMut() -> Result<(), ()>) + Send + 'static,
116+
F: (FnMut() -> Result<TaskStepStatusOk, TaskStepStatusErr>) + Send + 'static,
116117
{
117118
self.steps.push(TaskStep::new(description, function));
118119
self
@@ -125,12 +126,13 @@ where
125126
/// * function - The executable body of the task's step.
126127
///
127128
/// ```
128-
/// # use tasklet::TaskBuilder;
129-
/// let _ = TaskBuilder::new(chrono::Local).add_step_default(|| Ok(()));
129+
/// # use tasklet::task::TaskStepStatusOk::Success;
130+
/// use tasklet::TaskBuilder;
131+
/// let _ = TaskBuilder::new(chrono::Local).add_step_default(|| Ok(Success));
130132
/// ```
131133
pub fn add_step_default<F>(mut self, function: F) -> TaskBuilder<T>
132134
where
133-
F: (FnMut() -> Result<(), ()>) + 'static + Send,
135+
F: (FnMut() -> Result<TaskStepStatusOk, TaskStepStatusErr>) + 'static + Send,
134136
{
135137
self.steps.push(TaskStep::default(function));
136138
self
@@ -167,6 +169,7 @@ where
167169
#[cfg(test)]
168170
mod test {
169171
use super::*;
172+
use crate::task::TaskStepStatusOk::Success;
170173

171174
/// Test helper macros.
172175
///
@@ -232,7 +235,7 @@ mod test {
232235
/// Test the normal functionality of the add_step() function of the `TaskBuilder`.
233236
#[test]
234237
pub fn test_task_builder_add_step() {
235-
let builder = TaskBuilder::new(chrono::Utc).add_step_default(|| Ok(()));
238+
let builder = TaskBuilder::new(chrono::Utc).add_step_default(|| Ok(Success));
236239
assert_none!(builder.schedule, builder.repeats, builder.description);
237240
assert_eq!(builder.timezone, chrono::Utc);
238241
assert_eq!(builder.steps.len(), 1);
@@ -245,7 +248,7 @@ mod test {
245248
.every("* * * * * * *")
246249
.repeat(5)
247250
.description("Some description")
248-
.add_step("Step 1", || Ok(()))
251+
.add_step("Step 1", || Ok(Success))
249252
.build();
250253
assert_some!(task.repeats);
251254
assert_eq!(task.description, "Some description");
@@ -258,7 +261,7 @@ mod test {
258261
pub fn test_task_builder_build_default() {
259262
let task = TaskBuilder::new(chrono::Utc)
260263
.repeat(5)
261-
.add_step("Step 1", || Ok(()))
264+
.add_step("Step 1", || Ok(Success))
262265
.build();
263266
assert_some!(task.repeats);
264267
assert_eq!(task.timezone, chrono::Utc);

‎src/scheduler.rs

+35-34
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,9 @@ use crate::task::{run_task, Status, Task, TaskCmd, TaskResponse};
33
use chrono::prelude::*;
44
use chrono::Utc;
55
use futures::future::join_all;
6+
use futures::StreamExt;
67
use log::{debug, error, info};
8+
use std::sync::{Arc, Mutex};
79
use std::time::Duration;
810
use tokio::sync::{mpsc, oneshot};
911
use tokio::task::JoinHandle;
@@ -160,28 +162,27 @@ where
160162
let mut receivers: Vec<oneshot::Receiver<TaskResponse>> = Vec::new();
161163

162164
for handle in &self.handles {
163-
let (send, recv) = oneshot::channel();
164-
let msg = TaskCmd::Run { sender: send };
165-
let _ = handle.sender.send(msg).await;
165+
let (sender, recv) = oneshot::channel();
166+
let _ = handle.sender.send(TaskCmd::Run { sender }).await;
166167
receivers.push(recv);
167168
}
168169

169-
let mut err_no: usize = 0;
170-
let mut total_runs: usize = 0;
171-
for recv in receivers {
172-
let task_response = recv.await;
173-
let task_response = task_response.unwrap();
174-
match task_response.status {
175-
Status::Executed => {
176-
total_runs += 1;
177-
}
178-
Status::Failed => {
179-
err_no += 1;
180-
total_runs += 1;
181-
}
182-
_ => {}
183-
}
184-
}
170+
let err_no: Arc<Mutex<usize>> = Arc::new(Mutex::new(0usize));
171+
let total_runs: Arc<Mutex<usize>> = Arc::new(Mutex::new(0usize));
172+
futures::stream::iter(receivers)
173+
.for_each(|r| async {
174+
match r.await.unwrap().status {
175+
Status::Executed => {
176+
*total_runs.lock().unwrap() += 1;
177+
}
178+
Status::Failed => {
179+
*err_no.lock().unwrap() += 1;
180+
*total_runs.lock().unwrap() += 1;
181+
}
182+
_ => { /* Do nothing */ }
183+
};
184+
})
185+
.await;
185186

186187
// Send for reschedule
187188
receivers = Vec::new();
@@ -193,26 +194,25 @@ where
193194
}
194195

195196
for recv in receivers {
196-
let res = recv.await;
197-
let res1 = res.unwrap();
198-
if res1.status == Status::Finished {
197+
let res = recv.await.unwrap();
198+
if res.status == Status::Finished {
199199
for handle in &self.handles {
200-
if handle.id == res1.id {
201-
debug!("Killing task {} due to end of execution circle.", res1.id);
200+
if handle.id == res.id {
201+
debug!("Killing task {} due to end of execution circle.", res.id);
202202
handle.handle.abort();
203203
}
204204
}
205-
let index = self.handles.iter().position(|x| x.id == res1.id).unwrap();
205+
let index = self.handles.iter().position(|x| x.id == res.id).unwrap();
206206
self.handles.remove(index);
207207
}
208208
}
209209

210210
// Build the response
211-
if total_runs > 0 {
212-
if err_no == 0 {
213-
ExecutionStatus::Success(total_runs)
211+
if *total_runs.lock().unwrap() > 0 {
212+
if *err_no.lock().unwrap() == 0 {
213+
ExecutionStatus::Success(*total_runs.lock().unwrap())
214214
} else {
215-
ExecutionStatus::HadError(total_runs, err_no)
215+
ExecutionStatus::HadError(*total_runs.lock().unwrap(), *err_no.lock().unwrap())
216216
}
217217
} else {
218218
ExecutionStatus::NoExecution
@@ -227,9 +227,8 @@ where
227227
// Send init signal to all the tasks that are not initialized yet.
228228
for handle in &self.handles {
229229
if !handle.is_init {
230-
let (send, recv) = oneshot::channel();
231-
let msg = TaskCmd::Init { sender: send };
232-
let _ = handle.sender.send(msg).await;
230+
let (sender, recv) = oneshot::channel();
231+
let _ = handle.sender.send(TaskCmd::Init { sender }).await;
233232
receivers.push(recv);
234233
count += 1;
235234
}
@@ -321,6 +320,8 @@ where
321320
#[cfg(test)]
322321
mod test {
323322
use super::*;
323+
use crate::task::TaskStepStatusErr::Error;
324+
use crate::task::TaskStepStatusOk::Success;
324325
use crate::TaskBuilder;
325326
use chrono::Local;
326327
use std::time::Duration;
@@ -351,9 +352,9 @@ mod test {
351352

352353
// Create a task.
353354
let mut task = Task::new("* * * * * * *", None, Some(1), Local);
354-
task.add_step_default(|| Ok(()));
355+
task.add_step_default(|| Ok(Success));
355356
// Return an error in the second step.
356-
task.add_step_default(|| Err(()));
357+
task.add_step_default(|| Err(Error(None)));
357358

358359
// Add a task.
359360
scheduler.add_task(task);

0 commit comments

Comments
 (0)