Function sendEvent

events/memory.go:199–237

(ev *Event, sub *subscriber)

}

func sendEvent(ev *Event, sub *subscriber) {
	go func(s *subscriber) {
		evCopy := *ev
		if s.autoAck {
			s.Channel <- evCopy
			return
		}
		evCopy.SetAckFunc(ackFunc(s, evCopy))
		evCopy.SetNackFunc(nackFunc(s, evCopy))
		s.Lock()
		s.retryMap[evCopy.ID] = 0
		s.Unlock()
		tick := time.NewTicker(s.ackWait)
		defer tick.Stop()
		for range tick.C {
			s.Lock()
			count, ok := s.retryMap[evCopy.ID]
			s.Unlock()
			if !ok {
				// success
				break
			}

			if s.retryLimit > -1 && count > s.retryLimit {
				if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
					logger.Errorf("Message retry limit reached, discarding: %v %d %d", evCopy.ID, count, s.retryLimit)
				}
				s.Lock()
				delete(s.retryMap, evCopy.ID)
				s.Unlock()
				return
			}
			s.Channel <- evCopy
			s.Lock()
			s.retryMap[evCopy.ID] = count + 1
			s.Unlock()
		}
	}(sub)
}

func ackFunc(s *subscriber, evCopy Event) func() error {
	return func() error {