//emit start handling events of topic:
// once ticker time will list [0,maxHandlingEvents] of topic, for each events will be handled by a go-routine
emit start handling events of topic:
once ticker time will list [0,maxHandlingEvents] of topic, for each events will be handled by a go-routine
// emit start handling events of topic : // once ticker time will list [ 0 maxHandlingEvents ] of topic for each events will be handled by a go - routine
func (d *redisDispatcher) emit(ctx context.Context, tpc Topic, handlers []HandleFunc, wg *sync.WaitGroup) {
	defer wg.Done()
	if d.isStopped() || len(handlers) == 0 {
		return
	}
	events, err := d.clientWrapper.ZRangeByScore(ctx, tpc, 0, d.options.ParallelThreshold)
	if nil != err {
		panic(fmt.Sprintf("get events from topic [%s] failed: %s", tpc, err))
	}
	if len(events) == 0 {
		return
	}
	wgTopic := &sync.WaitGroup{}
	wgTopic.Add(len(events))
	for _, evt := range events {
		go d.handle(ctx, tpc, evt, wgTopic)
	}
	wgTopic.Wait()
}
func ( d * redisDispatcher ) emit ( ctx context . Context , tpc Topic , handlers [ ] HandleFunc , wg * sync . WaitGroup ) { defer wg . Done ( ) if d . isStopped ( ) || len ( handlers ) == 0 { return } events , err := d . clientWrapper . ZRangeByScore ( ctx , tpc , 0 , d . options . ParallelThreshold ) if nil != err { panic ( fmt . Sprintf ( " " , tpc , err ) ) } if len ( events ) == 0 { return } wgTopic := & sync . WaitGroup { } wgTopic . Add ( len ( events ) ) for _ , evt := range events { go d . handle ( ctx , tpc , evt , wgTopic ) } wgTopic . Wait ( ) }
emit start handling events of topic:
once ticker time will list [0,maxHandlingEvents] of topic, for each events will be handled by a go-routine
emit start handling events of topic : once ticker time will list [ 0 maxHandlingEvents ] of topic for each events will be handled by a go - routine