Programming►Utility

TaskQueue

A very simple but powerful utility class I used all over. It gives us a chance to do things in paralle and still can wait for the result to come.

Code

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
/**
 * Promise based TaskQueue
 */
type OnTaskDone = (queue: TaskQueue) => void;
type Task = () => Promise<any>;

class TaskQueue {
  concurrent = 1;
  tasks: Array<Task> = [];
  errors: Array<Error> = [];
  runningTask = 0;
  name = "";
  onTaskDone: OnTaskDone | null;
  log: Function;

  constructor(concurrent = 1, onTaskDone: OnTaskDone, name: string, showLog = false) {
    this.concurrent = concurrent;
    this.tasks = [];
    this.errors = [];
    this.runningTask = 0;
    this.name = name || "TaskQueue";
    this.onTaskDone = onTaskDone;

    if (showLog) {
      this.log = console.log;
    } else {
      this.log = () => {};
    }
  }

  addTask(task: Task) {
    this.tasks.push(task);
  }

  reset() {
    this.errors.length = 0;
    this.tasks.length = 0;
    this.runningTask = 0;
  }

  status() {
    if (this.tasks.length == 0 && this.errors.length == 0 && this.runningTask == 0) {
      return "finished";
    } else if (this.errors.length > 0) {
      return "failed";
    } else if (this.runningTask > 0 || this.tasks.length > 0) {
      return "running";
    }
  }

  /**
   * Task will be executed by the sequence added in, but not garantee that previous
   * job is finished.
   */
  async execute() {
    if (this.status() == "finished" && this.onTaskDone) {
      this.onTaskDone(this);
      this.onTaskDone = null;
    }

    while (this.runningTask < this.concurrent && this.tasks.length > 0) {
      let task = this.tasks.shift();

      this.runningTask++;
      this.log(this.name, "remain", this.tasks.length, "running", this.runningTask);

      try {
        await task!();
      } catch (error) {
        this.errors.push(error as Error);
      } finally {
        this.runningTask--;
        this.execute();

        this.log(this.name, "remain", this.tasks.length, "running", this.runningTask);
        if (this.runningTask == 0 && this.tasks.length == 0 && this.onTaskDone) {
          this.onTaskDone(this);
        }
      }
    }
  }
}
export default TaskQueue;