Fix issue with delayed tasks for Redis brokers (#34)

* Reset delay on signature for Redis brokers

* Address PR comments

Author: pricejc-sublime

v1/brokers/redis/goredis.go

@@ -174,15 +174,17 @@ func (b *BrokerGR) Publish(ctx context.Context, signature *tasks.Signature) erro
 	// Adjust routing key (this decides which queue the message will be published to)
 	b.Broker.AdjustRoutingKey(signature)
 
+	// We'll capture the delay here and set it to zero in the signature because otherwise
+	// the task will be continously delayed.
+	delay := signature.Delay
+	signature.Delay = 0
 	msg, err := json.Marshal(signature)
 	if err != nil {
 		return fmt.Errorf("JSON marshal error: %s", err)
 	}
 
-	// Check the delay signature field, if it is set and it is in the future,
-	// delay the task
-	if signature.Delay > 0 {
-		score := time.Now().Add(signature.Delay).UnixNano()
+	if delay > 0 {
+		score := time.Now().Add(delay).UnixNano()
 		err = b.rclient.ZAdd(context.Background(), b.redisDelayedTasksKey, redis.Z{Score: float64(score), Member: msg}).Err()
 		return err
 	}

v1/brokers/redis/redis.go

@@ -184,6 +184,10 @@ func (b *Broker) Publish(ctx context.Context, signature *tasks.Signature) error
 	// Adjust routing key (this decides which queue the message will be published to)
 	b.Broker.AdjustRoutingKey(signature)
 
+	// We'll capture the delay here and set it to zero in the signature because otherwise
+	// the task will be continously delayed.
+	delay := signature.Delay
+	signature.Delay = 0
 	msg, err := json.Marshal(signature)
 	if err != nil {
 		return fmt.Errorf("JSON marshal error: %s", err)
@@ -192,10 +196,8 @@ func (b *Broker) Publish(ctx context.Context, signature *tasks.Signature) error
 	conn := b.open()
 	defer conn.Close()
 
-	// Check the delay signature field, if it is set and it is in the future,
-	// delay the task
-	if signature.Delay > 0 {
-		score := time.Now().Add(signature.Delay).UnixNano()
+	if delay > 0 {
+		score := time.Now().Add(delay).UnixNano()
 		_, err = conn.Do("ZADD", b.redisDelayedTasksKey, score, msg)
 		return err
 	}