Method Dequeue

internal/queue/timequeue.go:89–143

(ctx context.Context)

}

func (t *TimeQueue[T]) Dequeue(ctx context.Context) T {
	t.dequeueMu.Lock()
	defer t.dequeueMu.Unlock()

	notify := make(chan struct{}, 1)
	t.notify.Store(&notify)
	defer t.notify.Store(nil)

	for {
		t.mu.Lock()
		var wait time.Duration
		if t.heap.Len() > 0 {
			val := t.heap.Peek()
			wait = time.Until(val.at)
			if wait <= 0 {
				defer t.mu.Unlock()
				return heap.Pop(&t.heap).(timeQueueEntry[T]).v
			}
		}
		if wait == 0 || wait > 3*time.Minute {
			wait = 3 * time.Minute
		}
		t.mu.Unlock()

		timer := time.NewTimer(wait)

		select {
		case <-timer.C:
			t.mu.Lock()
			if len(t.heap) == 0 {
				t.mu.Unlock()
				continue
			}
			val := t.heap.Peek()
			if val.at.After(time.Now()) {
				t.mu.Unlock()
				continue
			}
			heap.Pop(&t.heap)
			t.mu.Unlock()
			return val.v
		case <-notify: // new task was added, loop again to ensure we have the earliest task.
			if !timer.Stop() {
				<-timer.C
			}
			continue
		case <-ctx.Done():
			if !timer.Stop() {
				<-timer.C
			}
			var zero T
			return zero
		}
	}
}

type timeQueueEntry[T any] struct {
	at time.Time