feat(sentinel): Add support for Redis Sentinel (#29)

Author: nttu-ysc

options.go

@@ -20,6 +20,8 @@ type options struct {
 	channelName      string
 	channelSize      int
 	cluster          bool
+	sentinel         bool
+	masterName       string
 }
 
 // WithAddr setup the addr of redis
@@ -43,6 +45,20 @@ func WithCluster(enable bool) Option {
 	}
 }
 
+// WithSentinel redis sentinel
+func WithSentinel(enable bool) Option {
+	return func(w *options) {
+		w.sentinel = enable
+	}
+}
+
+// WithMasterName sentinel master name
+func WithMasterName(masterName string) Option {
+	return func(w *options) {
+		w.masterName = masterName
+	}
+}
+
 // WithChannelSize redis channel size
 func WithChannelSize(size int) Option {
 	return func(w *options) {

redis.go

@@ -49,6 +49,13 @@ func NewWorker(opts ...Option) *Worker {
 				Addrs:    strings.Split(w.opts.addr, ","),
 				Password: w.opts.password,
 			})
+		} else if w.opts.sentinel {
+			w.rdb = redis.NewFailoverClient(&redis.FailoverOptions{
+				MasterName:    w.opts.masterName,
+				SentinelAddrs: strings.Split(w.opts.addr, ","),
+				Password:      w.opts.password,
+				DB:            w.opts.db,
+			})
 		} else {
 			options := &redis.Options{
 				Addr:     w.opts.addr,

redis_test.go

@@ -130,6 +130,37 @@ func TestRedisCluster(t *testing.T) {
 	// you will see the execute time > 1000ms
 }
 
+func TestRedisSentinel(t *testing.T) {
+	t.Helper()
+	m := &mockMessage{
+		Message: "foo",
+	}
+	hosts := []string{host + ":26379", host + ":26380"}
+
+	w := NewWorker(
+		WithAddr(strings.Join(hosts, ",")),
+		WithMasterName("mymaster"),
+		WithChannel("testSentinel"),
+		WithSentinel(true),
+		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
+			time.Sleep(500 * time.Millisecond)
+			return nil
+		}),
+	)
+	q := queue.NewPool(
+		5,
+		queue.WithWorker(w),
+	)
+	time.Sleep(100 * time.Millisecond)
+	assert.NoError(t, q.Queue(m))
+	assert.NoError(t, q.Queue(m))
+	assert.NoError(t, q.Queue(m))
+	assert.NoError(t, q.Queue(m))
+	time.Sleep(1000 * time.Millisecond)
+	q.Release()
+	// you will see the execute time > 1000ms
+}
+
 func TestEnqueueJobAfterShutdown(t *testing.T) {
 	m := mockMessage{
 		Message: "foo",