actix-taskqueue 0.1.0

A generic task queue service for Actix
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126

# Actix Task Queue

Actix Task Queue is a generic task queue service. You have many different task queue for many
different pair of Task and Task Result in your application as possible.

A task queue for a pair of Task and Task Result will be a [SystemService](https://docs.rs/actix/latest/actix/registry/trait.SystemService.html).

![](resources/overview.png)

To use the queue, you first need to define a data structure for your Task and Task Result:

For example:
```
#[derive(Debug, Default, Clone, Copy)]
struct Task(i32);
struct TaskResult(i32);
```

Next, you need to implement the `QueueConsumer` trait for your pair of Task and Task Result:

```
#[async_trait]
impl QueueConsumer<Task, TaskResult> for TaskWorker<Task, TaskResult> {
    async fn execute(&self, task: Task) -> Result<TaskResult, WorkerExecuteError> {
      ...
    }

    fn get_queue(&self) -> Addr<TaskQueue<Task>> {
      ...
    }

    fn retry(&self, task: Task) -> Task {
      ...
    }

    fn drop(&self, task: Task) {
      ...
    }

    fn result(&self, result: TaskResult) {
      ...
    }
}
```

When you need to run the next available task in the queue, call the `next()` method of
your task worker.

```
worker.next().await
```

This method will return a `Result<TaskResult, WorkerExecuteError>`.

---

The following example describe a task queue of `i32` numbers, when executing, the worker will:

- If a number is larger or equal to 5, add 5 to that number.
- If a number is larger than 0 but smaller than 5, add 1 to that number and send it back to the queue to retry.
- If a number is smaller than 0, drop that number from the queue.

```
#[derive(Debug, Default, Clone, Copy)]
struct PlusFive(i32);
struct PlusFiveResult(i32);

#[async_trait]
impl QueueConsumer<PlusFive, PlusFiveResult> for TaskWorker<PlusFive, PlusFiveResult> {
    async fn execute(&self, task: PlusFive) -> Result<PlusFiveResult, WorkerExecuteError> {
        let PlusFive(n) = task;
        if n >= 5 {
            return Ok(PlusFiveResult(n + 5));
        } else if n > 0 {
            return Err(WorkerExecuteError::Retryable);
        } else {
            return Err(WorkerExecuteError::NonRetryable);
        }
    }

    fn get_queue(&self) -> Addr<TaskQueue<PlusFive>> {
        TaskQueue::<PlusFive>::from_registry()
    }

    fn retry(&self, task: PlusFive) -> PlusFive {
        let PlusFive(n) = task;
        println!("RETRYING VALUE = {}", n);
        PlusFive(n + 1)
    }

    fn drop(&self, task: PlusFive) {
        let PlusFive(n) = task;
        println!("DROPPED TASK WITH VALUE = {}", n);
    }

    fn result(&self, result: PlusFiveResult) {
        let PlusFiveResult(n) = result;
        println!("RESULT = {}", n);
    }
}

#[actix_rt::main]
async fn main() {
    let queue = TaskQueue::<PlusFive>::from_registry();
    let worker = TaskWorker::<PlusFive, PlusFiveResult>::new();

    queue.do_send(Push::new(PlusFive(5)));
    queue.do_send(Push::new(PlusFive(8)));
    queue.do_send(Push::new(PlusFive(3)));
    queue.do_send(Push::new(PlusFive(11)));
    queue.do_send(Push::new(PlusFive(0)));
    queue.do_send(Push::new(PlusFive(20)));

    worker.next().await;
    worker.next().await;
    worker.next().await;
    worker.next().await;
    worker.next().await;
    worker.next().await;
    worker.next().await;
    worker.next().await;
    worker.next().await;
    worker.next().await;
}

```