fix:mastername

go.mod

@@ -1,3 +1,14 @@
 module qoobing.com/gomod/redis
 
-go 1.16
+go 1.18
+
+require (
+	github.com/gomodule/redigo v1.9.2
+	qoobing.com/gomod/log v1.2.8
+	qoobing.com/gomod/str v1.0.5
+)
+
+require (
+	github.com/tylerb/gls v0.0.0-20150407001822-e606233f194d // indirect
+	github.com/tylerb/is v2.1.4+incompatible // indirect
+)

logging/logging.go

@@ -25,7 +25,8 @@ import (
 )
 
 var (
-	_ redis.ConnWithTimeout = (*loggingConn)(nil)
+	_            redis.ConnWithTimeout = (*loggingConn)(nil)
+	logCallDepth                       = 3
 )
 
 // NewLoggingConn returns a logging wrapper around a connection.
@@ -55,7 +56,7 @@ func (c *loggingConn) Close() error {
 	err := c.Conn.Close()
 	var buf bytes.Buffer
 	fmt.Fprintf(&buf, "%sClose() -> (%v)", c.prefix, err)
-	c.logger.Output(2, buf.String()) // nolint: errcheck
+	c.logger.Output(logCallDepth, buf.String()) // nolint: errcheck
 	return err
 }
 
@@ -115,7 +116,7 @@ func (c *loggingConn) print(method, commandName string, args []interface{}, repl
 		buf.WriteString(", ")
 	}
 	fmt.Fprintf(&buf, "%v)", err)
