Method WaitStopWithTimeout

pkg/task/queue/queue_set.go:216–244

(timeout time.Duration)

}

func (tqs *TaskQueueSet) WaitStopWithTimeout(timeout time.Duration) {
	checkTick := time.NewTicker(time.Millisecond * 100)
	defer checkTick.Stop()

	timeoutTick := time.NewTicker(timeout)
	defer timeoutTick.Stop()

	for {
		select {
		case <-checkTick.C:
			stopped := func() bool {
				for _, q := range tqs.Queues.List() {
					if q.GetStatusType() != QueueStatusStop {
						return false
					}
				}

				return true
			}()

			if stopped {
				return
			}

		case <-timeoutTick.C:
			return
		}
	}
}

// Shutdown stops every queue in the set and blocks until either all workers
// reach QueueStatusStop or ctx is canceled / deadlines. Returns ctx.Err() on