Compare commits

..

No commits in common. "master" and "v1.1.1" have entirely different histories.

6 changed files with 98 additions and 372 deletions

13
go.mod
View File

@ -1,14 +1,3 @@
module qoobing.com/gomod/redis module qoobing.com/gomod/redis
go 1.18 go 1.16
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
)

View File

@ -1,11 +1,45 @@
package redis package redis
import ( import (
"qoobing.com/gomod/redis/redis" "fmt"
"time"
"github.com/gomodule/redigo/redis"
"qoobing.com/gomod/redis/sentinel" "qoobing.com/gomod/redis/sentinel"
) )
type Config = sentinel.Config func newSentinelPool() *redis.Pool {
sntnl := &sentinel.Sentinel{
var NewPool = redis.NewPool Addrs: []string{":26379", ":26380", ":26381"},
var NewSentinelPool = sentinel.NewPool 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
},
}
}

View File

@ -1,98 +0,0 @@
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
},
}
}

View File

@ -1,65 +0,0 @@
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"))
}

View File

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

View File

@ -1,118 +1,28 @@
package xstream package xstream
import ( import (
"context"
"errors" "errors"
"fmt" "fmt"
"net" "net"
"time" "time"
"github.com/gomodule/redigo/redis" "github.com/garyburd/redigo/redis"
"qoobing.com/gomod/log" "qoobing.com/gomod/log"
"qoobing.com/gomod/redis/sentinel" "qoobing.com/gomod/redis/sentinel"
"qoobing.com/gomod/str" "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 ( var (
redisPools = map[string]*redis.Pool{} redisPools = map[string]*redis.Pool{}
) )
type Config struct { type Config struct {
Group string `toml:"group"` stream string
Stream string `toml:"stream"` group string
Sentinel sentinel.Config `toml:"sentinel"` sentinelConfig sentinel.Config
} }
type Message struct { type xstreamMessage struct {
Id string Id string
Value []byte Value []byte
} }
@ -129,9 +39,9 @@ type xstreamReader struct {
func NewReader(cfg Config) (r *xstreamReader) { func NewReader(cfg Config) (r *xstreamReader) {
// redis pool // redis pool
poolId := cfg.Sentinel.Sentinels poolId := cfg.sentinelConfig.Sentinels
if _, ok := redisPools[poolId]; !ok { if _, ok := redisPools[poolId]; !ok {
redisPools[poolId] = sentinel.NewPool(cfg.Sentinel) redisPools[poolId] = sentinel.NewPool(cfg.sentinelConfig)
} }
// initialize // initialize
@ -144,7 +54,7 @@ func NewReader(cfg Config) (r *xstreamReader) {
restart: true, restart: true,
} }
r.id = str.GetRandomNumString(12) r.id = str.GetRandomNumString(12)
r.name = newName(r.conn, r.id, cfg.Group, cfg.Stream) r.name = newName(r.conn, r.id, cfg.group)
if err := r.initGroup(); err != nil { if err := r.initGroup(); err != nil {
panic(fmt.Sprintf("initGroup failed: error:%s", err)) panic(fmt.Sprintf("initGroup failed: error:%s", err))
} }
@ -169,18 +79,17 @@ func NewReader(cfg Config) (r *xstreamReader) {
return r return r
} }
func newName(conn redis.Conn, id, group, xstream string) string { func newName(conn redis.Conn, id, group string) string {
var ( var (
mkeys = []any{} mkeys = []any{}
prefix = fmt.Sprintf("xstream[%s]-group[%s]-", xstream, group) consumer = fmt.Sprintf("%s-comsumer", group)
consumer = func(i int) string { return fmt.Sprintf("%s%03d", prefix, i) }
consumers = map[string]string{} consumers = map[string]string{}
MAX_CONSUMER_NUM = 30 MAX_CONSUMER_NUM = 30
) )
// get all name // get all name
for i := 0; i < MAX_CONSUMER_NUM; i++ { for i := 0; i < MAX_CONSUMER_NUM; i++ {
mkeys = append(mkeys, consumer(i)) mkeys = append(mkeys, fmt.Sprintf("%s-%03d", consumer, i))
} }
ret, err := redis.Strings(conn.Do("MGET", mkeys...)) ret, err := redis.Strings(conn.Do("MGET", mkeys...))
if err != nil { if err != nil {
@ -189,50 +98,38 @@ func newName(conn redis.Conn, id, group, xstream string) string {
} }
for i, s := range ret { for i, s := range ret {
if s != "" { if s != "" {
name := consumer(i) name := fmt.Sprintf("%s-%03d", consumer, i)
consumers[name] = s consumers[name] = s
log.Infof("name:[%s]", name)
} }
} }
var (
i = 0
retry = 0
consumer_i = ""
)
LOOP:
// get unused name // get unused name
for ; i < MAX_CONSUMER_NUM; i++ { for i := 0; i < MAX_CONSUMER_NUM; i++ {
if _, ok := consumers[consumer(i)]; !ok { name := fmt.Sprintf("%s-%03d", consumer, i)
consumer_i = consumer(i) if _, ok := consumers[name]; !ok {
consumer = name
break break
} }
} if i == MAX_CONSUMER_NUM-1 {
if i == MAX_CONSUMER_NUM { panic("too many consumer or something bug")
panic(fmt.Sprintf("too many consumer(more than %d)???", MAX_CONSUMER_NUM))
}
} }
// set to redis (try lock) // set to redis
success, err := redis.Int(conn.Do("SETNX", consumer_i, id)) success, err := redis.Int(conn.Do("SETNX", consumer, id))
conn.Do("GETEX", consumer_i, "EX", 120) /* for SETNX failed forever */ conn.Do("GETEX", consumer, "EX", 120)
if err != nil && retry >= 2 { if err != nil {
log.Warningf("redis SETNX(%s,%s) error: %s", consumer_i, id, err) log.Warningf("redis SETNX(%s,%s) error: %s", consumer, id, err)
panic(fmt.Sprintf("redis SETNX error after try [%d] times", retry)) panic("redis SETNX error")
}
// 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 { if success == 0 {
i++ log.Warningf("consumer name(%s) used by others", consumer)
retry++ panic("consumer name used by others")
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) conn.Do("GETEX", consumer, "EX", 120)
return consumer_i return consumer
} }
func (r *xstreamReader) Close() { func (r *xstreamReader) Close() {
@ -250,10 +147,10 @@ func (r *xstreamReader) Close() {
func (r *xstreamReader) fetchMessage() (ret []any, err error) { func (r *xstreamReader) fetchMessage() (ret []any, err error) {
// func (r *xstreamReader) fetchMessage(ctx context.Context) (ret []any, err error) { // func (r *xstreamReader) fetchMessage(ctx context.Context) (ret []any, err error) {
args := []interface{}{ args := []interface{}{
"GROUP", r.cfg.Group, r.name, "GROUP", r.cfg.group, r.name,
"COUNT", 1, "COUNT", 1,
"BLOCK", 864000000, "BLOCK", 864000000,
"STREAMS", r.cfg.Stream, "STREAMS", r.cfg.stream,
} }
if r.restart { if r.restart {
args = append(args, "0-0") args = append(args, "0-0")
@ -274,7 +171,7 @@ func (r *xstreamReader) fetchMessage() (ret []any, err error) {
return ret, nil return ret, nil
} }
func (r *xstreamReader) FetchMessage(ctx context.Context) (msg Message, err error) { func (r *xstreamReader) FetchMessage() (msg xstreamMessage, err error) {
// func (r *xstreamReader) FetchMessage(ctx context.Context) (msg xstreamMessage, err error) { // func (r *xstreamReader) FetchMessage(ctx context.Context) (msg xstreamMessage, err error) {
// Step 1. get result from redis xstream // Step 1. get result from redis xstream
var ret, qmsg, msgs []any = nil, nil, nil var ret, qmsg, msgs []any = nil, nil, nil
@ -320,10 +217,10 @@ func (r *xstreamReader) FetchMessage(ctx context.Context) (msg Message, err erro
return msg, nil return msg, nil
} }
func (r *xstreamReader) CommitMessages(ctx context.Context, msg Message) (err error) { func (r *xstreamReader) CommitMessages(msg xstreamMessage) (err error) {
args := []interface{}{ args := []interface{}{
r.cfg.Stream, // key r.cfg.stream, // key
r.cfg.Group, // group r.cfg.group, // group
msg.Id, // id msg.Id, // id
} }
@ -350,11 +247,11 @@ func (r *xstreamReader) redisDo(cmd string, args ...any) (reply any, err error)
} }
func (r *xstreamReader) initGroup() error { func (r *xstreamReader) initGroup() error {
groups, err := r.redisDo("XINfO", "GROUPS", r.cfg.Stream) groups, err := r.redisDo("XINfO", "GROUPS", r.cfg.stream)
groupInfosInf, _ := groups.([]interface{}) groupInfosInf, _ := groups.([]interface{})
if err != nil || len(groupInfosInf) == 0 { if err != nil || len(groupInfosInf) == 0 {
log.Infof("create group, name: %s", r.cfg.Group) log.Infof("create group, name: %s", r.cfg.group)
_, err = r.redisDo("XGROUP", "CREATE", r.cfg.Stream, r.cfg.Group, "$", "MKSTREAM") _, err = r.redisDo("XGROUP", "CREATE", r.cfg.stream, r.cfg.group, "$", "MKSTREAM")
return err return err
} }
@ -362,11 +259,11 @@ func (r *xstreamReader) initGroup() error {
groupInfoInf := groupInfo.([]interface{}) groupInfoInf := groupInfo.([]interface{})
groupName := string(groupInfoInf[1].([]uint8)) groupName := string(groupInfoInf[1].([]uint8))
log.Infof("group name: %s", groupName) log.Infof("group name: %s", groupName)
if groupName == r.cfg.Group { if groupName == r.cfg.group {
return nil return nil
} }
} }
log.Infof("create group, name: %s", r.cfg.Group) log.Infof("create group, name: %s", r.cfg.group)
_, err = r.redisDo("XGROUP", "CREATE", r.cfg.Stream, r.cfg.Group, "$", "MKSTREAM") _, err = r.redisDo("XGROUP", "CREATE", r.cfg.stream, r.cfg.group, "$", "MKSTREAM")
return err return err
} }