//emit Start consuming events of a topic, go-routines count will be restricted with parallelThreshold
// Ends until no event can emit by current time
emit Start consuming events of a topic, go-routines count will be restricted with parallelThreshold
Ends until no event can emit by current time
// emit Start consuming events of a topic go - routines count will be restricted with parallelThreshold // Ends until no event can emit by current time
func (d *localDispatcher) emit(ctx context.Context, events *localEvents, wg *sync.WaitGroup) {
	if nil != wg {
		defer wg.Done()
	}
	if d.isStopped() || nil == events {
		return
	}
	n := time.Now()
	e := events.PopEvent(ctx)
	for {
		if nil == e || e.EmitAt.After(n) {
			events.PushEvent(ctx, e)
			return
		}
		go func(de *DispatchedEvent) {
			events.parallel <- true
			defer func() {
				<-events.parallel
			}()
			handled := d.handle(ctx, de, events.handlers)
			if !handled {
				if de.Retry < d.options.MaxRetry {
					de.Retry = atomic.NewInt32(de.Retry).Inc()
					events.PushEvent(ctx, de)
					return
				}
				panic(fmt.Sprintf("event [%v] handle failed after max retry", de))
			}
			d.total.Dec()
		}(e)
		e = events.PopEvent(ctx)
	}
}
func ( d * localDispatcher ) emit ( ctx context . Context , events * localEvents , wg * sync . WaitGroup ) { if nil != wg { defer wg . Done ( ) } if d . isStopped ( ) || nil == events { return } n := time . Now ( ) e := events . PopEvent ( ctx ) for { if nil == e || e . EmitAt . After ( n ) { events . PushEvent ( ctx , e ) return } go func ( de * DispatchedEvent ) { events . parallel <- true defer func ( ) { <- events . parallel } ( ) handled := d . handle ( ctx , de , events . handlers ) if ! handled { if de . Retry < d . options . MaxRetry { de . Retry = atomic . NewInt32 ( de . Retry ) . Inc ( ) events . PushEvent ( ctx , de ) return } panic ( fmt . Sprintf ( " " , de ) ) } d . total . Dec ( ) } ( e ) e = events . PopEvent ( ctx ) } }
emit Start consuming events of a topic, go-routines count will be restricted with parallelThreshold
Ends until no event can emit by current time
emit Start consuming events of a topic go - routines count will be restricted with parallelThreshold Ends until no event can emit by current time
//quit if no event or the first event should emit after now //start consume with go-routines // return the heap if max retry not reached //will panic current go-routine