feat(worker): support XReadGroup read data

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

Author: appleboy

_example/server-client/client/main.go

@@ -34,6 +34,7 @@ func main() {
 	w := redisdb.NewWorker(
 		redisdb.WithAddr("127.0.0.1:6379"),
 		redisdb.WithStreamName("foobar"),
+		redisdb.WithGroup("foobar"),
 		redisdb.WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
 			var v *job
 			if err := json.Unmarshal(m.Bytes(), &v); err != nil {

_example/server-client/server/main.go

@@ -29,6 +29,7 @@ func main() {
 	w := redisdb.NewWorker(
 		redisdb.WithAddr("127.0.0.1:6379"),
 		redisdb.WithStreamName("foobar"),
+		redisdb.WithDisableConsumer(),
 	)
 
 	// define the queue
@@ -48,7 +49,7 @@ func main() {
 		}(i)
 	}
 
-	time.Sleep(1 * time.Second)
+	time.Sleep(2 * time.Second)
 	// shutdown the service and notify all the worker
 	q.Release()
 }

options.go

@@ -19,6 +19,9 @@ type options struct {
 	password         string
 	streamName       string
 	cluster          bool
+	group            string
+	consumer         string
+	disableConsumer  bool
 }
 
 // WithAddr setup the addr of redis
@@ -49,6 +52,27 @@ func WithStreamName(name string) Option {
 	}
 }
 
+// WithGroup group name
+func WithGroup(name string) Option {
+	return func(w *options) {
+		w.group = name
+	}
+}
+
+// WithConsumer consumer name
+func WithConsumer(name string) Option {
+	return func(w *options) {
+		w.consumer = name
+	}
+}
+
+// WithDisableConsumer disable consumer
+func WithDisableConsumer() Option {
+	return func(w *options) {
+		w.disableConsumer = true
+	}
+}
+
 // WithPassword redis password
 func WithPassword(passwd string) Option {
 	return func(w *options) {
@@ -80,7 +104,9 @@ func WithLogger(l queue.Logger) Option {
 func newOptions(opts ...Option) options {
 	defaultOpts := options{
 		addr:       "127.0.0.1:6379",
-		streamName: "queue",
+		streamName: "golang-queue",
+		group:      "golang-queue",
+		consumer:   "golang-queue",
 		logger:     queue.NewLogger(),
 		runFunc: func(context.Context, core.QueuedMessage) error {
 			return nil

redis.go

@@ -64,7 +64,20 @@ func NewWorker(opts ...Option) *Worker {
 		w.opts.logger.Fatal(err)
 	}
 
-	go w.fetchTask()
+	if !w.opts.disableConsumer {
+		err = w.rdb.XGroupCreateMkStream(
+			context.Background(),
+			w.opts.streamName,
+			w.opts.group,
+			"$",
+		).Err()
+
+		if err != nil {
+			w.opts.logger.Fatal(err)
+		}
+
+		go w.fetchTask()
+	}
 
 	return w
 }
@@ -78,8 +91,10 @@ func (w *Worker) fetchTask() {
 		}
 
 		ctx := context.Background()
-		data, err := w.rdb.XRead(ctx, &redis.XReadArgs{
-			Streams: []string{w.opts.streamName, "0"},
+		data, err := w.rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
+			Group:    w.opts.group,
+			Consumer: w.opts.consumer,
+			Streams:  []string{w.opts.streamName, ">"},
 			// count is number of entries we want to read from redis
 			Count: 1,
 			// we use the block command to make sure if no entry is found we wait
@@ -95,12 +110,12 @@ func (w *Worker) fetchTask() {
 			for _, message := range result.Messages {
 				select {
 				case w.tasks <- message:
-					// delete message
-					result := w.rdb.XDel(ctx, w.opts.streamName, message.ID)
-					if result.Val() != 1 {
-						w.opts.logger.Errorf("can't delete message: %s", message.ID)
+					if err := w.rdb.XAck(ctx, w.opts.streamName, w.opts.group, message.ID).Err(); err != nil {
+						w.opts.logger.Errorf("can't ack message: %s", message.ID)
 					}
 				case <-w.stop:
+					// Todo: re-queue the task
+					w.opts.logger.Info("re-queue the task: ", message.ID)
 					return
 				}
 			}