chore(redis): support block time in XReadGroup

ref: #15

Signed-off-by: Bo-Yi.Wu <[email protected]>

Author: appleboy

options.go

@@ -2,6 +2,7 @@ package redisdb
 
 import (
 	"context"
+	"time"
 
 	"github.com/golang-queue/queue"
 	"github.com/golang-queue/queue/core"
@@ -22,6 +23,7 @@ type options struct {
 	group            string
 	consumer         string
 	maxLength        int64
+	blockTime        time.Duration
 }
 
 // WithAddr setup the addr of redis
@@ -38,6 +40,15 @@ func WithMaxLength(m int64) Option {
 	}
 }
 
+// WithBlockTime setup the block time for publish messages
+// we use the block command to make sure if no entry is found we wait
+// until an entry is found
+func WithBlockTime(m time.Duration) Option {
+	return func(w *options) {
+		w.blockTime = m
+	}
+}
+
 // WithPassword redis password
 func WithDB(db int) Option {
 	return func(w *options) {
@@ -111,6 +122,7 @@ func newOptions(opts ...Option) options {
 		runFunc: func(context.Context, core.QueuedMessage) error {
 			return nil
 		},
+		blockTime: 60 * time.Second,
 	}
 
 	// Loop through each option

redis.go

@@ -16,8 +16,6 @@ import (
 
 var _ core.Worker = (*Worker)(nil)
 
-const blockTime = 60000
-
 // Worker for Redis
 type Worker struct {
 	// redis config
@@ -103,10 +101,10 @@ func (w *Worker) fetchTask() {
 			Count: 1,
 			// we use the block command to make sure if no entry is found we wait
 			// until an entry is found
-			Block: blockTime,
+			Block: w.opts.blockTime,
 		}).Result()
 		if err != nil {
-			w.opts.logger.Errorf("error while reading from redis %v",err)
+			w.opts.logger.Errorf("error while reading from redis %v", err)
 			continue
 		}
 		// we have received the data we should loop it and queue the messages