Requaos/unique jobs

.gitignore

@@ -1 +1,10 @@
-vendor/
+vendor/
+
+# GoLand/IntelliJ
+.idea
+
+# VSCode workspace settings
+.vscode
+
+# macOS
+*.DS_Store

fetcher.go

@@ -82,17 +82,16 @@ func (f *simpleFetcher) Fetch() {
 
 func (f *simpleFetcher) tryFetchMessage() {
 	message, err := f.client.BRPopLPush(f.queue, f.inprogressQueue(), 1*time.Second).Result()
-
-	if err != nil {
-		// If redis returns null, the queue is empty.
-		// Just ignore empty queue errors; print all other errors.
-		if err != redis.Nil {
-			Logger.Println("ERR: ", f.queue, err)
-		}
-
-		time.Sleep(1 * time.Second)
-	} else {
+	switch err {
+	case nil:
 		f.sendMessage(message)
+	case redis.Nil:
+		// If redis returns null, the queue is empty.
+		// Just ignore empty queue errors.
+		time.Sleep(1*time.Second)
+	default:
+		// Print all other errors.
+		Logger.Println("ERR: ", f.queue, err)
 	}
 }
 
@@ -108,7 +107,12 @@ func (f *simpleFetcher) sendMessage(message string) {
 }
 
 func (f *simpleFetcher) Acknowledge(message *Msg) {
-	f.client.LRem(f.inprogressQueue(), -1, message.OriginalJson()).Result()
+	count := int64(-1)
+	val, err := message.Get("unique").Bool()
+	if err == nil && val {
+		count = 0
+	}
+	f.client.LRem(f.inprogressQueue(), count, message.OriginalJson()).Result()
 }
 
 func (f *simpleFetcher) Messages() chan *Msg {

middleware_retry.go

@@ -1,6 +1,8 @@
 package workers
 
 import (
+	"crypto/sha1"
+	"encoding/hex"
 	"fmt"
 	"math"
 	"math/rand"
@@ -58,8 +60,15 @@ func RetryMiddleware(queue string, mgr *Manager, next JobFunc) JobFunc {
 
 		}()
 
-		err = next(message)
-		if err != nil {
+		switch next(message) {
+		case nil:
+			val, err := message.Get("unique").Bool()
+			if err == nil && val {
+				rc := mgr.opts.client
+				sum := sha1.Sum([]byte(message.Args().ToJson()))
+				rc.Del(hex.EncodeToString(sum[:])).Result()
+			}
+		default:
 			err = retryProcessError(queue, mgr, message, err)
 		}
 

producer.go

@@ -2,6 +2,8 @@ package workers
 
 import (
 	"crypto/rand"
+	"crypto/sha1"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -31,6 +33,7 @@ type EnqueueOptions struct {
 	RetryCount int     `json:"retry_count,omitempty"`
 	Retry      bool    `json:"retry,omitempty"`
 	At         float64 `json:"at,omitempty"`
+	Unique     bool    `json:"unique,omitempty"`
 }
 
 func NewProducer(options Options) (*Producer, error) {
@@ -69,7 +72,25 @@ func (p *Producer) EnqueueAt(queue, class string, at time.Time, args interface{}
 	return p.EnqueueWithOptions(queue, class, args, EnqueueOptions{At: timeToSecondsWithNanoPrecision(at)})
 }
 
+func (p *Producer) EnqueueUnique(queue, class string, at time.Time, args interface{}) (string, error) {
+	return p.EnqueueWithOptions(queue, class, args, EnqueueOptions{At: nowToSecondsWithNanoPrecision(), Unique: true})
+}
+
 func (p *Producer) EnqueueWithOptions(queue, class string, args interface{}, opts EnqueueOptions) (string, error) {
+	rc := p.opts.client
+
+	if opts.Unique {
+		bytes, err := json.Marshal(args)
+		if err != nil {
+			return "", err
+		}
+		sum := sha1.Sum(bytes)
+		if !rc.SetNX(hex.EncodeToString(sum[:]), "unique", 0).Val() {
+			// already in the list
+			return "", nil
+		}
+	}
+
 	now := nowToSecondsWithNanoPrecision()
 	data := EnqueueData{
 		Queue:          queue,
@@ -90,7 +111,6 @@ func (p *Producer) EnqueueWithOptions(queue, class string, args interface{}, opt
 		return data.Jid, err
 	}
 
-	rc := p.opts.client
 
 	_, err = rc.SAdd(p.opts.Namespace+"queues", queue).Result()
 	if err != nil {