✨ add transaction handler

Author: yuanyu90221

README.md

@@ -139,4 +139,72 @@ func main() {
 	}
 }
 
+```
+
+## redis with transaction operation
+
+```golang
+package main
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"os"
+
+	"github.com/leetcode-golang-classroom/golang-sample-with-redis/internal/config"
+	"github.com/leetcode-golang-classroom/golang-sample-with-redis/internal/logger"
+	myredis "github.com/leetcode-golang-classroom/golang-sample-with-redis/internal/redis"
+	"github.com/leetcode-golang-classroom/golang-sample-with-redis/internal/util"
+	"github.com/redis/go-redis/v9"
+)
+
+func main() {
+	jsonLogger := slog.New(slog.NewJSONHandler(
+		os.Stdout, &slog.HandlerOptions{
+			AddSource: true,
+		},
+	))
+	ctx := logger.CtxWithLogger(context.Background(), jsonLogger)
+	config.Init(ctx)
+	redisURL := config.AppCfg.RedisUrl
+
+	rdb, err := myredis.New(redisURL)
+	if err != nil {
+		util.FailOnError(ctx, err, fmt.Sprintf("failed to connect to %s\n", redisURL))
+	}
+	defer rdb.Close()
+	_, err = rdb.Ping(ctx)
+	if err != nil {
+		util.FailOnError(ctx, err, fmt.Sprintf("failed to ping to %s\n", redisURL))
+	}
+	// try 10 times
+	for i := 0; i < 10; i++ {
+		err = rdb.Watch(ctx, func(tx *redis.Tx) error {
+			pipe := tx.Pipeline()
+			err = pipe.IncrBy(ctx, "p1", 100).Err()
+			if err != nil {
+				return err
+			}
+			err = pipe.DecrBy(ctx, "p0", 100).Err()
+			if err != nil {
+				return err
+			}
+			_, err = pipe.Exec(ctx)
+			return err
+		}, "p0")
+
+		if err == nil {
+			jsonLogger.Info("transaction execution success")
+			break
+		} else if err == redis.TxFailedErr {
+			jsonLogger.Info("transaction execution failed", slog.Int("i", i))
+			continue
+		} else {
+			util.FailOnError(ctx, err, "failed")
+		}
+	}
+
+}
+
 ```

cmd/pipelined-sample/main.go

@@ -0,0 +1,62 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"os"
+
+	"github.com/leetcode-golang-classroom/golang-sample-with-redis/internal/config"
+	"github.com/leetcode-golang-classroom/golang-sample-with-redis/internal/logger"
+	myredis "github.com/leetcode-golang-classroom/golang-sample-with-redis/internal/redis"
+	"github.com/leetcode-golang-classroom/golang-sample-with-redis/internal/util"
+	"github.com/redis/go-redis/v9"
+)
+
+func main() {
+	jsonLogger := slog.New(slog.NewJSONHandler(
+		os.Stdout, &slog.HandlerOptions{
+			AddSource: true,
+		},
+	))
+	ctx := logger.CtxWithLogger(context.Background(), jsonLogger)
+	config.Init(ctx)
+	redisURL := config.AppCfg.RedisUrl
+
+	rdb, err := myredis.New(redisURL)
+	if err != nil {
+		util.FailOnError(ctx, err, fmt.Sprintf("failed to connect to %s\n", redisURL))
+	}
+	defer rdb.Close()
+	_, err = rdb.Ping(ctx)
+	if err != nil {
+		util.FailOnError(ctx, err, fmt.Sprintf("failed to ping to %s\n", redisURL))
+	}
+	// try 10 times
+	for i := 0; i < 10; i++ {
+		err = rdb.Watch(ctx, func(tx *redis.Tx) error {
+			pipe := tx.Pipeline()
+			err = pipe.IncrBy(ctx, "p1", 100).Err()
+			if err != nil {
+				return err
+			}
+			err = pipe.DecrBy(ctx, "p0", 100).Err()
+			if err != nil {
+				return err
+			}
+			_, err = pipe.Exec(ctx)
+			return err
+		}, "p0")
+
+		if err == nil {
+			jsonLogger.Info("transaction execution success")
+			break
+		} else if err == redis.TxFailedErr {
+			jsonLogger.Info("transaction execution failed", slog.Int("i", i))
+			continue
+		} else {
+			util.FailOnError(ctx, err, "failed")
+		}
+	}
+
+}

internal/redis/redis.go

@@ -37,7 +37,12 @@ func (rh *RedisHandler) Get(ctx context.Context, key string) (string, error) {
 	return result, err
 }
 
-// Pipelined -
+// Watch
+func (rh *RedisHandler) Watch(ctx context.Context, fn func(*redis.Tx) error, keys ...string) error {
+	return rh.client.Watch(ctx, fn, keys...)
+}
+
+// Pipelined - handle pipeline with transaction
 func (rh *RedisHandler) Pipelined(ctx context.Context, handler func(pipe redis.Pipeliner) error) ([]redis.Cmder, error) {
 	return rh.client.Pipelined(ctx, handler)
 }

mage-tools/magefile.go

@@ -29,7 +29,11 @@ func Build() error {
 	if err != nil {
 		return err
 	}
-	return sh.Run("go", "build", "-o", "./bin/pipeline-sample", "./cmd/pipeline-sample/main.go")
+	err = sh.Run("go", "build", "-o", "./bin/pipeline-sample", "./cmd/pipeline-sample/main.go")
+	if err != nil {
+		return err
+	}
+	return sh.Run("go", "build", "-o", "./bin/pipelined-sample", "./cmd/pipelined-sample/main.go")
 }
 
 // start the basic-sample
@@ -62,6 +66,16 @@ func LaunchPipelineSample() error {
 	return nil
 }
 
+// start the pipelined-sample
+func LaunchPipelinedSample() error {
+	mg.Deps(Build)
+	err := sh.RunV("./bin/pipelined-sample")
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
 // run the test
 func Test() error {
 	err := sh.RunV("go", "test", "-v", "./...")