From 95023bd3b5af5c00bc52c18f33e42e40aaf8a89e Mon Sep 17 00:00:00 2001
From: Ken Hibino <ken.hibino7@gmail.com>
Date: Sat, 16 Nov 2019 07:51:53 -0800
Subject: [PATCH] Extract a push logic to a function

---
 asynq.go | 32 ++++++++++++++++----------------
 1 file changed, 16 insertions(+), 16 deletions(-)

diff --git a/asynq.go b/asynq.go
index 387bac5..0c5cfe7 100644
--- a/asynq.go
+++ b/asynq.go
@@ -51,11 +51,7 @@ func NewClient(opt *RedisOpt) *Client {
 // Enqueue pushes a given task to the specified queue.
 func (c *Client) Enqueue(queue string, task *Task, executeAt time.Time) error {
 	if time.Now().After(executeAt) {
-		bytes, err := json.Marshal(task)
-		if err != nil {
-			return err
-		}
-		return c.rdb.RPush(queuePrefix+queue, string(bytes)).Err()
+		return push(c.rdb, queue, task)
 	}
 	dt := &delayedTask{
 		ID:    uuid.New().String(),
@@ -156,18 +152,9 @@ func (w *Workers) pollScheduledTasks() {
 			}
 
 			if w.rdb.ZRem(scheduled, j).Val() > 0 {
-				// Do we need to encode this again?
-				// Can we skip this entirely by defining Task field to be a string field?
-				bytes, err := json.Marshal(job.Task)
+				err = push(w.rdb, job.Queue, job.Task)
 				if err != nil {
-					log.Printf("could not marshal job.Task %v: %v\n", job.Task, err)
-					// TODO(hibiken): Put item that cannot marshal into dead queue.
-					continue
-				}
-				err = w.rdb.RPush(queuePrefix+job.Queue, string(bytes)).Err()
-				if err != nil {
-					log.Printf("command RPUSH %q %s failed: %v",
-						queuePrefix+job.Queue, string(bytes), err)
+					log.Printf("could not push task to queue %q: %v", job.Queue, err)
 					// TODO(hibiken): Handle this error properly. Add back to scheduled ZSET?
 					continue
 				}
@@ -175,3 +162,16 @@ func (w *Workers) pollScheduledTasks() {
 		}
 	}
 }
+
+func push(rdb *redis.Client, queue string, t *Task) error {
+	bytes, err := json.Marshal(t)
+	if err != nil {
+		return fmt.Errorf("could not encode task into JSON: %v", err)
+	}
+	err = rdb.SAdd(allQueues, queue).Err()
+	if err != nil {
+		return fmt.Errorf("could not execute command SADD %q %q: %v",
+			allQueues, queue, err)
+	}
+	return rdb.RPush(queuePrefix+queue, string(bytes)).Err()
+}