//handle events for a topic, will panic if max retry have reached
handle events for a topic, will panic if max retry have reached
// handle events for a topic will panic if max retry have reached
func (d *redisDispatcher) handle(ctx context.Context, tpc Topic, evt *DispatchedEvent, wg *sync.WaitGroup) bool {
	defer wg.Done()
	handlers := d.handlers[tpc]
	if d.isStopped() || nil == evt || evt.EmitAt.After(time.Now()) || len(handlers) == 0 {
		return false
	}
	lockKey := fmt.Sprintf("%s:%s:%s", tpc, "handling", fmt.Sprintf("%x", md5.Sum([]byte(evt.JsonStable()))))
	lockValue := strconv.FormatInt(rand.Int63(), 10)
	if !d.clientWrapper.DLock(ctx, lockKey, lockValue, d.options.MaxTaskExecuteDuration) {
		return false
	}
	defer d.clientWrapper.DUnlock(ctx, lockKey, lockValue)
	handled := false
	for _, handle := range handlers {
		if err1 := handle(ctx, evt.Event); nil != err1 {
			log.Printf("handle event [%+v] failed: %s", evt, err1)
			handled = false
			break
		}
		handled = true
	}
	if err2 := d.clientWrapper.ZRem(ctx, tpc, evt); nil != err2 {
		panic(fmt.Sprintf("remove event [%+v] failed: %s", evt, err2))
	}
	if !handled {
		if evt.Retry < d.options.MaxRetry {
			evt.Retry = evt.Retry + 1
			if err3 := d.clientWrapper.ZAdd(ctx, tpc, evt); nil != err3 {
				panic(fmt.Sprintf("event [%v] handle failed after max retry: %s", evt, err3))
			}
		}
	}
	return handled
}
func ( d * redisDispatcher ) handle ( ctx context . Context , tpc Topic , evt * DispatchedEvent , wg * sync . WaitGroup ) bool { defer wg . Done ( ) handlers := d . handlers [ tpc ] if d . isStopped ( ) || nil == evt || evt . EmitAt . After ( time . Now ( ) ) || len ( handlers ) == 0 { return false } lockKey := fmt . Sprintf ( " " , tpc , " " , fmt . Sprintf ( " " , md5 . Sum ( [ ] byte ( evt . JsonStable ( ) ) ) ) ) lockValue := strconv . FormatInt ( rand . Int63 ( ) , 10 ) if ! d . clientWrapper . DLock ( ctx , lockKey , lockValue , d . options . MaxTaskExecuteDuration ) { return false } defer d . clientWrapper . DUnlock ( ctx , lockKey , lockValue ) handled := false for _ , handle := range handlers { if err1 := handle ( ctx , evt . Event ) ; nil != err1 { log . Printf ( " " , evt , err1 ) handled = false break } handled = true } if err2 := d . clientWrapper . ZRem ( ctx , tpc , evt ) ; nil != err2 { panic ( fmt . Sprintf ( " " , evt , err2 ) ) } if ! handled { if evt . Retry < d . options . MaxRetry { evt . Retry = evt . Retry + 1 if err3 := d . clientWrapper . ZAdd ( ctx , tpc , evt ) ; nil != err3 { panic ( fmt . Sprintf ( " " , evt , err3 ) ) } } } return handled }
handle events for a topic, will panic if max retry have reached
handle events for a topic will panic if max retry have reached
//if the low score does not need handle, skip all the events until next ticker //lock if there is a handler is handling //remove the event first //put event back to retry if lower than max retry, !!!may be have concurrency problem