Method Get

pkg/queue/queue.go:55–85

Get retrieves and removes the next non-cancelled task from the queue, adding it to the running tasks. Blocks until a task is available or the queue is closed.

()

// Get retrieves and removes the next non-cancelled task from the queue, adding it to the running tasks.
// Blocks until a task is available or the queue is closed.
func (tq *TaskQueue[T]) Get() (*Task[T], error) {
	tq.mu.Lock()
	defer tq.mu.Unlock()

	for tq.tasks.Len() == 0 && !tq.closed {
		tq.cond.Wait()
	}

	if tq.closed && tq.tasks.Len() == 0 {
		return nil, fmt.Errorf("queue is closed and empty")
	}

	for tq.tasks.Len() > 0 {
		element := tq.tasks.Front()
		task := element.Value.(*Task[T])

		tq.tasks.Remove(element)
		task.element = nil

		if !task.Cancelled() {
			tq.runningTaskMap[task.ID] = task
			return task, nil
		}
	}

	if !tq.closed {
		return tq.Get()
	}

	return nil, fmt.Errorf("queue is closed and empty")
}

// Done stops(cancels) and removes the task from the running tasks.
func (tq *TaskQueue[T]) Done(taskID string) {