// Start routine to generate timeout data
Start routine to generate timeout data
// Start routine to generate timeout data
func (q *TimeoutQueue) Start(stop <-chan struct{}) {
	if !atomic.CompareAndSwapUint32(&q.running, 0, 1) {
		return
	}
	q.mu.Lock()
	q.stop = stop
	q.mu.Unlock()
	wg := new(sync.WaitGroup)
	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			select {
			case <-stop:
				return
			case <-q.hasExpiredData:
				for _, d := range q.expiredData {
					data := d
					select {
					case <-stop:
						return
					case q.timeoutCh <- data:
					}
				}
				q.expiredData = nil
				q.hasExpiredData = make(chan struct{})
				atomic.StoreUint32(&q.expireNotified, 0)
			}
		}
	}()
	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			select {
			case <-stop:
				if !q.timer.Stop() {
					<-q.timer.C
				}
				return
			case newWait := <-q.resetCh:
				if !q.timer.Stop() {
					if atomic.LoadUint32(&q.timeGot) == 0 {
						<-q.timer.C
						atomic.StoreUint32(&q.timeGot, 1)
						q.sort()
					}
				}
				q.timer.Reset(time.Duration(newWait))
				atomic.StoreUint32(&q.timeGot, 0)
			case <-q.timer.C:
				atomic.StoreUint32(&q.timeGot, 1)
				q.sort()
			}
		}
	}()
	go func() {
		wg.Wait()
		atomic.StoreUint32(&q.running, 0)
	}()
}
func ( q * TimeoutQueue ) Start ( stop <- chan struct { } ) { if ! atomic . CompareAndSwapUint32 ( & q . running , 0 , 1 ) { return } q . mu . Lock ( ) q . stop = stop q . mu . Unlock ( ) wg := new ( sync . WaitGroup ) wg . Add ( 1 ) go func ( ) { defer wg . Done ( ) for { select { case <- stop : return case <- q . hasExpiredData : for _ , d := range q . expiredData { data := d select { case <- stop : return case q . timeoutCh <- data : } } q . expiredData = nil q . hasExpiredData = make ( chan struct { } ) atomic . StoreUint32 ( & q . expireNotified , 0 ) } } } ( ) wg . Add ( 1 ) go func ( ) { defer wg . Done ( ) for { select { case <- stop : if ! q . timer . Stop ( ) { <- q . timer . C } return case newWait := <- q . resetCh : if ! q . timer . Stop ( ) { if atomic . LoadUint32 ( & q . timeGot ) == 0 { <- q . timer . C atomic . StoreUint32 ( & q . timeGot , 1 ) q . sort ( ) } } q . timer . Reset ( time . Duration ( newWait ) ) atomic . StoreUint32 ( & q . timeGot , 0 ) case <- q . timer . C : atomic . StoreUint32 ( & q . timeGot , 1 ) q . sort ( ) } } } ( ) go func ( ) { wg . Wait ( ) atomic . StoreUint32 ( & q . running , 0 ) } ( ) }
Start routine to generate timeout data
Start routine to generate timeout data
// already running and not stopped // handle delivery of expired data // handle timer reset // timer already fired // no sort performed, sort now // reset timer to new values