-	c.logger.Output(3, buf.String()) // nolint: errcheck
+	c.logger.Output(logCallDepth+1, buf.String()) // nolint: errcheck
 }
 
 func (c *loggingConn) Do(commandName string, args ...interface{}) (interface{}, error) {

redis.go

@@ -1,45 +1,11 @@
 package redis
 
 import (
-	"fmt"
-	"time"
-
-	"github.com/gomodule/redigo/redis"
+	"qoobing.com/gomod/redis/redis"
 	"qoobing.com/gomod/redis/sentinel"
 )
 
-func newSentinelPool() *redis.Pool {
-	sntnl := &sentinel.Sentinel{
-		Addrs:      []string{":26379", ":26380", ":26381"},
-		MasterName: "mymaster",
-		Dial: func(addr string) (redis.Conn, error) {
-			timeout := 500 * time.Millisecond
-			c, err := redis.DialTimeout("tcp", addr, timeout, timeout, timeout)
-			if err != nil {
-				return nil, err
-			}
-			return c, nil
-		},
-	}
-	return &redis.Pool{
-		MaxIdle:     3,
-		MaxActive:   64,
-		Wait:        true,
-		IdleTimeout: 240 * time.Second,
-		Dial: func() (redis.Conn, error) {
-			masterAddr, err := sntnl.MasterAddr()
-			if err != nil {
-				return nil, err
-			}
-			c, err := redis.Dial("tcp", masterAddr)
-			if err != nil {
-				return nil, err
-			}
-			if !sentinel.TestRole(c, "master") {
-				c.Close()
-				return nil, fmt.Errorf("%s is not redis master", masterAddr)
-			}
-			return c, nil
-		},
-	}
-}
+type Config = sentinel.Config
+
+var NewPool = redis.NewPool
+var NewSentinelPool = sentinel.NewPool

redis/redis.go

@@ -0,0 +1,98 @@
+package redis
+
+import (
+	"fmt"
+	"log"
+	"os"
+	"time"
+
+	"github.com/gomodule/redigo/redis"
+	"qoobing.com/gomod/redis/logging"
+	"qoobing.com/gomod/redis/sentinel"
+)
+
+type Config = sentinel.Config
+
+func NewPool(cfg Config) *redis.Pool {
+	if cfg.Master == "" {
+		panic("config is invalid: Master is empty")
+	} else if cfg.Master != "" && cfg.MasterName != "" {
+		//panic("config is invalid: Master & MasterName must not set both")
+	}
+	var (
+		masterAddr     = cfg.Master
+		masterUsername = cfg.Username
+		masterPassword = cfg.Password
+	)
+	if cfg.Wait == nil {
+		cfg.Wait = new(bool)
+		*cfg.Wait = true
+	}
+
+	if cfg.IdleTimeout == nil {
+		cfg.IdleTimeout = new(int)
+		*cfg.IdleTimeout = 300
+	} else if *cfg.IdleTimeout <= 0 {
+		*cfg.IdleTimeout = 86400 * 365 * 10
+	}
+
+	if cfg.MaxIdle == nil {
+		cfg.MaxIdle = new(int)
+		*cfg.MaxIdle = 100
+	} else if *cfg.MaxIdle < 0 {
+		*cfg.MaxIdle = 100
+	}
+
+	if cfg.MaxActive == nil {
+		cfg.MaxActive = new(int)
+		*cfg.MaxActive = 100
+	} else if *cfg.MaxActive < 0 {
+		*cfg.MaxActive = 100
+	}
+
+	var (
+		logPrefix    = "redis"
+		logStdPrefix = "DBUG "
+		logStdWriter = os.Stdout
+		logStdFlags  = log.Ldate | log.Lmicroseconds | log.Lshortfile
+		logStdLogger = log.New(logStdWriter, logStdPrefix, logStdFlags)
+	)
+	return &redis.Pool{
+		MaxIdle:     *cfg.MaxIdle,
+		MaxActive:   *cfg.MaxActive,
+		Wait:        *cfg.Wait,
+		IdleTimeout: time.Duration(*cfg.IdleTimeout) * time.Second,
+		Dial: func() (redis.Conn, error) {
+			c, err := redis.Dial("tcp", masterAddr)
+			if err != nil {
+				return nil, err
+			}
+
+			var okstr = "OK"
+			if masterPassword != "" && masterUsername != "" {
+				okstr, err = redis.String(c.Do("AUTH", masterUsername, masterPassword))
+			} else if masterPassword != "" {
+				okstr, err = redis.String(c.Do("AUTH", masterPassword))
+			}
+
+			if err != nil {
+				return nil, fmt.Errorf("redis master AUTH failed: <%s>", err.Error())
+			} else if okstr != "OK" {
+				return nil, fmt.Errorf("redis master AUTH failed: <%s>", okstr)
+			}
+
+			//// if !TestRole(c, "master") {
+			//// 	c.Close()
+			//// 	err = fmt.Errorf(
+			//// 		"master(%s) got by name '%s' is not redis master",
+			//// 		masterAddr, masterName)
+			//// 	return nil, err
+			//// }
+
+			if cfg.Debug {
+				c = logging.NewLoggingConn(c, logStdLogger, logPrefix)
+			}
+			return c, nil
+		},
+	}
+}

redis_test.go

@@ -0,0 +1,65 @@
+package redis
+
+import (
+	"fmt"
+	"log"
+	"testing"
+
+	"qoobing.com/gomod/redis/sentinel"
+	"qoobing.com/gomod/redis/xstream"
+)
+
+func TestNewConsumer(t *testing.T) {
+	cfg := xstream.Config{
+		Group:  "group_a",
+		Stream: "testxstream_queue",
+		Sentinel: sentinel.Config{
+			Debug:            true,                     //调试开关（会在日志打印REDIS语句)
+			MasterName:       "walletmaster",           //REDIS主名称（一个哨兵集群可管理多个REDIS主从结构）
+			Username:         "web3wallet",             //REDIS连接用户名
+			Password:         "web3onthemoon@2023#dev", //REDIS连接用户密码
+			Sentinels:        "sentinel-0.redis:5000",  //哨兵节点列表，逗号分隔，一般配置三个
+			SentinelUsername: "sentinel",               //哨兵节点连接用户名，不填默认default
+			SentinelPassword: "password@mhxzkhl#dev",   //哨兵节点连接密码，不填认为集群无密码
+		},
+	}
+
+	for i := 0; i < 15; i++ {
+		i := i
+		name := fmt.Sprintf("name-%d", i)
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+			log.Println("name", name)
+			if r := xstream.NewReader(cfg); r == nil {
+				t.Error("create new reader failed")
+			}
+		})
+	}
+}
+
+func TestNewPool(t *testing.T) {
+	cfg := Config{
+		Debug:            true,                     //调试开关（会在日志打印REDIS语句)
+		Master:           "redis-0.redis:2379",     //REDIS主
+		Username:         "web3wallet",             //REDIS连接用户名
+		Password:         "web3onthemoon@2023#dev", //REDIS连接用户密码
+		Sentinels:        "sentinel-0.redis:5000",  //哨兵节点列表，逗号分隔，一般配置三个
+		SentinelUsername: "sentinel",               //哨兵节点连接用户名，不填默认default
+		SentinelPassword: "password@mhxzkhl#dev",   //哨兵节点连接密码，不填认为集群无密码
+	}
+	pool := NewPool(cfg)
+	if pool == nil {
+		t.Errorf("new pool failed")
+	}
+	rds := pool.Get()
+	if rds == nil {
+		t.Errorf("get redis failed")
+	}
+	if ret, err := rds.Do("SET", "ABC", "123"); err != nil {
+		t.Errorf("get redis failed: %s", err)
+	} else {
+		fmt.Println("ABC", ret)
+	}
+	fmt.Println(rds.Do("GET", "ABC"))
+
+}

