base version

Author: sergii4

Makefile

@@ -0,0 +1,20 @@
+export GO111MODULE=on
+
+LINT_VERSION := v1.16.0
+HAS_LINT := $(shell command -v golangci-lint 2> /dev/null)
+
+test-redis-queue:
+	docker-compose up --build  --abort-on-container-exit
+	docker-compose down 
+
+lint:
+ifndef HAS_LINT
+	curl -sfL https://install.goreleaser.com/github.com/golangci/golangci-lint.sh | sh -s -- -b $(GOPATH)/bin $(LINT_VERSION)
+endif
+	golangci-lint run \
+		--disable-all=true \
+		--enable=errcheck \
+		./...
+
+format:
+	go fmt ./...

README.md

@@ -0,0 +1,71 @@
+# Redis delayed queue
+
+Based on redis doc: [delayed tasks](https://redislabs.com/ebook/part-2-core-concepts/chapter-6-application-components-in-redis/6-4-task-queues/6-4-2-delayed-tasks/), [distributed locks](https://redislabs.com/ebook/part-2-core-concepts/chapter-6-application-components-in-redis/6-2-distributed-locking/)
+
+## Start
+
+For start we use redis cluster instance and go code adds/removes to/from [ZSET](https://redis.io/commands#sorted_set) and SETNX for [distributed locks](https://redislabs.com/ebook/part-2-core-concepts/chapter-6-application-components-in-redis/6-2-distributed-locking/)
+
+### Docker redis cluster instance
+
+[Docker Hub](https://hub.docker.com/r/grokzen/redis-cluster)
+[Github](https://github.com/Grokzen/docker-redis-cluster)
+
+### Go code 
+
+Libs
+
+- [Go redis client](https://github.com/go-redis/redis)
+- [Go redis distributed lock](https://github.com/bsm/redislock)
+
+## Implementation
+
+Normally when we talk about times in Redis, we usually start talking about ZSETs. What if, for any item we wanted to execute in the future, we added it to a ZSET  with its score being the time when we want it to execute? Then we have to check for items that should be executed now.
+
+Enqueue:
+```
+func (rq *RedisQueue) Enqueue(uuid string, delay time.Duration) {
+	_ = rq.client.ZAdd(rq.key, &redis.Z{Member: uuid, Score: float64(time.Now().Add(delay).Unix())})
+}
+```
+Dequeue:
+```
+func (rq *RedisQueue) Dequeue() ([]Message, error) {
+	var ms []Message
+	start := int64(0)
+	for i := rq.batch; i >= 0; {
+		vals, err := rq.client.ZRangeWithScores(rq.key, start, start).Result()
+		if err != nil {
+			return nil, errors.Wrap(err, "cannot get range from zset")
+		}
+		if len(vals) == 0 || vals[0].Score > float64(time.Now().Unix()) {
+			break
+		}
+
+		id := vals[0].Member.(string)
+		lock := rq.acquireLock(id)
+		if lock == nil {
+			start++
+			continue
+		}
+		ms = append(ms, Message{Message: id, OnProcessed: func() {
+			_ = rq.client.ZRem(rq.key, id)
+			if err := lock.Release(); err != nil {
+				fmt.Printf("release lock erros = %+v\n", err)
+			}
+		}})
+		start++
+		i--
+
+	}
+
+	return ms, nil
+}
+
+```
+
+## Test
+To run test:
+```
+make test-redis-queue 
+```

docker-compose.yml

@@ -0,0 +1,19 @@
+version: "2.1"
+services:
+  tests:
+    image: golang:1.12
+    working_dir: /go/src/github.com/sergii4/redis-delayed-queue
+    volumes:
+      - $PWD:/go/src/github.com/sergii4/redis-delayed-queue
+      - go-modules:/go/pkg/mod # Put modules cache into a separate volume
+    depends_on:
+      - testredis
+    command: ["/bin/sh", "-c", "GO111MODULE=on go test -v -timeout 30s"]
+
+  testredis:
+    image: grokzen/redis-cluster:latest
+    logging:
+      driver: "none"
+
+volumes:
+  go-modules: # Define the volume

go.mod

@@ -0,0 +1,11 @@
+module github.com/sergii4/redis-delayed-queue
+
+go 1.12
+
+require (
+	github.com/bsm/redislock v0.5.0
+	github.com/go-redis/redis/v7 v7.2.0
+	github.com/pkg/errors v0.9.1
+	github.com/satori/go.uuid v1.2.0
+	github.com/stretchr/testify v1.5.1
+)

go.sum

@@ -0,0 +1,57 @@
+github.com/bsm/redislock v0.5.0 h1:ODM11/cbuUXQqLgZWK6XQnufaTjsBE2UcwBc2EAFNDA=
+github.com/bsm/redislock v0.5.0/go.mod h1:qagqKlV+xiLy26iV34Y3zRPxRcJjQYbV7pZfWFeSZ8M=
+github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I=
+github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
+github.com/go-redis/redis/v7 v7.2.0 h1:CrCexy/jYWZjW0AyVoHlcJUeZN19VWlbepTh1Vq6dJs=
+github.com/go-redis/redis/v7 v7.2.0/go.mod h1:JDNMw23GTyLNC4GZu9njt15ctBQVn7xjRfnwdHj/Dcg=
+github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs=
+github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
+github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
+github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
+github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/ginkgo v1.10.1 h1:q/mM8GF/n0shIN8SaAZ0V+jnLPzen6WIVZdiwrRlMlo=
+github.com/onsi/ginkgo v1.10.1/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/gomega v1.7.0 h1:XPnZz8VVBHjVsy1vzJmRwIcSwiUO+JFfrv/xGiigmME=
+github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww=
+github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.5.1 h1:nOGnQDM7FYENwehXlg/kFVnos3rEvtKTjRvOWSzb6H4=
+github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20190923162816-aa69164e4478 h1:l5EDrHhldLYb3ZRHDUhXF7Om7MvYXnkV9/iQNo1lX6g=
+golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f h1:wMNYb4v58l5UBM7MYRLPG6ZhfOqbKu7X5eyFl8ZhKvA=
+golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20191010194322-b09406accb47 h1:/XfQ9z7ib8eEJX2hdgFTZJ/ntt0swNk5oYBziWeTCvY=
+golang.org/x/sys v0.0.0-20191010194322-b09406accb47/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
+golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4=
+gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
+gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
+gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
+gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.4 h1:/eiJrUcujPVeJ3xlSWaiNi3uSVmDGBK1pDHUHAnao1I=
+gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=

queue.go

@@ -0,0 +1,81 @@
+package redisqueue
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/bsm/redislock"
+	"github.com/go-redis/redis/v7"
+	"github.com/pkg/errors"
+)
+
+type Message struct {
+	Message     string
+	OnProcessed func()
+}
+
+type Queue interface {
+	Enqueue(message string, after time.Duration)
+	Dequeue() ([]Message, error)
+}
+
+// RedisQueue implements Redis Delayed tasks algorith: https://redislabs.com/ebook/part-2-core-concepts/chapter-6-application-components-in-redis/6-4-task-queues/6-4-2-delayed-tasks/
+type RedisQueue struct {
+	client *redis.ClusterClient
+	locker *redislock.Client
+	key    string
+	batch  int
+	ttl    time.Duration
+}
+
+func NewQueue(client *redis.ClusterClient, locker *redislock.Client, key string, batch int, ttl time.Duration) Queue {
+	return &RedisQueue{client: client, locker: locker, key: key, batch: batch, ttl: ttl}
+}
+
+// Enqueue puts task/item with uuid to the queue/container with delay
+func (rq *RedisQueue) Enqueue(uuid string, delay time.Duration) {
+	_ = rq.client.ZAdd(rq.key, &redis.Z{Member: uuid, Score: float64(time.Now().Add(delay).Unix())})
+}
+
+// Dequeue receives batch of messages from the queue
+func (rq *RedisQueue) Dequeue() ([]Message, error) {
+	var ms []Message
+	start := int64(0)
+	for i := rq.batch; i >= 0; {
+		vals, err := rq.client.ZRangeWithScores(rq.key, start, start).Result()
+		if err != nil {
+			return nil, errors.Wrap(err, "cannot get range from zset")
+		}
+		if len(vals) == 0 || vals[0].Score > float64(time.Now().Unix()) {
+			break
+		}
+
+		id := vals[0].Member.(string)
+		lock := rq.acquireLock(id)
+		if lock == nil {
+			start++
+			continue
+		}
+		ms = append(ms, Message{Message: id, OnProcessed: func() {
+			_ = rq.client.ZRem(rq.key, id)
+			if err := lock.Release(); err != nil {
+				fmt.Printf("release lock erros = %+v\n", err)
+			}
+		}})
+		start++
+		i--
+
+	}
+
+	return ms, nil
+
+}
+
+// acquireLock gets the lock for the item with key
+func (rq *RedisQueue) acquireLock(key string) *redislock.Lock {
+	lock, err := rq.locker.Obtain(key, 1000*time.Millisecond, nil)
+	if err != nil {
+		return nil
+	}
+	return lock
+}

queue_test.go

@@ -0,0 +1,158 @@
+package redisqueue
+
+import (
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/bsm/redislock"
+	"github.com/go-redis/redis/v7"
+	uuid "github.com/satori/go.uuid"
+	"github.com/stretchr/testify/require"
+)
+
+func TestQueueLock(t *testing.T) {
+	t.Run("same number enqueued/dequeued, 0 delayed task left in queue", func(t *testing.T) {
+		batch := 20
+		numItems := 100
+		ttl := 100 * time.Millisecond
+		q := setUpQueue(t, uuid.NewV4().String(), batch, ttl)
+		// arange
+		for i := 0; i < numItems; i++ {
+			q.Enqueue("item"+uuid.NewV4().String(), 0)
+		}
+		ch := make(chan Message, numItems)
+		var wg sync.WaitGroup
+		numOfWorkers := 8
+		wg.Add(numOfWorkers)
+		fail := make(chan error)
+		// act
+		// run concurrently the same number goroutines as items
+		go func() {
+			for i := 0; i < numOfWorkers; i++ {
+				// allows goroutins deque enough messages
+				time.Sleep(5 * ttl)
+				go func() {
+					defer wg.Done()
+					ms, err := q.Dequeue()
+					if err != nil {
+						fail <- err
+						return
+					}
+					for _, m := range ms {
+						ch <- m
+					}
+				}()
+			}
+		}()
+		go func() {
+			// close items when every goroutine is executed
+			wg.Wait()
+			close(ch)
+		}()
+
+		count := 0
+		done := make(chan struct{})
+		// read from message channel
+		go func() {
+			for m := range ch {
+				m.OnProcessed()
+				count++
+			}
+			close(done)
+		}()
+		// wait for done, fail or timeout
+		select {
+		case <-done:
+		case err := <-fail:
+			t.Fatal(err)
+		case <-time.After(10 * time.Second):
+			t.Fatal("timeout")
+		}
+		// assert
+		require.Equal(t, numItems, count)
+	})
+	t.Run("first routine processed all messeges last none", func(t *testing.T) {
+		numItems := 5
+		ttl := 100 * time.Millisecond
+		q := setUpQueue(t, uuid.NewV4().String(), numItems, ttl)
+		// arange
+		for i := 0; i < numItems; i++ {
+			q.Enqueue("item"+uuid.NewV4().String(), 0)
+		}
+		for i := 0; i < numItems; i++ {
+			q.Enqueue("item"+uuid.NewV4().String(), 5*time.Second)
+		}
+		ms, err := q.Dequeue()
+		if err != nil {
+			require.NoError(t, err)
+		}
+		require.Len(t, ms, numItems)
+	})
+
+}
+
+func TestQueueuBecameEmpty(t *testing.T) {
+	numItems := 10
+	ttl := 100 * time.Millisecond
+	q := setUpQueue(t, uuid.NewV4().String(), numItems, ttl)
+	// arange
+	for i := 0; i < numItems; i++ {
+		q.Enqueue("item"+uuid.NewV4().String(), 0)
+	}
+
+	ms, err := q.Dequeue()
+	require.NoError(t, err)
+	require.Equal(t, numItems, len(ms))
+	for _, msg := range ms {
+		msg.OnProcessed()
+	}
+
+	ms, err = q.Dequeue()
+	require.NoError(t, err)
+	require.Equal(t, 0, len(ms))
+}
+
+func setUpQueue(t *testing.T, key string, batch int, ttl time.Duration) Queue {
+	client := setRedisClusterClient(t)
+	locker := redislock.New(client)
+
+	return NewQueue(client, locker, key, batch, ttl)
+}
+
+func setRedisClusterClient(t *testing.T) *redis.ClusterClient {
+	client := redis.NewClusterClient(&redis.ClusterOptions{
+		Addrs: []string{"testredis:7000", "testredis:7001"},
+	})
+
+	waitOKState(t, client)
+
+	return client
+}
+
+func waitOKState(t *testing.T, client *redis.ClusterClient) {
+	done := make(chan struct{})
+	fail := make(chan error)
+	go func() {
+		for {
+			info, err := client.ClusterInfo().Result()
+			if err != nil {
+				fail <- err
+				return
+			}
+			if strings.HasPrefix(info, "cluster_state:ok") {
+				// fmt.Println(info)
+				close(done)
+				return
+			}
+		}
+	}()
+	select {
+	case err := <-fail:
+		t.Fatalf("cannot connect to redis instance: %s", err)
+	case <-time.After(20 * time.Second):
+		t.Fatal("cluster state: fail")
+	case <-done:
+	}
+}