| 27 | } |
| 28 | |
| 29 | func (tq *TaskQueue[T]) Add(task *Task[T]) error { |
| 30 | tq.mu.Lock() |
| 31 | defer tq.mu.Unlock() |
| 32 | |
| 33 | if tq.closed { |
| 34 | return errors.New("queue is closed") |
| 35 | } |
| 36 | |
| 37 | if _, exists := tq.taskMap[task.ID]; exists { |
| 38 | return fmt.Errorf("task with ID %s already exists", task.ID) |
| 39 | } |
| 40 | |
| 41 | if task.Cancelled() { |
| 42 | return fmt.Errorf("task %s has been cancelled", task.ID) |
| 43 | } |
| 44 | |
| 45 | element := tq.tasks.PushBack(task) |
| 46 | task.element = element |
| 47 | tq.taskMap[task.ID] = task |
| 48 | |
| 49 | tq.cond.Signal() |
| 50 | return nil |
| 51 | } |
| 52 | |
| 53 | // Get retrieves and removes the next non-cancelled task from the queue, adding it to the running tasks. |
| 54 | // Blocks until a task is available or the queue is closed. |