OrgGo/eventials.goevents
1
0
Fork 0
mirror of https://github.com/eventials/goevents.git synced 2025-04-26 13:48:59 +08:00
Code Issues Projects Releases Wiki Activity

fix delay

Browse Source
This commit is contained in:
Guilherme 2017-10-10 18:55:26 -03:00
parent bfd7ed11f2
commit 030c26dba8
1 changed files with 11 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
amqp/consumer.go
View File

@ -173,7 +173,12 @@ func (c *consumer) dispatch(msg amqplib.Delivery) {
delay, isRetry := getXRetryDelayHeader(msg)
if isRetry {
<-time.After(delay)
logger.WithFields(log.Fields{
"delay": delay.String(),
"message_id": msg.MessageId,
}).Info("Delaying message.")
time.Sleep(delay)
}
retryCount, _ := getXRetryCountHeader(msg)
@ -278,10 +283,8 @@ func (c *consumer) publishMessage(msg amqplib.Publishing, queue string) error {
}
func (c *consumer) retryMessage(msg amqplib.Delivery, h *handler, retryCount int32, delay time.Duration) {
delayNs := delay.Nanoseconds()
if h.delayedRetry {
delayNs *= 2
delay *= 2
}
retryMsg := amqplib.Publishing{
@ -289,7 +292,7 @@ func (c *consumer) retryMessage(msg amqplib.Delivery, h *handler, retryCount int
"x-retry-death": time.Now().UTC(),
"x-retry-count": retryCount + 1,
"x-retry-max": h.maxRetries,
"x-retry-delay": delayNs,
"x-retry-delay": delay.String(),
"x-action-key": getAction(msg),
},
Timestamp: time.Now(),
@ -479,10 +482,9 @@ func getXRetryCountHeader(msg amqplib.Delivery) (int32, bool) {
func getXRetryDelayHeader(msg amqplib.Delivery) (time.Duration, bool) {
if d, ok := msg.Headers["x-retry-delay"]; ok {
f, ok := d.(int64)
if ok {
return time.Duration(f), true
t, err := time.ParseDuration(d.(string))
if err != nil {
return t, true
}
}