//start ticker running go-routine to check events, check if
start ticker running go-routine to check events, check if
// start ticker running go - routine to check events check if
func (d *redisDispatcher) start(ctx context.Context) {
	ticker := time.NewTicker(d.options.TickerInterval)
	d.state = atomic.NewInt32(stateRedisRunning)
	go func() {
		for {
			select {
			case <-ticker.C:
				if d.isStopped() {
					return
				}
				if d.isHandling() || len(d.handlers) == 0 {
					break
				}
				d.state.Store(stateRedisHandling)
				wg := &sync.WaitGroup{}
				wg.Add(len(d.handlers))
				for tpc, handlers := range d.handlers {
					go d.emit(ctx, tpc, handlers, wg)
				}
				wg.Wait()
				d.state.Store(stateRedisRunning)
			case <-ctx.Done():
				log.Printf("context done")
				return
			}
		}
	}()
}
func ( d * redisDispatcher ) start ( ctx context . Context ) { ticker := time . NewTicker ( d . options . TickerInterval ) d . state = atomic . NewInt32 ( stateRedisRunning ) go func ( ) { for { select { case <- ticker . C : if d . isStopped ( ) { return } if d . isHandling ( ) || len ( d . handlers ) == 0 { break } d . state . Store ( stateRedisHandling ) wg := & sync . WaitGroup { } wg . Add ( len ( d . handlers ) ) for tpc , handlers := range d . handlers { go d . emit ( ctx , tpc , handlers , wg ) } wg . Wait ( ) d . state . Store ( stateRedisRunning ) case <- ctx . Done ( ) : log . Printf ( " " ) return } } } ( ) }
start ticker running go-routine to check events, check if
start ticker running go - routine to check events check if
//if redisDispatcher has been stopped, shut down go-routine //if ticker is still handling or no handlers, break select