sentinel/sentinel.go

@@ -15,15 +15,18 @@ import (
 )
 
 type Config struct {
-	Debug       bool   `toml:"debug"`      //调试开关（会在日志打印REDIS语句)
-	Username    string `toml:"username"`   //REDIS连接用户名
-	Password    string `toml:"password"`   //REDIS连接用户密码
-	MasterName  string `toml:"mastername"` //REDIS主名称（一个哨兵集群可管理多个REDIS主从结构）
-	Sentinels   string `toml:"sentinels"`  //哨兵节点列表，逗号分隔，一般配置三个
-	Wait        *bool  `toml:"wait"`
-	MaxIdle     *int   `toml:"max_idle"`
-	MaxActive   *int   `toml:"max_active"`
-	IdleTimeout *int   `toml:"idle_timeout"`
+	Debug            bool   `toml:"debug"`             //调试开关（会在日志打印REDIS语句)
+	Master           string `toml:"master"`            //REDIS主host&port 与 MasterName 互斥
+	MasterName       string `toml:"mastername"`        //REDIS主名称,值不为空是启动sentinal模式, 与Master互斥
+	Username         string `toml:"username"`          //REDIS连接用户名
+	Password         string `toml:"password"`          //REDIS连接用户密码
+	Sentinels        string `toml:"sentinels"`         //哨兵节点列表，逗号分隔，一般配置三个
+	SentinelUsername string `toml:"sentinel_username"` //哨兵节点连接用户名，不填默认default
+	SentinelPassword string `toml:"sentinel_password"` //哨兵节点连接密码，不填认为集群无密码
+	Wait             *bool  `toml:"wait"`              //当无可用连接时，是否等待
+	MaxIdle          *int   `toml:"max_idle"`          //最大空闲连接数
+	MaxActive        *int   `toml:"max_active"`        //最大活跃连接数
+	IdleTimeout      *int   `toml:"idle_timeout"`      //空闲超时时间
 }
 
 type Sentinel struct {
@@ -51,10 +54,18 @@ type Sentinel struct {
 }
 
 func NewPool(cfg Config) *redis.Pool {
+	if cfg.MasterName == "" {
+		panic("config is invalid: MasterName is empty")
+	} else if cfg.Master != "" && cfg.MasterName != "" {
+		//panic("config is invalid: Master & MasterName must not set both")
+	}
 	var (
-		masterName     = cfg.MasterName
-		masterPassword = cfg.Password
-		sntnl          = &Sentinel{
+		masterName       = cfg.MasterName
+		masterUsername   = cfg.Username
+		masterPassword   = cfg.Password
+		sentinelUsername = cfg.SentinelUsername
+		sentinelPassword = cfg.SentinelPassword
+		sntnl            = &Sentinel{
 			Addrs:      strings.Split(cfg.Sentinels, ","),
 			MasterName: masterName,
 			Dial: func(addr string) (redis.Conn, error) {
@@ -63,6 +74,18 @@ func NewPool(cfg Config) *redis.Pool {
 				if err != nil {
 					return nil, err
 				}
+				var okstr = "OK"
+				if sentinelPassword != "" && sentinelUsername != "" {
+					okstr, err = redis.String(c.Do("AUTH", sentinelUsername, sentinelPassword))
+				} else if sentinelPassword != "" {
+					okstr, err = redis.String(c.Do("AUTH", sentinelPassword))
+				}
+
+				if err != nil {
+					return nil, fmt.Errorf("redis sentinel AUTH failed: <%s>", err.Error())
+				} else if okstr != "OK" {
+					return nil, fmt.Errorf("redis sentinel AUTH failed: <%s>", okstr)
+				}
 				return c, nil
 			},
 		}
@@ -71,7 +94,6 @@ func NewPool(cfg Config) *redis.Pool {
 		cfg.Wait = new(bool)
 		*cfg.Wait = true
 	}
-
 	if cfg.IdleTimeout == nil {
 		cfg.IdleTimeout = new(int)
 		*cfg.IdleTimeout = 300
@@ -93,6 +115,13 @@ func NewPool(cfg Config) *redis.Pool {
 		*cfg.MaxActive = 100
 	}
 
+	var (
+		logPrefix    = "redis"
+		logStdPrefix = "DBUG "
+		logStdWriter = os.Stdout
+		logStdFlags  = log.Ldate | log.Lmicroseconds | log.Lshortfile
+		logStdLogger = log.New(logStdWriter, logStdPrefix, logStdFlags)
+	)
 	return &redis.Pool{
 		MaxIdle:     *cfg.MaxIdle,
 		MaxActive:   *cfg.MaxActive,
@@ -109,7 +138,13 @@ func NewPool(cfg Config) *redis.Pool {
 				return nil, err
 			}
 
-			okstr, err := redis.String(c.Do("AUTH", masterPassword))
+			var okstr = "OK"
+			if masterPassword != "" && masterUsername != "" {
+				okstr, err = redis.String(c.Do("AUTH", masterUsername, masterPassword))
+			} else if masterPassword != "" {
+				okstr, err = redis.String(c.Do("AUTH", masterPassword))
+			}
+
 			if err != nil {
 				return nil, fmt.Errorf("redis master AUTH failed: <%s>", err.Error())
 			} else if okstr != "OK" {
@@ -118,11 +153,14 @@ func NewPool(cfg Config) *redis.Pool {
 
 			if !TestRole(c, "master") {
 				c.Close()
-				return nil, fmt.Errorf("%s is not redis master", masterAddr)
+				err = fmt.Errorf(
+					"master(%s) got by name '%s' is not redis master",
+					masterAddr, masterName)
+				return nil, err
 			}
 
 			if cfg.Debug {
-				c = logging.NewLoggingConn(c, log.New(os.Stdout, "XXXXXXX", log.LstdFlags), "coreredis")
+				c = logging.NewLoggingConn(c, logStdLogger, logPrefix)
 			}
 			return c, nil
 		},

xstream/reader.go

@@ -0,0 +1,372 @@
+package xstream
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"time"
+
+	"github.com/gomodule/redigo/redis"
+	"qoobing.com/gomod/log"
+	"qoobing.com/gomod/redis/sentinel"
+	"qoobing.com/gomod/str"
+)
+
+//////////////////////////////////////////////////////////////////////////////////
+/////////////////////////  example  //////////////////////////////////////////////
+//////////////////////////////////////////////////////////////////////////////////
+//////  const (
+//////  	CORESERVER_NEWACCOUNT_GROUP = "groupid-coreserver-newaccount"
+//////  	CORESERVER_NEWACCOUNT_QUEUE = "coreserver.newaccount"
+//////  )
+//////
+//////  type NewAccountMessage struct {
+//////  	Sign       string    `json:"sign"`
+//////  	Appid      string    `json:"appid"`
+//////  	Userid     int64     `json:"userid"`
+//////  	Account    string    `json:"account"`
+//////  	CreateTime time.Time `json:"createtime"`
+//////  }
+//////
+//////  func Run(ctx context.Context) {
+//////  	var SLEEP_TIME = 10 * time.Second
+//////  	go func() {
+//////  		for {
+//////  			log.Infof("start run newaccount marketing engine ...")
+//////  			runNewAccountMarketingEngine(ctx)
+//////  			log.Infof("newaccount marketing engine exit unexpected, retry %s later again", SLEEP_TIME.String())
+//////  			time.Sleep(SLEEP_TIME)
+//////  		}
+//////  	}()
+//////  }
+//////
+//////  func runNewAccountMarketingEngine(ctx context.Context) {
+//////  	// r read message from redis
+//////  	cfg := config.Instance()
+//////  	r := xstream.NewReader(xstream.Config{
+//////  		Group:    CORESERVER_NEWACCOUNT_GROUP,
+//////  		Stream:   CORESERVER_NEWACCOUNT_QUEUE,
+//////  		Sentinel: cfg.CoreserverRedis,
+//////  	})
+//////  	defer r.Close()
+//////
+//////  	// w write message to kafka
+//////  	w := &kafka.Writer{
+//////  		Addr:     kafka.TCP(cfg.MarketingKafka.Addresses...),
+//////  		Balancer: &kafka.LeastBytes{},
+//////  	}
+//////  	defer w.Close()
+//////
+//////  	var error_wait_time = 0 * time.Millisecond
+//////  	for {
+//////  		// Step 0. wait a moment if need
+//////  		if error_wait_time > 0 {
+//////  			time.Sleep(error_wait_time)
+//////  		}
+//////
+//////  		// Step 1. fetch message
+//////  		var err error = nil
+//////  		var msg xstream.Message
+//////  		var newacctMsg NewAccountMessage
+//////  		log.Infof("waiting fetch newaccount message from redis queue...")
+//////  		if msg, err = r.FetchMessage(ctx); err != nil {
+//////  			log.Errorf("failed fetch message from redis queue, err:%s", err)
+//////  			//error_wait_time = 1000 * time.Millisecond
+//////  			break
+//////  		}
+//////
+//////          // Step 2. decode message
+//////          if err := json.Unmarshal(msg.Value, &newacctMsg); err != nil {
+//////          	log.Errorf("failed json.unmarshal message[%s], err:%s", string(msg.Value), err)
+//////          	continue
+//////          } else if newacctMsg.Userid == 0 || newacctMsg.Account == "" {
+//////          	log.Errorf("invalid newaccount message, userid=[%d]", newacctMsg.Userid)
+//////          	continue
+//////          }
+//////          log.Infof("fetch newaccount message success, account=[%s]", newacctMsg.Account)
+//////
+//////          // Step 3. handle message
+//////          // xxxxxxx handle message xxxxxxxxxxxxxxxxxxxxxxxx
+//////          // xxxxxxx handle message xxxxxxxxxxxxxxxxxxxxxxxx
+//////          // xxxxxxx handle message xxxxxxxxxxxxxxxxxxxxxxxx
+//////          // xxxxxxx handle message xxxxxxxxxxxxxxxxxxxxxxxx
+//////          // xxxxxxx handle message xxxxxxxxxxxxxxxxxxxxxxxx
+//////  		// Step 4. send ack to coreserver redis
+//////  		if err := r.CommitMessages(ctx, msg); err != nil {
+//////  			log.Errorf("failed commit message to redis, err:%s", err)
+//////  			error_wait_time = 1000 * time.Millisecond
+//////  			continue
+//////  		}
+//////  	}
+//////  }
+///////////////////////////// end example ////////////////////////////////////////////
+
+var (
+	redisPools = map[string]*redis.Pool{}
+)
+
+type Config struct {
+	Group    string          `toml:"group"`
+	Stream   string          `toml:"stream"`
+	Sentinel sentinel.Config `toml:"sentinel"`
+}
+
+type Message struct {
+	Id    string
+	Value []byte
+}
+
+type xstreamReader struct {
+	cfg      Config
+	id       string
+	name     string
+	conn     redis.Conn
+	connPool *redis.Pool
+	closed   bool
+	restart  bool
+}
+
+func NewReader(cfg Config) (r *xstreamReader) {
+	// redis pool
+	poolId := cfg.Sentinel.Sentinels
+	if _, ok := redisPools[poolId]; !ok {
+		redisPools[poolId] = sentinel.NewPool(cfg.Sentinel)
+	}
+
+	// initialize
+	connPool := redisPools[poolId]
+	r = &xstreamReader{
+		cfg:      cfg,
+		conn:     connPool.Get(),
+		connPool: connPool,
+		closed:   false,
+		restart:  true,
+	}
+	r.id = str.GetRandomNumString(12)
+	r.name = newName(r.conn, r.id, cfg.Group, cfg.Stream)
+	if err := r.initGroup(); err != nil {
+		panic(fmt.Sprintf("initGroup failed: error:%s", err))
+	}
+
+	// heart beat
+	go func() {
+		beatconn := redisPools[poolId].Get()
+		defer beatconn.Close()
+		for !r.closed {
+			id, err := redis.String(beatconn.Do("GETEX", r.name, "EX", 120))
+			if err != nil {
+				log.Warningf("redis GETEX failed, err:%s", err)
+				r.Close()
+			} else if id != r.id {
+				log.Warningf("name id in redis (%s) not equal to local name id(%s)", id, r.id)
+				r.Close()
+			}
+			time.Sleep(15 * time.Second)
+		}
+	}()
+
+	return r
+}
+
+func newName(conn redis.Conn, id, group, xstream string) string {
+	var (
+		mkeys            = []any{}
+		prefix           = fmt.Sprintf("xstream[%s]-group[%s]-", xstream, group)
+		consumer         = func(i int) string { return fmt.Sprintf("%s%03d", prefix, i) }
+		consumers        = map[string]string{}
+		MAX_CONSUMER_NUM = 30
+	)
+
+	// get all name
+	for i := 0; i < MAX_CONSUMER_NUM; i++ {
+		mkeys = append(mkeys, consumer(i))
+	}
+	ret, err := redis.Strings(conn.Do("MGET", mkeys...))
+	if err != nil {
+		log.Warningf("redis MGET error: %s", err)
+		panic("redis MGET error")
+	}
+	for i, s := range ret {
+		if s != "" {
+			name := consumer(i)
+			consumers[name] = s
+		}
+	}
+
+	var (
+		i          = 0
+		retry      = 0
+		consumer_i = ""
+	)
+LOOP:
+	// get unused name
+	for ; i < MAX_CONSUMER_NUM; i++ {
+		if _, ok := consumers[consumer(i)]; !ok {
+			consumer_i = consumer(i)
+			break
+		}
+	}
+	if i == MAX_CONSUMER_NUM {
+		panic(fmt.Sprintf("too many consumer(more than %d)???", MAX_CONSUMER_NUM))
+	}
+
+	// set to redis (try lock)
+	success, err := redis.Int(conn.Do("SETNX", consumer_i, id))
+	conn.Do("GETEX", consumer_i, "EX", 120) /* for SETNX failed forever */
+	if err != nil && retry >= 2 {
+		log.Warningf("redis SETNX(%s,%s) error: %s", consumer_i, id, err)
+		panic(fmt.Sprintf("redis SETNX error after try [%d] times", retry))
+	}
+
+	// return or retry if something error
+	if err != nil /*&& retry < 2*/ {
+		retry++
+		log.Warningf("redis SETNX(%s,%s) error: %s, retry(%d)", consumer_i, id, err, retry)
+		goto LOOP
+	}
+	if success == 0 {
+		i++
+		retry++
+		log.Warningf("consumer name(%s) used by others, retry(%d)", consumer_i, retry)
+		goto LOOP
+	}
+	log.Infof("success to get new consumer-name:[%s]", consumer_i)
+	return consumer_i
+}
+
+func (r *xstreamReader) Close() {
+	// delete client consumer name
+	id, err := redis.String(r.redisDo("GETEX", r.name, "EX", 120))
+	if err == nil && id == r.id {
+		r.redisDo("DEL", r.name)
+	}
+
+	// clonse redis connection(put to pool)
+	r.closed = true
+	r.conn.Close()
+}
+
+func (r *xstreamReader) fetchMessage() (ret []any, err error) {
+	// func (r *xstreamReader) fetchMessage(ctx context.Context) (ret []any, err error) {
+	args := []interface{}{
+		"GROUP", r.cfg.Group, r.name,
+		"COUNT", 1,
+		"BLOCK", 864000000,
+		"STREAMS", r.cfg.Stream,
+	}
+	if r.restart {
+		args = append(args, "0-0")
+	} else {
+		args = append(args, ">")
+	}
+
+	ret, err = redis.Values(r.redisDo("XREADGROUP", args...))
+
+	if err != nil {
+		return ret, err
+	} else if ret == nil {
+		return ret, nil
+	}
+	if len(ret) != 1 {
+		panic("len(ret) is not 1")
+	}
+	return ret, nil
+}
+
+func (r *xstreamReader) FetchMessage(ctx context.Context) (msg Message, err error) {
+	// func (r *xstreamReader) FetchMessage(ctx context.Context) (msg xstreamMessage, err error) {
+	// Step 1. get result from redis xstream
+	var ret, qmsg, msgs []any = nil, nil, nil
+	for {
+		ret, err = r.fetchMessage()
+		//////////////////////////////////////////////////////////////////////
+		////ret format example:
+		////  [["coreserver.txneedrepair", [["1675180029481-0", ["a", "b"]]]]]
+		//////////////////////////////////////////////////////////////////////
+		if err != nil {
+			log.Warningf("fetchMessage meassage err:%s", err)
+			return msg, errors.New("fetchMessage meassage err")
+		}
+		if len(ret) == 0 {
+			log.Warningf("redis stream result length is 0, which shoud bigger than 0")
+			return msg, errors.New("redis fetchMessage ret length is 0")
+		}
+
+		qmsg, _ = redis.Values(ret[0], nil)  // ["coreserver.txneedrepair", [["1675180029481-0", ["a", "b"]]]]
+		msgs, _ = redis.Values(qmsg[1], nil) // [["1675180029481-0", ["a", "b"]]]
+
+		if len(msgs) == 0 && r.restart {
+			r.restart = false
+			continue
+		}
+		if len(msgs) != 1 {
+			log.Warningf("meassage length from xstream is not 1(but %d)", len(msgs))
+			return msg, errors.New("no new msg")
+		}
+		break
+	}
+
+	// Step 2. parse xstream item result
+	var (
+		amsg, _   = redis.Values(msgs[0], nil) // ["1675180029481-0", ["a", "b"]]
+		rmsg, _   = redis.Values(amsg[1], nil) // ["a", "b"]
+		msgid, _  = redis.String(amsg[0], nil) // "1675180029481-0"
+		msgval, _ = redis.String(rmsg[1], nil) // "b"
+	)
+	msg.Id = msgid
+	msg.Value = []byte(msgval)
+	log.Infof("success fetched meassage '%s': %s", msg.Id, msgval)
+	return msg, nil
+}
+
+func (r *xstreamReader) CommitMessages(ctx context.Context, msg Message) (err error) {
+	args := []interface{}{
+		r.cfg.Stream, // key
+		r.cfg.Group,  // group
+		msg.Id,       // id
+	}
+
+	if cnt, err := redis.Int(r.redisDo("XACK", args...)); err != nil {
+		return err
+	} else if cnt != 1 {
+		log.Warningf("attention: 'XACK' return %d (not 1), but we trade as success", cnt)
+	}
+	return nil
+}
+
+func (r *xstreamReader) redisDo(cmd string, args ...any) (reply any, err error) {
+	if r.closed {
+		return nil, net.ErrClosed
+	} else if r.conn.Err() != nil {
+		log.Warningf("redis connection closed/unavaliable, we will get new one from pool")
+		oldconn := r.conn
+		newconn := r.connPool.Get()
+		r.conn = newconn
+		oldconn.Close()
+	}
+
+	return r.conn.Do(cmd, args...)
+}
+
+func (r *xstreamReader) initGroup() error {
+	groups, err := r.redisDo("XINfO", "GROUPS", r.cfg.Stream)
+	groupInfosInf, _ := groups.([]interface{})
+	if err != nil || len(groupInfosInf) == 0 {
+		log.Infof("create group, name: %s", r.cfg.Group)
+		_, err = r.redisDo("XGROUP", "CREATE", r.cfg.Stream, r.cfg.Group, "$", "MKSTREAM")
+		return err
+	}
+
+	for _, groupInfo := range groupInfosInf {
+		groupInfoInf := groupInfo.([]interface{})
+		groupName := string(groupInfoInf[1].([]uint8))
+		log.Infof("group name: %s", groupName)
+		if groupName == r.cfg.Group {
+			return nil
+		}
+	}
+	log.Infof("create group, name: %s", r.cfg.Group)
+	_, err = r.redisDo("XGROUP", "CREATE", r.cfg.Stream, r.cfg.Group, "$", "MKSTREAM")
+	return err
+}

xstream/writer.go

@@ -0,0 +1,20 @@
+package xstream
+
+//// func Example() {
+//// 	var (
+//// 		m      = model.NewModelDefault()
+//// 		queue  = "coreserver.newaccount"
+//// 		msgval = map[string]interface{}{
+//// 			"sign":    "",
+//// 			"appid":   c.AppId,
+//// 			"userid":  c.UserId,
+//// 			"account": input.Address,
+//// 		}
+//// 		msgstr, _ = json.Marshal(msgval)
+//// 	)
+//// 	defer m.Close()
+////
+//// 	if _, err = m.Redis.Do("XADD", queue, "*", "msg", msgstr); err != nil {
+//// 		log.Warningf("!!!!FAILED add newaccount to redis queue, msg is:'%s'", msgstr)
+//// 	}
+//// }