waitForTask returns a task that can be processed or a nil if context is canceled. sleepDelay is used to sleep before check a task, e.g. in case of failed previous task. If queue is empty, then it will be checked every DelayOnQueueIsEmpty.
(sleepDelay time.Duration)
| 940 | // sleepDelay is used to sleep before check a task, e.g. in case of failed previous task. |
| 941 | // If queue is empty, then it will be checked every DelayOnQueueIsEmpty. |
| 942 | func (q *TaskQueue) waitForTask(sleepDelay time.Duration) task.Task { |
| 943 | // Check Done channel. |
| 944 | select { |
| 945 | case <-q.ctx.Done(): |
| 946 | return nil |
| 947 | default: |
| 948 | } |
| 949 | |
| 950 | // Shortcut: return the first task if the queue is not empty and delay is not required. |
| 951 | if q.Length() > 0 && sleepDelay == 0 { |
| 952 | return q.GetFirst() |
| 953 | } |
| 954 | |
| 955 | // Initialize wait settings. |
| 956 | waitBegin := time.Now() |
| 957 | waitUntil := q.DelayOnQueueIsEmpty |
| 958 | if sleepDelay != 0 { |
| 959 | waitUntil = sleepDelay |
| 960 | } |
| 961 | |
| 962 | checkTicker := time.NewTicker(q.WaitLoopCheckInterval) |
| 963 | q.waitInProgress.Store(true) |
| 964 | q.cancelDelay.Store(false) |
| 965 | |
| 966 | // Snapshot original status |
| 967 | origStatusType, origStatusText := q.status.Snapshot() |
| 968 | |
| 969 | defer func() { |
| 970 | checkTicker.Stop() |
| 971 | q.waitInProgress.Store(false) |
| 972 | q.cancelDelay.Store(false) |
| 973 | // Restore original status |
| 974 | q.status.Restore(origStatusType, origStatusText) |
| 975 | }() |
| 976 | |
| 977 | // Wait for the queued task with some delay. |
| 978 | // Every tick increases the 'elapsed' counter until it outgrows the waitUntil value. |
| 979 | // Or, delay can be canceled to handle new head task immediately. |
| 980 | for { |
| 981 | checkTask := false |
| 982 | select { |
| 983 | case <-q.ctx.Done(): |
| 984 | // Queue is stopped. |
| 985 | return nil |
| 986 | case <-checkTicker.C: |
| 987 | // Check and update waitUntil. |
| 988 | elapsed := time.Since(waitBegin) |
| 989 | |
| 990 | if q.cancelDelay.Load() { |
| 991 | // Reset waitUntil to check task immediately. |
| 992 | waitUntil = elapsed |
| 993 | } |
| 994 | |
| 995 | // Wait loop is done or canceled: break select to check for the head task. |
| 996 | if elapsed >= waitUntil { |
| 997 | // Increase waitUntil to wait on the next iteration and go check for the head task. |
| 998 | checkTask = true |
| 999 | } |