update

.idea/.gitignore

+# Default ignored files
+/shelf/
+/workspace.xml
+# Editor-based HTTP Client requests
+/httpRequests/
+# Datasource local storage ignored files
+/dataSources/
+/dataSources.local.xml

app/app.go

+// Package app is starting and stopping services gracefully, using golang.org/x/sync/errgroup to ensure that multiple services are started properly at the same time.
+package app
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"golang.org/x/sync/errgroup"
+
+	xlogger "xmall/pkg/logger"
+	"xmall/pkg/prof"
+)
+
+// IServer server interface
+type IServer interface {
+	Start() error
+	Stop() error
+	String() string
+}
+
+// Close app close
+type Close func() error
+
+// App servers
+type App struct {
+	servers []IServer
+	closes  []Close
+}
+
+// New create an app
+func New(servers []IServer, closes []Close) *App {
+	return &App{
+		servers: servers,
+		closes:  closes,
+	}
+}
+
+// Run servers
+func (a *App) Run() {
+	// ctx will be notified whenever an error occurs in one of the goroutines
+	eg, ctx := errgroup.WithContext(context.Background())
+
+	// start all servers
+	for _, server := range a.servers {
+		s := server
+		eg.Go(func() error {
+			xlogger.Info(s.String())
+			return s.Start()
+		})
+	}
+
+	// watch and stop app
+	eg.Go(func() error {
+		return a.watch(ctx)
+	})
+
+	fmt.Println(buff)
+
+	if err := eg.Wait(); err != nil {
+		panic(err)
+	}
+}
+
+// watch the os signal and the ctx signal from the errgroup, and stop the service if either signal is triggered
+func (a *App) watch(ctx context.Context) error {
+	sig := make(chan os.Signal, 1)
+	signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP, syscall.SIGTRAP)
+	profile := prof.NewProfile()
+
+	for {
+		select {
+		case <-ctx.Done(): // service error
+			_ = a.stop()
+			return ctx.Err()
+
+		case sigType := <-sig: // system notification signal
+			fmt.Printf("received system notification signal: %s\n", sigType.String())
+			switch sigType {
+			case syscall.SIGTRAP:
+				profile.StartOrStop() // start or stop sampling profile
+			case syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP:
+				if err := a.stop(); err != nil {
+					return err
+				}
+				fmt.Println("stop app successfully")
+				return nil
+			}
+		}
+	}
+}
+
+// stopping services and releasing resources
+func (a *App) stop() error {
+	for _, closeFn := range a.closes {
+		if err := closeFn(); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+const (
+	buff = `
+////////////////////////////////////////////////////////////////////
+//                          _ooOoo_                               //
+//                         o8888888o                              //
+//                         88" . "88                              //
+//                         (| ^_^ |)                              //
+//                         O\  =  /O                              //
+//                      ____/'---'\____                           //
+//                    .'  \\|     |//  '.                         //
+//                   /  \\|||  :  |||//  \                        //
+//                  /  _||||| -:- |||||-  \                       //
+//                  |   | \\\  -  /// |   |                       //
+//                  | \_|  ''\---/''  |   |                       //
+//                  \  .-\__  '-'  ___/-. /                       //
+//                ___'. .'  /--.--\  '. . ___                     //
+//              ."" '< '.___\_<|>_/___.'  >'"".                   //
+//            | | :  '- \'.;'\ _ /';.'/ - ' : | |                 //
+//            \  \ '-.   \_ __\ /__ _/   .-' /  /                 //
+//      ======='-.____'-.___\_____/___.-'____.-'=======           //
+//                          '=---='                               //
+//      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^           //
+//             佛祖保佑       永不宕机     永无BUG                   //
+////////////////////////////////////////////////////////////////////
+	`
+)

cache/cache.go

+// Package cache is memory and redis cache libraries.
+package cache
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/davecgh/go-spew/spew"
+)
+
+var (
+	// DefaultExpireTime default expiry time
+	DefaultExpireTime = time.Hour * 24
+	// DefaultNotFoundExpireTime expiry time when result is empty 1 minute,
+	// often used for cache time when data is empty (cache pass-through)
+	DefaultNotFoundExpireTime = time.Minute * 10
+
+	// NotFoundPlaceholder placeholder
+	NotFoundPlaceholder      = "*"
+	NotFoundPlaceholderBytes = []byte(NotFoundPlaceholder)
+	ErrPlaceholder           = errors.New("cache: placeholder")
+
+	// DefaultClient generate a cache client, where keyPrefix is generally the business prefix
+	DefaultClient Cache
+
+	CacheType_Redis  = "redis"
+	CacheType_Memory = "memory"
+)
+
+// Cache driver interface
+type Cache interface {
+	IncrBy(ctx context.Context, key string, value int64) (int64, error)
+	Set(ctx context.Context, key string, val interface{}, expiration time.Duration) error
+	Get(ctx context.Context, key string, val interface{}) error
+	MultiSet(ctx context.Context, valMap map[string]interface{}, expiration time.Duration) error
+	MultiGet(ctx context.Context, keys []string, valueMap interface{}) error
+	MGet(ctx context.Context, keys []string) ([]any, error)
+	Del(ctx context.Context, keys ...string) error
+	SetCacheWithNotFound(ctx context.Context, key string) error
+}
+
+// Set data
+func Set(ctx context.Context, key string, val interface{}, expiration time.Duration) error {
+	return DefaultClient.Set(ctx, key, val, expiration)
+}
+
+// Get data
+func Get(ctx context.Context, key string, val interface{}) error {
+	return DefaultClient.Get(ctx, key, val)
+}
+
+// MultiSet multiple set data
+func MultiSet(ctx context.Context, valMap map[string]interface{}, expiration time.Duration) error {
+	return DefaultClient.MultiSet(ctx, valMap, expiration)
+}
+
+// MultiGet multiple get data
+func MultiGet(ctx context.Context, keys []string, valueMap interface{}) error {
+	return DefaultClient.MultiGet(ctx, keys, valueMap)
+}
+
+// Del multiple delete data
+func Del(ctx context.Context, keys ...string) error {
+	return DefaultClient.Del(ctx, keys...)
+}
+
+// SetCacheWithNotFound .
+func SetCacheWithNotFound(ctx context.Context, key string) error {
+	return DefaultClient.SetCacheWithNotFound(ctx, key)
+}
+
+const max = 122
+
+func prettyBytes(output []byte) string {
+	if len(output) > max {
+		output = output[0:max]
+		return string(output) + "......"
+	}
+	return string(output)
+}
+
+func prettyVal(val any) string {
+	output := spew.Sdump(val)
+	if len(output) > max {
+		output = output[0:max] + "......"
+	}
+	return output
+}

cache/cache_test.go

+package cache
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"xmall/pkg/encoding"
+	"xmall/pkg/gotest"
+	"xmall/pkg/utils"
+)
+
+type cacheUser struct {
+	ID   uint64
+	Name string
+}
+
+func newCache() *gotest.Cache {
+	record1 := &cacheUser{
+		ID:   1,
+		Name: "foo",
+	}
+	record2 := &cacheUser{
+		ID:   2,
+		Name: "bar",
+	}
+
+	testData := map[string]interface{}{
+		utils.Uint64ToStr(record1.ID): record1,
+		utils.Uint64ToStr(record2.ID): record2,
+	}
+
+	c := gotest.NewCache(testData)
+	cachePrefix := ""
+	DefaultClient = NewRedisCache(c.RedisClient, cachePrefix, encoding.JSONEncoding{}, func() interface{} {
+		return &cacheUser{}
+	})
+	c.ICache = DefaultClient
+
+	return c
+}
+
+func TestCache(t *testing.T) {
+	c := newCache()
+	defer c.Close()
+	testData := c.TestDataSlice[0].(*cacheUser)
+
+	key := utils.Uint64ToStr(testData.ID)
+	err := Set(c.Ctx, key, c.TestDataMap[key], time.Minute)
+	assert.NoError(t, err)
+
+	val := &cacheUser{}
+	err = Get(c.Ctx, key, val)
+	assert.NoError(t, err)
+	assert.Equal(t, testData.Name, val.Name)
+
+	err = Del(c.Ctx, key)
+	assert.NoError(t, err)
+
+	err = MultiSet(c.Ctx, c.TestDataMap, time.Minute)
+	assert.NoError(t, err)
+
+	var keys []string
+	for k := range c.TestDataMap {
+		keys = append(keys, k)
+	}
+	vals := make(map[string]*cacheUser)
+	err = MultiGet(c.Ctx, keys, vals)
+	assert.NoError(t, err)
+	assert.Equal(t, len(c.TestDataSlice), len(vals))
+
+	err = SetCacheWithNotFound(c.Ctx, "not_found")
+	assert.NoError(t, err)
+}

cache/memory.go

+package cache
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"reflect"
+	"time"
+
+	"github.com/dgraph-io/ristretto"
+	"github.com/spf13/cast"
+
+	"xmall/pkg/encoding"
+	ctxUtil "xmall/pkg/gin/xctx"
+	"xmall/pkg/logger"
+	xslice "xmall/pkg/utils/xslice"
+	"xmall/pkg/xerrors/xerror"
+)
+
+type memoryCache struct {
+	client            *ristretto.Cache
+	KeyPrefix         string
+	encoding          encoding.Encoding
+	DefaultExpireTime time.Duration
+	newObject         func() interface{}
+}
+
+// NewMemoryCache create a memory cache
+func NewMemoryCache(keyPrefix string, encode encoding.Encoding, newObject func() interface{}) Cache {
+	// see: https://dgraph.io/blog/post/introducing-ristretto-high-perf-go-cache/
+	//		https://www.start.io/blog/we-chose-ristretto-cache-for-go-heres-why/
+	config := &ristretto.Config{
+		NumCounters: 1e7,     // number of keys to track frequency of (10M).
+		MaxCost:     1 << 30, // maximum cost of cache (1GB).
+		BufferItems: 64,      // number of keys per Get buffer.
+	}
+	store, err := ristretto.NewCache(config)
+	if err != nil {
+		panic("new ristretto cache err:" + err.Error())
+	}
+	return &memoryCache{
+		client:    store,
+		KeyPrefix: keyPrefix,
+		encoding:  encode,
+		newObject: newObject,
+	}
+}
+
+const logPrefix = "ristretto:"
+
+// Set data
+func (m *memoryCache) Set(ctx context.Context, key string, val interface{}, expiration time.Duration) error {
+	buf, err := encoding.Marshal(m.encoding, val)
+	if err != nil {
+		return xerror.Errorf("encoding.Marshal error: %v, key=%s, val=%+v ", err, key, val)
+	}
+	if len(buf) == 0 {
+		buf = NotFoundPlaceholderBytes
+	}
+	cacheKey, err := BuildCacheKey(m.KeyPrefix, key)
+	if err != nil {
+		return xerror.Errorf("BuildCacheKey error: %v, key=%s", err, key)
+	}
+	st := time.Now()
+	ok := m.client.SetWithTTL(cacheKey, buf, 0, expiration)
+	logger.Info(fmt.Sprintf("%s set %s", logPrefix, cacheKey), logger.Any("ok", ok), logger.Any("val", prettyBytes(buf)), logger.Any("cost", time.Since(st).String()), logger.Any("expiration", expiration.String()), ctxUtil.CtxTraceIDField(ctx))
+	if !ok {
+		return xerror.New("SetWithTTL failed")
+	}
+
+	return nil
+}
+
+func (m *memoryCache) IncrBy(ctx context.Context, key string, val int64) (int64, error) {
+	cacheKey, err := BuildCacheKey(m.KeyPrefix, key)
+	if err != nil {
+		return 0, xerror.Errorf("BuildCacheKey error: %v, key=%s", err, key)
+	}
+	v, ok := m.client.Get(cacheKey)
+	if !ok {
+		_ = m.client.Set(cacheKey, val, 0)
+		return val, nil
+	}
+	st := time.Now()
+	cur := cast.ToInt64(v)
+	_ = m.client.Set(cacheKey, cur+val, 0)
+	logger.Info(fmt.Sprintf("%s incrby %s", logPrefix, cacheKey), logger.Any("ok", ok), logger.Any("cost", time.Since(st).String()), logger.Any("result", val), ctxUtil.CtxTraceIDField(ctx))
+	return cur + val, nil
+}
+
+// Get data
+func (m *memoryCache) Get(ctx context.Context, key string, val interface{}) error {
+	cacheKey, err := BuildCacheKey(m.KeyPrefix, key)
+	if err != nil {
+		return xerror.Errorf("BuildCacheKey error: %v, key=%s", err, key)
+	}
+
+	st := time.Now()
+	data, ok := m.client.Get(cacheKey)
+	if !ok {
+		logger.Info(fmt.Sprintf("%s get %s", logPrefix, cacheKey), logger.Any("ok", ok), logger.Any("cost", time.Since(st).String()), ctxUtil.CtxTraceIDField(ctx))
+		return CacheNotFound
+	}
+
+	dataBytes, ok := data.([]byte)
+	if !ok {
+		return xerror.Errorf("data type error, key=%s, type=%T", key, data)
+	}
+	if len(dataBytes) == 0 || bytes.Equal(dataBytes, NotFoundPlaceholderBytes) {
+		logger.Info(fmt.Sprintf("%s get %s", logPrefix, cacheKey), logger.Any("ok", ok), logger.Any("result", "NotFoundPlaceholder"), ctxUtil.CtxTraceIDField(ctx))
+		return ErrPlaceholder
+	}
+
+	err = encoding.Unmarshal(m.encoding, dataBytes, val)
+	if err != nil {
+		return xerror.Errorf("encoding.Unmarshal error: %v, key=%s, cacheKey=%s, type=%T, data=%s ",
+			err, key, cacheKey, val, dataBytes)
+	}
+	logger.Info(fmt.Sprintf("%s get %s", logPrefix, cacheKey), logger.Any("ok", ok), logger.Any("cost", time.Since(st).String()), logger.Any("result", prettyVal(val)), ctxUtil.CtxTraceIDField(ctx))
+	return nil
+}
+
+// Del delete data
+func (m *memoryCache) Del(ctx context.Context, keys ...string) error {
+	if len(keys) == 0 {
+		return nil
+	}
+
+	key := keys[0]
+	cacheKey, err := BuildCacheKey(m.KeyPrefix, key)
+	if err != nil {
+		return xerror.Errorf("build cache key error, err=%v, key=%s", err, key)
+	}
+	st := time.Now()
+	m.client.Del(cacheKey)
+	logger.Info(fmt.Sprintf("%s del %s", logPrefix, cacheKey), logger.Any("cost", time.Since(st).String()), ctxUtil.CtxTraceIDField(ctx))
+	return nil
+}
+
+// MultiSet multiple set data
+func (m *memoryCache) MultiSet(ctx context.Context, valueMap map[string]interface{}, expiration time.Duration) error {
+	var err error
+	for key, value := range valueMap {
+		err = m.Set(ctx, key, value, expiration)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// MultiGet multiple get data
+// deprecated
+func (m *memoryCache) MultiGet(ctx context.Context, keys []string, value interface{}) error {
+	valueMap := reflect.ValueOf(value)
+	var err error
+	for _, key := range keys {
+		object := m.newObject()
+		err = m.Get(ctx, key, object)
+		if err != nil {
+			continue
+		}
+		valueMap.SetMapIndex(reflect.ValueOf(key), reflect.ValueOf(object))
+	}
+	return nil
+}
+
+func (m *memoryCache) MGet(ctx context.Context, keys []string) ([]any, error) {
+	st := time.Now()
+	rs := make([]any, len(keys))
+	for i, key := range keys {
+		rs[i], _ = m.client.Get(key)
+	}
+	logger.Info(fmt.Sprintf("%s: mget %s", logPrefix, xslice.Join(keys, " ")), logger.Any("result", rs), logger.Any("cost", time.Since(st).String()), logger.Any("expiration", DefaultNotFoundExpireTime.String()), ctxUtil.CtxTraceIDField(ctx))
+	return rs, nil
+}
+
+// SetCacheWithNotFound set not found
+func (m *memoryCache) SetCacheWithNotFound(ctx context.Context, key string) error {
+	cacheKey, err := BuildCacheKey(m.KeyPrefix, key)
+	if err != nil {
+		return xerror.Errorf("BuildCacheKey error: %v, key=%s", err, key)
+	}
+	st := time.Now()
+	ok := m.client.SetWithTTL(cacheKey, []byte(NotFoundPlaceholder), 0, DefaultNotFoundExpireTime)
+	if !ok {
+		return xerror.New("SetWithTTL failed")
+	}
+	logger.Info(fmt.Sprintf("%s: set %s [NotFound]", logPrefix, cacheKey), logger.Any("ok", ok), logger.Any("cost", time.Since(st).String()), logger.Any("expiration", DefaultNotFoundExpireTime.String()), ctxUtil.CtxTraceIDField(ctx))
+
+	return nil
+}

cache/memory_test.go

+package cache
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"xmall/pkg/encoding"
+	"xmall/pkg/gotest"
+	"xmall/pkg/utils"
+)
+
+type memoryUser struct {
+	ID   uint64
+	Name string
+}
+
+func newMemoryCache() *gotest.Cache {
+	record1 := &memoryUser{
+		ID:   1,
+		Name: "foo",
+	}
+	record2 := &memoryUser{
+		ID:   2,
+		Name: "bar",
+	}
+
+	testData := map[string]interface{}{
+		utils.Uint64ToStr(record1.ID): record1,
+		utils.Uint64ToStr(record2.ID): record2,
+	}
+
+	c := gotest.NewCache(testData)
+	cachePrefix := ""
+	c.ICache = NewMemoryCache(cachePrefix, encoding.JSONEncoding{}, func() interface{} {
+		return &memoryUser{}
+	})
+
+	return c
+}
+
+func TestMemoryCache(t *testing.T) {
+	c := newMemoryCache()
+	defer c.Close()
+	testData := c.TestDataSlice[0].(*memoryUser)
+	iCache := c.ICache.(Cache)
+
+	key := utils.Uint64ToStr(testData.ID)
+	err := iCache.Set(c.Ctx, key, c.TestDataMap[key], time.Minute)
+	assert.NoError(t, err)
+
+	time.Sleep(time.Millisecond)
+	val := &memoryUser{}
+	err = iCache.Get(c.Ctx, key, val)
+	assert.NoError(t, err)
+	assert.Equal(t, testData.Name, val.Name)
+
+	err = iCache.Del(c.Ctx, key)
+	assert.NoError(t, err)
+
+	time.Sleep(time.Millisecond)
+	err = iCache.MultiSet(c.Ctx, c.TestDataMap, time.Minute)
+	assert.NoError(t, err)
+
+	time.Sleep(time.Millisecond)
+	var keys []string
+	for k := range c.TestDataMap {
+		keys = append(keys, k)
+	}
+	vals := make(map[string]*memoryUser)
+	err = iCache.MultiGet(c.Ctx, keys, vals)
+	assert.NoError(t, err)
+	assert.Equal(t, len(c.TestDataSlice), len(vals))
+
+	err = iCache.SetCacheWithNotFound(c.Ctx, "not_found")
+	assert.NoError(t, err)
+}
+
+func TestMemoryCacheError(t *testing.T) {
+	c := newMemoryCache()
+	defer c.Close()
+	testData := c.TestDataSlice[0].(*memoryUser)
+	iCache := c.ICache.(Cache)
+
+	// Set empty key error test
+	key := utils.Uint64ToStr(testData.ID)
+	err := iCache.Set(c.Ctx, "", c.TestDataMap[key], time.Minute)
+	assert.Error(t, err)
+
+	// Set empty value error test
+	key = utils.Uint64ToStr(testData.ID)
+	err = iCache.Set(c.Ctx, key, nil, time.Minute)
+	assert.Error(t, err)
+
+	// Get empty key error test
+	val := &memoryUser{}
+	err = iCache.Get(c.Ctx, "", val)
+	assert.Error(t, err)
+
+	// Get empty result  test
+	key = utils.Uint64ToStr(testData.ID)
+	err = iCache.Get(c.Ctx, key, val)
+	assert.Error(t, err)
+
+	// Get result error test
+	key = utils.Uint64ToStr(testData.ID)
+	_ = iCache.Set(c.Ctx, key, c.TestDataMap[key], time.Minute)
+	time.Sleep(time.Millisecond)
+	err = iCache.Get(c.Ctx, key, nil)
+	assert.Error(t, err)
+
+	// Del empty key error test
+	err = iCache.Del(c.Ctx)
+	assert.NoError(t, err)
+	err = iCache.Del(c.Ctx, "")
+	assert.Error(t, err)
+
+	// empty key test
+	err = iCache.SetCacheWithNotFound(c.Ctx, "")
+	assert.Error(t, err)
+}

cache/redis_test.go

+package cache
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"xmall/pkg/encoding"
+	"xmall/pkg/gotest"
+	"xmall/pkg/utils"
+)
+
+type redisUser struct {
+	ID   uint64
+	Name string
+}
+
+func newTestData() map[string]interface{} {
+	record1 := &redisUser{
+		ID:   1,
+		Name: "foo",
+	}
+	record2 := &redisUser{
+		ID:   2,
+		Name: "bar",
+	}
+
+	return map[string]interface{}{
+		utils.Uint64ToStr(record1.ID): record1,
+		utils.Uint64ToStr(record2.ID): record2,
+	}
+}
+
+func newRedisCache() *gotest.Cache {
+	testData := newTestData()
+	c := gotest.NewCache(testData)
+	cachePrefix := ""
+	c.ICache = NewRedisCache(c.RedisClient, cachePrefix, encoding.JSONEncoding{}, func() interface{} {
+		return &redisUser{}
+	})
+
+	return c
+}
+
+func newRedisClusterCache() *gotest.RCCache {
+	testData := newTestData()
+	c := gotest.NewRCCache(testData)
+	cachePrefix := ""
+	c.ICache = NewRedisClusterCache(c.RedisClient, cachePrefix, encoding.JSONEncoding{}, func() interface{} {
+		return &redisUser{}
+	})
+
+	return c
+}
+
+func TestRedisCache(t *testing.T) {
+	c := newRedisCache()
+	defer c.Close()
+	testData := c.TestDataSlice[0].(*redisUser)
+	iCache := c.ICache.(Cache)
+
+	key := utils.Uint64ToStr(testData.ID)
+	err := iCache.Set(c.Ctx, key, c.TestDataMap[key], time.Minute)
+	assert.NoError(t, err)
+	//err = iCache.Set(c.Ctx, key, c.TestDataMap[key], 0)
+	//assert.NoError(t, err)
+
+	val := &redisUser{}
+	err = iCache.Get(c.Ctx, key, val)
+	assert.NoError(t, err)
+	assert.Equal(t, testData.Name, val.Name)
+
+	err = iCache.Del(c.Ctx, key)
+	assert.NoError(t, err)
+
+	err = iCache.MultiSet(c.Ctx, c.TestDataMap, time.Minute)
+	assert.NoError(t, err)
+	//err = iCache.MultiSet(c.Ctx, c.TestDataMap, 0)
+	//assert.NoError(t, err)
+
+	var keys []string
+	for k := range c.TestDataMap {
+		keys = append(keys, k)
+	}
+	vals := make(map[string]*redisUser)
+	err = iCache.MultiGet(c.Ctx, keys, vals)
+
+	assert.NoError(t, err)
+	assert.Equal(t, len(c.TestDataSlice), len(vals))
+	err = iCache.SetCacheWithNotFound(c.Ctx, "not_found")
+	assert.NoError(t, err)
+}
+
+func TestRedisCacheError(t *testing.T) {
+	c := newRedisCache()
+	defer c.Close()
+	testData := c.TestDataSlice[0].(*redisUser)
+	iCache := c.ICache.(Cache)
+
+	// Set empty key error test
+	key := utils.Uint64ToStr(testData.ID)
+	err := iCache.Set(c.Ctx, "", c.TestDataMap[key], time.Minute)
+	assert.Error(t, err)
+
+	// Set empty value error test
+	key = utils.Uint64ToStr(testData.ID)
+	err = iCache.Set(c.Ctx, key, nil, time.Minute)
+	assert.Error(t, err)
+
+	// Get empty key error test
+	val := &redisUser{}
+	err = iCache.Get(c.Ctx, "", val)
+	assert.Error(t, err)
+
+	// Get empty result  test
+	key = utils.Uint64ToStr(testData.ID)
+	err = iCache.Get(c.Ctx, key, val)
+	assert.Error(t, err)
+
+	// Get result error test
+	key = utils.Uint64ToStr(testData.ID)
+	_ = iCache.Set(c.Ctx, key, c.TestDataMap[key], time.Minute)
+	time.Sleep(time.Millisecond)
+	err = iCache.Get(c.Ctx, key, nil)
+	assert.Error(t, err)
+
+	_ = iCache.MultiSet(c.Ctx, nil, time.Minute)
+	_ = iCache.MultiGet(c.Ctx, nil, time.Minute)
+
+	// Del empty key error test
+	err = iCache.Del(c.Ctx)
+	assert.NoError(t, err)
+	err = iCache.Del(c.Ctx, "")
+	assert.NoError(t, err)
+}
+
+func TestRedisClusterCache(t *testing.T) {
+	c := newRedisClusterCache()
+	defer c.Close()
+	testData := c.TestDataSlice[0].(*redisUser)
+	iCache := c.ICache.(Cache)
+
+	key := utils.Uint64ToStr(testData.ID)
+	err := iCache.Set(c.Ctx, key, c.TestDataMap[key], time.Minute)
+	assert.NoError(t, err)
+
+	val := &redisUser{}
+	err = iCache.Get(c.Ctx, key, val)
+	assert.NoError(t, err)
+	assert.Equal(t, testData.Name, val.Name)
+
+	err = iCache.Del(c.Ctx, key)
+	assert.NoError(t, err)
+
+	err = iCache.MultiSet(c.Ctx, c.TestDataMap, time.Minute)
+	assert.NoError(t, err)
+
+	var keys []string
+	for k := range c.TestDataMap {
+		keys = append(keys, k)
+	}
+	vals := make(map[string]*redisUser)
+	err = iCache.MultiGet(c.Ctx, keys, vals)
+
+	assert.NoError(t, err)
+	assert.Equal(t, len(c.TestDataSlice), len(vals))
+	err = iCache.SetCacheWithNotFound(c.Ctx, "not_found")
+	assert.NoError(t, err)
+}
+
+func TestRedisClusterCacheError(t *testing.T) {
+	c := newRedisClusterCache()
+	defer c.Close()
+	testData := c.TestDataSlice[0].(*redisUser)
+	iCache := c.ICache.(Cache)
+
+	// Set empty key error test
+	key := utils.Uint64ToStr(testData.ID)
+	err := iCache.Set(c.Ctx, "", c.TestDataMap[key], time.Minute)
+	assert.Error(t, err)
+
+	// Set empty value error test
+	key = utils.Uint64ToStr(testData.ID)
+	err = iCache.Set(c.Ctx, key, nil, time.Minute)
+	assert.Error(t, err)
+
+	// Get empty key error test
+	val := &redisUser{}
+	err = iCache.Get(c.Ctx, "", val)
+	assert.Error(t, err)
+
+	// Get empty result  test
+	key = utils.Uint64ToStr(testData.ID)
+	err = iCache.Get(c.Ctx, key, val)
+	assert.Error(t, err)
+
+	// Get result error test
+	key = utils.Uint64ToStr(testData.ID)
+	_ = iCache.Set(c.Ctx, key, c.TestDataMap[key], time.Minute)
+	time.Sleep(time.Millisecond)
+	err = iCache.Get(c.Ctx, key, nil)
+	assert.Error(t, err)
+
+	_ = iCache.MultiSet(c.Ctx, nil, time.Minute)
+	_ = iCache.MultiGet(c.Ctx, nil, time.Minute)
+
+	// Del empty key error test
+	err = iCache.Del(c.Ctx)
+	assert.NoError(t, err)
+	err = iCache.Del(c.Ctx, "")
+	assert.NoError(t, err)
+}
+
+func TestBuildCacheKey(t *testing.T) {
+	_, err := BuildCacheKey("", "")
+	assert.Error(t, err)
+	_, err = BuildCacheKey("foo", "bar")
+	assert.NoError(t, err)
+}

captcha/captcha.go

+package captcha
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	"github.com/mojocn/base64Captcha"
+	"github.com/redis/go-redis/v9"
+)
+
+type Captcha struct {
+	captcha *base64Captcha.Captcha
+}
+
+func (r *Captcha) Verify(id, answer string, clear bool) bool {
+	return r.captcha.Store.Verify(id, answer, clear)
+}
+
+func (r *Captcha) Generate(ctx context.Context) (id, b64s, answer string, err error) {
+	return r.captcha.Generate()
+}
+
+type RedisStore struct {
+	client     *redis.Client // 支持单节点和集群模式
+	prefix     string        // Key前缀隔离
+	expiration time.Duration // 验证码有效期
+}
+
+func NewRedisStore(redisClient *redis.Client, prefix string) *RedisStore {
+	return &RedisStore{
+		client:     redisClient,
+		prefix:     prefix,
+		expiration: 10 * time.Minute,
+	}
+}
+
+func (rs *RedisStore) Set(id string, value string) error {
+	ctx := context.Background()
+	fullValue := value + "|" + time.Now().Format(time.RFC3339)
+	return rs.client.Set(ctx, rs.prefix+id, fullValue, rs.expiration).Err()
+}
+
+func (rs *RedisStore) Get(id string, clear bool) string {
+	ctx := context.Background()
+	key := rs.prefix + id
+	val, err := rs.client.Get(ctx, key).Result()
+	if err != nil {
+		panic(err)
+	}
+	if clear {
+		go rs.client.Del(ctx, key)
+	}
+	parts := strings.Split(val, "|")
+	if len(parts) != 2 {
+		return ""
+	}
+	return parts[0]
+}
+
+func (r *RedisStore) Verify(id, answer string, clear bool) bool {
+	stored := r.Get(id, clear)
+	return strings.EqualFold(stored, answer)
+}
+
+func NewCaptcha(redisClient *redis.Client) *Captcha {
+	store := NewRedisStore(redisClient, "captcha:")
+	driver := &base64Captcha.DriverDigit{
+		Height:   80,
+		Width:    240,
+		Length:   5,
+		MaxSkew:  0.7,
+		DotCount: 80,
+	}
+	return &Captcha{
+		captcha: base64Captcha.NewCaptcha(driver, store),
+	}
+}

conf/consul_parse.go

+package conf
+
+import (
+	"bytes"
+	"log"
+	"xmall/pkg/xerrors/xerror"
+
+	consulapi "github.com/armon/consul-api"
+	"github.com/spf13/viper"
+)
+
+const (
+	DefaultEndPoint   = "consul.default.svc.cluster.local:8500"
+	DefaultPath       = "xmall/conf"
+	DefaultToken      = "9a6bc59e-73d7-7af5-a669-818a200a9ded"
+	DefaultConfigType = "yml"
+)
+
+// JwtOption set the jwt options.
+type ConsulOption func(*ConsulParser)
+
+func WithToken(token string) ConsulOption {
+	return func(o *ConsulParser) {
+		o.token = token
+	}
+}
+
+func WithConfigType(configType string) ConsulOption {
+	return func(o *ConsulParser) {
+		o.configType = configType
+	}
+}
+
+type ConsulParser struct {
+	endpoint     string
+	configPath   string
+	configType   string
+	token        string
+	cfg          *consulapi.Config
+	consulClient *consulapi.Client
+}
+
+func NewConsulParser(endpoint, configPath string, opts ...ConsulOption) (*ConsulParser, error) {
+	if endpoint == "" {
+		return nil, xerror.New("endpoint cannot be empty")
+	}
+	if configPath == "" {
+		return nil, xerror.New("configPath cannot be empty")
+	}
+	o := &ConsulParser{endpoint: endpoint, configPath: configPath}
+	for _, opt := range opts {
+		opt(o)
+	}
+	ccfg := consulapi.Config{Address: endpoint, Token: o.token}
+	o.cfg = &ccfg
+	consulClient, err := consulapi.NewClient(&ccfg)
+	if err != nil {
+		log.Fatal(err)
+	}
+	o.consulClient = consulClient
+	return o, nil
+}
+
+func (x *ConsulParser) Read(obj any) error {
+	kv, _, err := x.consulClient.KV().Get(x.configPath, nil)
+	if err != nil {
+		return err
+	}
+	viper.SetConfigType(x.configType)
+	err = viper.ReadConfig(bytes.NewBuffer(kv.Value))
+	if err != nil {
+		return err
+	}
+	err = viper.Unmarshal(obj)
+	if err != nil {
+		return err
+	}
+	return nil
+}

conf/local_parse.go

+// Package conf is parsing yaml, json, toml configuration files to go struct.
+package conf
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"path"
+	"path/filepath"
+	"strings"
+	"xmall/pkg/xerrors/xerror"
+
+	"github.com/fsnotify/fsnotify"
+	"github.com/spf13/viper"
+)
+
+// Parse configuration files to struct, including yaml, toml, json, etc., and turn on listening for configuration file changes if fs is not empty
+func Parse(configFile string, obj interface{}, reloads ...func()) error {
+	confFileAbs, err := filepath.Abs(configFile)
+	if err != nil {
+		return err
+	}
+
+	filePathStr, filename := filepath.Split(confFileAbs)
+	ext := strings.TrimLeft(path.Ext(filename), ".")
+	filename = strings.ReplaceAll(filename, "."+ext, "") // excluding suffix names
+
+	viper.AddConfigPath(filePathStr) // path
+	viper.SetConfigName(filename)    // file name
+	viper.SetConfigType(ext)         // get the configuration type from the file name
+	err = viper.ReadInConfig()
+	if err != nil {
+		return xerror.New(err.Error())
+	}
+
+	err = viper.Unmarshal(obj)
+	if err != nil {
+		return xerror.New(err.Error())
+	}
+
+	if len(reloads) > 0 {
+		watchConfig(obj, reloads...)
+	}
+
+	return nil
+}
+
+// ParseConfigData parse data to struct
+func ParseConfigData(data []byte, format string, obj interface{}) error {
+	viper.SetConfigType(format)
+	err := viper.ReadConfig(bytes.NewBuffer(data))
+	if err != nil {
+		return err
+	}
+
+	return viper.Unmarshal(obj)
+}
+
+// listening for profile updates
+func watchConfig(obj interface{}, reloads ...func()) {
+	viper.WatchConfig()
+
+	// Note: OnConfigChange is called twice on Windows
+	viper.OnConfigChange(func(e fsnotify.Event) {
+		err := viper.Unmarshal(obj)
+		if err != nil {
+			fmt.Println("viper.Unmarshal error: ", err)
+		} else {
+			for _, reload := range reloads {
+				reload()
+			}
+		}
+	})
+}
+
+// Show print configuration information (hide sensitive fields)
+func Show(obj interface{}, fields ...string) string {
+	var out string
+
+	data, err := json.MarshalIndent(obj, "", "    ")
+	if err != nil {
+		fmt.Println("json.MarshalIndent error: ", err)
+		return ""
+	}
+
+	buf := bufio.NewReader(bytes.NewReader(data))
+	for {
+		line, err := buf.ReadString('\n')
+		if err != nil {
+			break
+		}
+		fields = append(fields, `"dsn"`, `"password"`, `"pwd"`, `"signKey"`, `"access-key-id"`, `"access-key-secret"`)
+
+		out += HideSensitiveFields(line, fields...)
+	}
+
+	return out
+}
+
+func HideSensitiveFields(line string, fields ...string) string {
+	for _, field := range fields {
+		if strings.Contains(line, field) {
+			index := strings.Index(line, field)
+			if strings.Contains(line, "@") && strings.Contains(line, ":") {
+				return replaceDSN(line)
+			}
+			return fmt.Sprintf("%s: \"******\",\n", line[:index+len(field)])
+		}
+	}
+
+	// replace dsn
+	if strings.Contains(line, "@") && strings.Contains(line, ":") {
+		return replaceDSN(line)
+	}
+
+	return line
+}
+
+// replace dsn password
+func replaceDSN(str string) string {
+	data := []byte(str)
+	start, end := 0, 0
+	for k, v := range data {
+		if v == ':' {
+			start = k
+		}
+		if v == '@' {
+			end = k
+			break
+		}
+	}
+
+	if start >= end {
+		return str
+	}
+
+	return fmt.Sprintf("%s******%s", data[:start+1], data[end:])
+}

conf/parse_test.go

+package conf
+
+import (
+	"fmt"
+	"os"
+	"testing"
+	"time"
+)
+
+var c = make(map[string]interface{})
+
+func TestShow(t *testing.T) {
+	t.Log(Show(c))
+	t.Log(Show(make(chan string)))
+}
+
+func Test_replaceDSN(t *testing.T) {
+	dsn := "default:123456@192.168.3.37:6379/0"
+	t.Log(replaceDSN(dsn))
+
+	dsn = "default:123456:192.168.3.37:6379/0"
+	t.Log(replaceDSN(dsn))
+}
+
+func Test_hideSensitiveFields(t *testing.T) {
+	var keywords []string
+	keywords = append(keywords, `"dsn"`, `"password"`, `"name"`)
+	str := Show(c, keywords...)
+
+	fmt.Printf(HideSensitiveFields(str))
+
+	str = "\ndefault:123456@192.168.3.37:6379/0\n"
+	fmt.Printf(HideSensitiveFields(str))
+}
+
+// test listening for configuration file updates
+func TestParse(t *testing.T) {
+	conf := make(map[string]interface{})
+
+	reloads := []func(){
+		func() {
+			fmt.Println("close and reconnect mysql")
+			fmt.Println("close and reconnect redis")
+		},
+	}
+
+	err := Parse("test.yml", &conf, reloads...)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	time.Sleep(time.Second)
+	content, _ := os.ReadFile("test.yml")
+	contentChange := append(content, byte('#'))
+	time.Sleep(time.Millisecond * 100)
+	_ = os.WriteFile("test.yml", contentChange, 0666) // change file
+	time.Sleep(time.Millisecond * 100)
+	_ = os.WriteFile("test.yml", content, 0666) // recovery documents
+	time.Sleep(time.Millisecond * 100)
+}
+
+func TestParseErr(t *testing.T) {
+	// result error test
+	err := Parse("test.yml", nil)
+	t.Log(err)
+
+	// not found error test
+	err = Parse("notfound.yml", &c)
+	t.Log(err)
+}
+
+func TestParseConfigData(t *testing.T) {
+	conf := make(map[string]interface{})
+
+	data, err := os.ReadFile("test.yml")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	err = ParseConfigData(data, "yaml", &conf)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	t.Log(Show(conf))
+}

conf/test.yml

+# app settings
+app:
+  name: "serverNameExample"
+  env: "dev1"
+  version: "v0.0.0"
+  password: "123456"
+
+database:
+  driver: "mysql"
+# mysql settings
+  mysql:
+    # dsn format, <user>:<pass>@(127.0.0.1:3306)/<db>?[k=v& ......]
+    dsn: "root:123456@(192.168.3.37:3306)/account?parseTime=true&loc=Local&charset=utf8,utf8mb4"
+
+# redis settings
+redis:
+  # dsn format, [user]:<pass>@]127.0.0.1:6379/[db]
+  dsn: "default:123456@192.168.3.37:6379/0"
+  dialTimeout: 10
+  readTimeout: 2
+  writeTimeout: 2

consulcli/consulcli.go

+// Package consulcli is connecting to the consul service client.
+package consulcli
+
+import (
+	"fmt"
+	"xmall/pkg/xerrors/xerror"
+
+	"github.com/hashicorp/consul/api"
+)
+
+// Init connecting to the consul service
+// Note: If the WithConfig(*api.Config) parameter is set, the addr parameter is ignored!
+func Init(addr string, opts ...Option) (*api.Client, error) {
+	o := defaultOptions()
+	o.apply(opts...)
+
+	if o.config != nil {
+		return api.NewClient(o.config)
+	}
+
+	if addr == "" {
+		return nil, fmt.Errorf("consul address cannot be empty")
+	}
+
+	c, err := api.NewClient(&api.Config{
+		Address:    addr,
+		Scheme:     o.scheme,
+		WaitTime:   o.waitTime,
+		Datacenter: o.datacenter,
+	})
+	if err != nil {
+		return nil, xerror.New(err.Error())
+	}
+	return c, nil
+}

consulcli/consulcli_test.go

+package consulcli
+
+import (
+	"testing"
+	"time"
+
+	"github.com/hashicorp/consul/api"
+)
+
+func TestInit(t *testing.T) {
+	addr := "192.168.3.37:8500"
+	cli, err := Init(addr,
+		WithScheme("http"),
+		WithWaitTime(time.Second*2),
+		WithDatacenter(""),
+		WithToken("your-token"),
+	)
+	t.Log(err, cli)
+
+	cli, err = Init("", WithConfig(&api.Config{
+		Address:    addr,
+		Scheme:     "http",
+		WaitTime:   time.Second * 2,
+		Datacenter: "",
+	}))
+	t.Log(err, cli)
+}

consulcli/option.go

+package consulcli
+
+import (
+	"time"
+
+	"github.com/hashicorp/consul/api"
+)
+
+// Option set the consul client options.
+type Option func(*options)
+
+type options struct {
+	scheme     string
+	waitTime   time.Duration
+	datacenter string
+	token      string
+
+	// if you set this parameter, all fields above are invalid
+	config *api.Config
+}
+
+func defaultOptions() *options {
+	return &options{
+		scheme:   "http",
+		waitTime: time.Second * 5,
+	}
+}
+
+func (o *options) apply(opts ...Option) {
+	for _, opt := range opts {
+		opt(o)
+	}
+}
+
+// WithWaitTime set wait time
+func WithWaitTime(waitTime time.Duration) Option {
+	return func(o *options) {
+		o.waitTime = waitTime
+	}
+}
+
+// WithScheme set scheme
+func WithScheme(scheme string) Option {
+	return func(o *options) {
+		o.scheme = scheme
+	}
+}
+
+// WithDatacenter set datacenter
+func WithDatacenter(datacenter string) Option {
+	return func(o *options) {
+		o.datacenter = datacenter
+	}
+}
+
+// WithToken set token
+func WithToken(token string) Option {
+	return func(o *options) {
+		o.token = token
+	}
+}
+
+// WithConfig set consul config
+func WithConfig(c *api.Config) Option {
+	return func(o *options) {
+		o.config = c
+	}
+}

container/group/example_test.go

+package group
+
+import "fmt"
+
+type Counter struct {
+	Value int
+}
+
+func (c *Counter) Incr() {
+	c.Value++
+}
+
+func ExampleGroup_Get() {
+	group := NewGroup(func() interface{} {
+		fmt.Println("Only Once")
+		return &Counter{}
+	})
+
+	// Create a new Counter
+	group.Get("pass").(*Counter).Incr()
+
+	// Get the created Counter again.
+	group.Get("pass").(*Counter).Incr()
+	// Output:
+	// Only Once
+}
+
+func ExampleGroup_Reset() {
+	group := NewGroup(func() interface{} {
+		return &Counter{}
+	})
+
+	// Reset the new function and clear all created objects.
+	group.Reset(func() interface{} {
+		fmt.Println("reset")
+		return &Counter{}
+	})
+
+	// Create a new Counter
+	group.Get("pass").(*Counter).Incr()
+	// Output:reset
+}

container/group/group.go

+// Package group provides a sample lazy load container.
+// The group only creating a new object not until the object is needed by user.
+// And it will cache all the objects to reduce the creation of object.
+package group
+
+import "sync"
+
+// Group is a lazy load container.
+type Group struct {
+	new  func() interface{}
+	vals map[string]interface{}
+	sync.RWMutex
+}
+
+// NewGroup news a group container.
+func NewGroup(fn func() interface{}) *Group {
+	if fn == nil {
+		panic("container.group: can't assign a nil to the new function")
+	}
+	return &Group{
+		new:  fn,
+		vals: make(map[string]interface{}),
+	}
+}
+
+// Get gets the object by the given key.
+func (g *Group) Get(key string) interface{} {
+	g.RLock()
+	v, ok := g.vals[key]
+	if ok {
+		g.RUnlock()
+		return v
+	}
+	g.RUnlock()
+
+	// slow path for group don`t have specified key value
+	g.Lock()
+	defer g.Unlock()
+	v, ok = g.vals[key]
+	if ok {
+		return v
+	}
+	v = g.new()
+	g.vals[key] = v
+	return v
+}
+
+// Reset resets the new function and deletes all existing objects.
+func (g *Group) Reset(fn func() interface{}) {
+	if fn == nil {
+		panic("container.group: can't assign a nil to the new function")
+	}
+	g.Lock()
+	g.new = fn
+	g.Unlock()
+	g.Clear()
+}
+
+// Clear deletes all objects.
+func (g *Group) Clear() {
+	g.Lock()
+	g.vals = make(map[string]interface{})
+	g.Unlock()
+}

container/group/group_test.go

+package group
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestGroupGet(t *testing.T) {
+	count := 0
+	g := NewGroup(func() interface{} {
+		count++
+		return count
+	})
+	v := g.Get("key_0")
+	if !reflect.DeepEqual(v.(int), 1) {
+		t.Errorf("expect 1, actual %v", v)
+	}
+
+	v = g.Get("key_1")
+	if !reflect.DeepEqual(v.(int), 2) {
+		t.Errorf("expect 2, actual %v", v)
+	}
+
+	v = g.Get("key_0")
+	if !reflect.DeepEqual(v.(int), 1) {
+		t.Errorf("expect 1, actual %v", v)
+	}
+	if !reflect.DeepEqual(count, 2) {
+		t.Errorf("expect count 2, actual %v", count)
+	}
+}
+
+func TestGroupReset(t *testing.T) {
+	g := NewGroup(func() interface{} {
+		return 1
+	})
+	g.Get("key")
+	call := false
+	g.Reset(func() interface{} {
+		call = true
+		return 1
+	})
+
+	length := 0
+	for range g.vals {
+		length++
+	}
+	if !reflect.DeepEqual(length, 0) {
+		t.Errorf("expect length 0, actual %v", length)
+	}
+
+	g.Get("key")
+	if !reflect.DeepEqual(call, true) {
+		t.Errorf("expect call true, actual %v", call)
+	}
+}
+
+func TestGroupClear(t *testing.T) {
+	g := NewGroup(func() interface{} {
+		return 1
+	})
+	g.Get("key")
+	length := 0
+	for range g.vals {
+		length++
+	}
+	if !reflect.DeepEqual(length, 1) {
+		t.Errorf("expect length 1, actual %v", length)
+	}
+
+	g.Clear()
+	length = 0
+	for range g.vals {
+		length++
+	}
+	if !reflect.DeepEqual(length, 0) {
+		t.Errorf("expect length 0, actual %v", length)
+	}
+}

dlock/dlock.go

+// Package dlock provides distributed locking primitives, supports redis and etcd.
+package dlock
+
+import "context"
+
+// Locker is the interface that wraps the basic locking operations.
+type Locker interface {
+	Lock(ctx context.Context) error
+	Unlock(ctx context.Context) error
+	TryLock(ctx context.Context) (bool, error)
+	Close() error
+}

dlock/etcd.go

+package dlock
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.etcd.io/etcd/client/v3/concurrency"
+)
+
+var defaultTTL = 15 // seconds
+
+type EtcdLock struct {
+	session *concurrency.Session
+	mutex   *concurrency.Mutex
+}
+
+// NewEtcd creates a new etcd locker with the given key and ttl.
+func NewEtcd(client *clientv3.Client, key string, ttl int) (Locker, error) {
+	if client == nil {
+		return nil, errors.New("etcd client is nil")
+	}
+
+	if key == "" {
+		return nil, errors.New("key is empty")
+	}
+
+	if ttl <= 0 {
+		ttl = defaultTTL
+	}
+	expiration := time.Duration(ttl) * time.Second
+	ctx, _ := context.WithTimeout(context.Background(), expiration) //nolint
+
+	session, err := concurrency.NewSession(
+		client,
+		concurrency.WithTTL(ttl),
+		concurrency.WithContext(ctx),
+	)
+	if err != nil {
+		return nil, err
+	}
+	mutex := concurrency.NewMutex(session, key)
+
+	locker := &EtcdLock{
+		session: session,
+		mutex:   mutex,
+	}
+
+	return locker, nil
+}
+
+// Lock blocks until the lock is acquired or the context is canceled.
+func (l *EtcdLock) Lock(ctx context.Context) error {
+	return l.mutex.Lock(ctx)
+}
+
+// Unlock releases the lock.
+func (l *EtcdLock) Unlock(ctx context.Context) error {
+	return l.mutex.Unlock(ctx)
+}
+
+// TryLock tries to acquire the lock without blocking.
+func (l *EtcdLock) TryLock(ctx context.Context) (bool, error) {
+	err := l.mutex.TryLock(ctx)
+	if err == nil {
+		return true, nil
+	}
+	if err == concurrency.ErrLocked {
+		return false, nil
+	}
+	return false, err
+}
+
+// Close releases the lock and the etcd session.
+func (l *EtcdLock) Close() error {
+	if l.session != nil {
+		return l.session.Close()
+	}
+	return nil
+}

dlock/etcd_test.go

+package dlock
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"xmall/pkg/etcdcli"
+
+	"go.uber.org/zap"
+)
+
+func TestEtcdLock_TryLock(t *testing.T) {
+	initLocker := func() Locker {
+		return getEtcdLock()
+	}
+	testLockAndUnlock(initLocker, false, t)
+}
+
+func TestEtcdLock_Lock(t *testing.T) {
+	initLocker := func() Locker {
+		return getEtcdLock()
+	}
+	testLockAndUnlock(initLocker, true, t)
+}
+
+func getEtcdLock() Locker {
+	endpoints := []string{"127.0.0.1:2379"}
+	cli, err := etcdcli.Init(endpoints,
+		etcdcli.WithDialTimeout(time.Second*2),
+		etcdcli.WithAuth("", ""),
+		etcdcli.WithAutoSyncInterval(0),
+		etcdcli.WithLog(zap.NewNop()),
+	)
+	if err != nil {
+		fmt.Println(err)
+		return nil
+	}
+
+	locker, err := NewEtcd(cli, "xmall/dlock", 10)
+	if err != nil {
+		fmt.Println(err)
+		return nil
+	}
+
+	return locker
+}

dlock/redis.go

+package dlock
+
+import (
+	"context"
+	"errors"
+
+	"github.com/go-redsync/redsync/v4"
+	"github.com/go-redsync/redsync/v4/redis/goredis/v9"
+	"github.com/redis/go-redis/v9"
+)
+
+// RedisLock implements Locker using Redis.
+type RedisLock struct {
+	mutex *redsync.Mutex
+}
+
+// NewRedisLock creates a new RedisLock.
+func NewRedisLock(client *redis.Client, key string, options ...redsync.Option) (Locker, error) {
+	if client == nil {
+		return nil, errors.New("redis client is nil")
+	}
+	if key == "" {
+		return nil, errors.New("key is empty")
+	}
+	return newLocker(client, key, options...), nil
+}
+
+// NewRedisClusterLock creates a new RedisClusterLock.
+func NewRedisClusterLock(clusterClient *redis.ClusterClient, key string, options ...redsync.Option) (Locker, error) {
+	if clusterClient == nil {
+		return nil, errors.New("cluster redis client is nil")
+	}
+	if key == "" {
+		return nil, errors.New("key is empty")
+	}
+	return newLocker(clusterClient, key, options...), nil
+}
+
+func newLocker(delegate redis.UniversalClient, key string, options ...redsync.Option) Locker {
+	pool := goredis.NewPool(delegate)
+	rs := redsync.New(pool)
+	mutex := rs.NewMutex(key, options...)
+
+	return &RedisLock{
+		mutex: mutex,
+	}
+}
+
+// TryLock tries to acquire the lock without blocking.
+func (l *RedisLock) TryLock(ctx context.Context) (bool, error) {
+	err := l.mutex.TryLockContext(ctx)
+	if err == nil {
+		return true, nil
+	}
+	return false, err
+}
+
+// Lock blocks until the lock is acquired or the context is canceled.
+func (l *RedisLock) Lock(ctx context.Context) error {
+	return l.mutex.LockContext(ctx)
+}
+
+// Unlock releases the lock, if unlocking the key is successful, the key will be automatically deleted
+func (l *RedisLock) Unlock(ctx context.Context) error {
+	_, err := l.mutex.UnlockContext(ctx)
+	return err
+}
+
+// Close no-op for RedisLock.
+func (l *RedisLock) Close() error {
+	return nil
+}

dlock/redis_test.go

+package dlock
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"xmall/pkg/goredis"
+)
+
+func TestRedisLock_TryLock(t *testing.T) {
+	initLocker := func() Locker {
+		return getRedisLock()
+	}
+	testLockAndUnlock(initLocker, false, t)
+}
+
+func TestRedisLock_Lock(t *testing.T) {
+	initLocker := func() Locker {
+		return getRedisLock()
+	}
+	testLockAndUnlock(initLocker, true, t)
+}
+
+func TestClusterRedis_TryLock(t *testing.T) {
+	initLocker := func() Locker {
+		return getClusterRedisLock()
+	}
+	testLockAndUnlock(initLocker, false, t)
+}
+
+func TestClusterRedis_Lock(t *testing.T) {
+	initLocker := func() Locker {
+		return getClusterRedisLock()
+	}
+	testLockAndUnlock(initLocker, true, t)
+}
+
+func getRedisLock() Locker {
+	redisCli, err := goredis.Init("default:123456@127.0.0.1:6379")
+	if err != nil {
+		fmt.Println(err)
+		return nil
+	}
+
+	locker, err := NewRedisLock(redisCli, "test_lock")
+	if err != nil {
+		return nil
+	}
+	return locker
+}
+
+func getClusterRedisLock() Locker {
+	addrs := []string{"127.0.0.1:6380", "127.0.0.1:6381", "127.0.0.1:6382"}
+	clusterClient, err := goredis.InitCluster(addrs, "", "123456")
+	if err != nil {
+		fmt.Println(err)
+		return nil
+	}
+
+	locker, err := NewRedisClusterLock(clusterClient, "test_cluster_lock")
+	if err != nil {
+		return nil
+	}
+	return locker
+}
+
+func testLockAndUnlock(initLocker func() Locker, isBlock bool, t *testing.T) {
+	waitGroup := &sync.WaitGroup{}
+	for i := 1; i <= 10; i++ {
+		waitGroup.Add(1)
+		go func(i int) {
+			defer waitGroup.Done()
+			ctx, _ := context.WithTimeout(context.Background(), time.Second*10)
+			NO := fmt.Sprintf("[NO-%d] ", i)
+
+			locker := initLocker()
+			if locker == nil {
+				t.Log("logger init failed")
+				return
+			}
+
+			var err error
+			var ok bool
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				default:
+				}
+
+				time.Sleep(time.Millisecond * 50)
+				if isBlock {
+					err = locker.Lock(ctx)
+					if err == nil {
+						ok = true
+					}
+				} else {
+					ok, err = locker.TryLock(ctx)
+				}
+				if err != nil {
+					//t.Log(NO+"try lock error:", err)
+					continue
+				}
+				if ok {
+					t.Log(NO + "acquire lock success, and do something")
+					time.Sleep(time.Millisecond * 200)
+					err = locker.Unlock(ctx)
+					if err != nil {
+						return
+					}
+					t.Log(NO + "unlock done")
+					return
+				}
+			}
+		}(i)
+	}
+
+	waitGroup.Wait()
+}

email/email.go

+package xemail
+
+import (
+	"crypto/rand"
+	"crypto/tls"
+	"fmt"
+	"math/big"
+	"xmall/pkg/xerrors/xerror"
+
+	"gopkg.in/gomail.v2"
+)
+
+// 配置结构体 (建议从环境变量或配置文件中读取)
+type Config struct {
+	SMTPServer   string
+	SMTPPort     int
+	SMTPUsername string
+	SMTPPassword string
+	FromEmail    string
+}
+
+func GenerateVerificationCode(length int) (string, error) {
+	max := big.NewInt(10)
+	code := ""
+	for i := 0; i < length; i++ {
+		num, err := rand.Int(rand.Reader, max)
+		if err != nil {
+			return "", err
+		}
+		code += num.String()
+	}
+	return code, nil
+}
+
+func SendVerificationCode(cfg Config, toEmail string, code string) error {
+	m := gomail.NewMessage()
+	m.SetHeader("From", cfg.FromEmail)
+	m.SetHeader("To", toEmail)
+	m.SetHeader("Subject", "[万桩严选商城]验证码")
+	m.SetBody("text/html", fmt.Sprintf(`
+		<html>
+		<body>
+			<h3>验证码通知</h3>
+			<p>您的验证码是：<strong>%s</strong></p>
+			<p>有效期15分钟，请勿泄露给他人</p>
+		</body>
+		</html>
+	`, code))
+	d := gomail.NewDialer(
+		cfg.SMTPServer,
+		cfg.SMTPPort,
+		cfg.SMTPUsername,
+		cfg.SMTPPassword,
+	)
+	d.TLSConfig = &tls.Config{ServerName: cfg.SMTPServer}
+	d.SSL = false
+	if err := d.DialAndSend(m); err != nil {
+		return xerror.New(err.Error())
+	}
+	return nil
+}

encoding/encoding.go

+// Package encoding Provides encoding and decoding of json, protobuf and gob.
+package encoding
+
+import (
+	"encoding"
+	"errors"
+	"reflect"
+	"strings"
+)
+
+var (
+	// ErrNotAPointer .
+	ErrNotAPointer = errors.New("v argument must be a pointer")
+)
+
+// Codec defines the interface gRPC uses to encode and decode messages.  Note
+// that implementations of this interface must be thread safe; a Codec's
+// methods can be called from concurrent goroutines.
+type Codec interface {
+	// Marshal returns the wire format of v.
+	Marshal(v interface{}) ([]byte, error)
+	// Unmarshal parses the wire format into v.
+	Unmarshal(data []byte, v interface{}) error
+	// Name returns the name of the Codec implementation. The returned string
+	// will be used as part of content type in transmission.  The result must be
+	// static; the result cannot change between calls.
+	Name() string
+}
+
+var registeredCodecs = make(map[string]Codec)
+
+// RegisterCodec registers the provided Codec for use with all transport clients and
+// servers.
+//
+// The Codec will be stored and looked up by result of its Name() method, which
+// should match the content-subtype of the encoding handled by the Codec.  This
+// is case-insensitive, and is stored and looked up as lowercase.  If the
+// result of calling Name() is an empty string, RegisterCodec will panic. See
+// Content-Type on
+// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for
+// more details.
+//
+// NOTE: this function must only be called during initialization time (i.e. in
+// an init() function), and is not thread-safe.  If multiple Compressors are
+// registered with the same name, the one registered last will take effect.
+func RegisterCodec(codec Codec) {
+	if codec == nil {
+		panic("cannot register a nil Codec")
+	}
+	if codec.Name() == "" {
+		panic("cannot register Codec with empty string result for Name()")
+	}
+	contentSubtype := strings.ToLower(codec.Name())
+	registeredCodecs[contentSubtype] = codec
+}
+
+// GetCodec gets a registered Codec by content-subtype, or nil if no Codec is
+// registered for the content-subtype.
+//
+// The content-subtype is expected to be lowercase.
+func GetCodec(contentSubtype string) Codec {
+	return registeredCodecs[contentSubtype]
+}
+
+// Encoding definition of coding interfaces
+type Encoding interface {
+	Marshal(v interface{}) ([]byte, error)
+	Unmarshal(data []byte, v interface{}) error
+}
+
+// Marshal encode data
+func Marshal(e Encoding, v interface{}) (data []byte, err error) {
+	if isPointer(v) {
+		bm, ok := v.(encoding.BinaryMarshaler)
+		if ok {
+			return bm.MarshalBinary()
+		}
+	}
+	return e.Marshal(v)
+}
+
+// Unmarshal decode data
+func Unmarshal(e Encoding, data []byte, v interface{}) (err error) {
+	if isPointer(v) {
+		bm, ok := v.(encoding.BinaryUnmarshaler)
+		if ok {
+			return bm.UnmarshalBinary(data)
+		}
+	}
+	return e.Unmarshal(data, v)
+}
+
+func isPointer(data interface{}) bool {
+	switch reflect.ValueOf(data).Kind() {
+	case reflect.Ptr, reflect.Interface:
+		return true
+	default:
+		return false
+	}
+}

encoding/encoding_test.go

+package encoding
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+type obj struct {
+	ID   uint64 `json:"id" example:"101"`
+	Name string `json:"name"`
+}
+
+func xEncoding(e Encoding) error {
+	o1 := &obj{ID: 1, Name: "foo"}
+	data, err := Marshal(e, o1)
+	if err != nil {
+		return err
+	}
+
+	o2 := &obj{}
+	err = Unmarshal(e, data, o2)
+	if err != nil {
+		return err
+	}
+
+	if o1.ID != o2.ID {
+		return errors.New("Unmarshal failed")
+	}
+
+	return nil
+}
+
+func TestEncoding(t *testing.T) {
+	err := xEncoding(GobEncoding{})
+	assert.NoError(t, err)
+
+	err = xEncoding(JSONEncoding{})
+	assert.NoError(t, err)
+
+	err = xEncoding(JSONGzipEncoding{})
+	assert.NoError(t, err)
+
+	err = xEncoding(JSONSnappyEncoding{})
+	assert.NoError(t, err)
+
+	err = xEncoding(MsgPackEncoding{})
+	assert.NoError(t, err)
+}
+
+func TestEncodingError(t *testing.T) {
+	gobE := GobEncoding{}
+	// gob error test
+	err := gobE.Unmarshal([]byte("foo"), nil)
+	assert.Error(t, err)
+
+	jsonE := JSONEncoding{}
+	// json error test
+	err = jsonE.Unmarshal([]byte("foo"), nil)
+	assert.Error(t, err)
+
+	jsonDE := JSONGzipEncoding{}
+	// gzip error test
+	_, err = jsonDE.Marshal(make(chan string))
+	assert.Error(t, err)
+	err = jsonDE.Unmarshal([]byte("foo"), nil)
+	assert.Error(t, err)
+	data, _ := GzipEncode([]byte("foo"))
+	err = jsonDE.Unmarshal(data, make(chan string))
+	assert.Error(t, err)
+	_, err = GzipDecode(nil)
+	assert.Error(t, err)
+
+	jsonSE := JSONSnappyEncoding{}
+	// snappy error test
+	_, err = jsonSE.Marshal(make(chan string))
+	assert.Error(t, err)
+	err = jsonSE.Unmarshal([]byte("foo"), nil)
+	assert.Error(t, err)
+
+	msgE := MsgPackEncoding{}
+	// pack error test
+	err = msgE.Unmarshal([]byte("foo"), nil)
+	assert.Error(t, err)
+}
+
+type codec struct{}
+
+func (c codec) Marshal(v interface{}) ([]byte, error) {
+	return []byte{}, nil
+}
+
+func (c codec) Unmarshal(data []byte, v interface{}) error {
+	return nil
+}
+
+func (c codec) Name() string {
+	return "json"
+}
+
+func TestRegisterCodec(t *testing.T) {
+	RegisterCodec(&codec{})
+	c := GetCodec("json")
+	assert.NotNil(t, c)
+
+	defer func() { recover() }()
+	RegisterCodec(nil)
+}
+
+type codec2 struct{ codec }
+
+func (c codec2) Name() string {
+	return ""
+}
+
+func TestRegisterCodec2(t *testing.T) {
+	defer func() { recover() }()
+	RegisterCodec(&codec2{})
+}
+
+type encoder struct{}
+
+func (e encoder) Marshal(v interface{}) ([]byte, error) {
+	return nil, errors.New("mock Marshal error")
+}
+
+func (e encoder) Unmarshal(data []byte, v interface{}) error {
+	return errors.New("mock Unmarshal error")
+}
+
+func (e encoder) MarshalBinary() (data []byte, err error) {
+	return nil, nil
+}
+
+func (e encoder) UnmarshalBinary(data []byte) error {
+	return nil
+}
+
+func TestMarshal(t *testing.T) {
+	_, err := Marshal(encoder{}, nil)
+	assert.Error(t, err)
+
+	_, err = Marshal(nil, &encoder{})
+	assert.NoError(t, err)
+
+	_, err = Marshal(encoder{}, &encoder{})
+	assert.NoError(t, err)
+}
+
+func TestUnmarshall(t *testing.T) {
+	err := Unmarshal(encoder{}, nil, nil)
+	assert.Error(t, err)
+
+	err = Unmarshal(nil, []byte("foo"), &encoder{})
+	assert.NoError(t, err)
+
+	err = Unmarshal(encoder{}, []byte("foo"), &encoder{})
+	assert.NoError(t, err)
+}
+
+func BenchmarkJsonMarshal(b *testing.B) {
+	a := make([]int, 0, 400)
+	for i := 0; i < 400; i++ {
+		a = append(a, i)
+	}
+	jsonEncoding := JSONEncoding{}
+	for n := 0; n < b.N; n++ {
+		_, err := jsonEncoding.Marshal(a)
+		if err != nil {
+			b.Error(err)
+		}
+	}
+}
+
+func BenchmarkJsonUnmarshal(b *testing.B) {
+	a := make([]int, 0, 400)
+	for i := 0; i < 400; i++ {
+		a = append(a, i)
+	}
+	jsonEncoding := JSONEncoding{}
+	data, err := jsonEncoding.Marshal(a)
+	if err != nil {
+		b.Error(err)
+	}
+	var result []int
+	for n := 0; n < b.N; n++ {
+		err = jsonEncoding.Unmarshal(data, &result)
+		if err != nil {
+			b.Error(err)
+		}
+	}
+}
+
+func BenchmarkMsgpack(b *testing.B) {
+	// run the Fib function b.N times
+	a := make([]int, 400)
+	for i := 0; i < 400; i++ {
+		a = append(a, i)
+	}
+	msgPackEncoding := MsgPackEncoding{}
+	data, err := msgPackEncoding.Marshal(a)
+	if err != nil {
+		b.Error(err)
+	}
+	var result []int
+	for n := 0; n < b.N; n++ {
+		err = msgPackEncoding.Unmarshal(data, &result)
+		if err != nil {
+			b.Error(err)
+		}
+	}
+}

encoding/gob_encoding.go

+package encoding
+
+import (
+	"bytes"
+	"encoding/gob"
+)
+
+// GobEncoding gob encode
+type GobEncoding struct{}
+
+// Marshal gob encode
+func (g GobEncoding) Marshal(v interface{}) ([]byte, error) {
+	var (
+		buffer bytes.Buffer
+	)
+
+	err := gob.NewEncoder(&buffer).Encode(v)
+	return buffer.Bytes(), err
+}
+
+// Unmarshal gob encode
+func (g GobEncoding) Unmarshal(data []byte, value interface{}) error {
+	err := gob.NewDecoder(bytes.NewReader(data)).Decode(value)
+	if err != nil {
+		return err
+	}
+	return nil
+}

encoding/json/json.go

+// Package json is a JSON encoding and decoding.
+package json
+
+import (
+	"encoding/json"
+	"reflect"
+
+	"google.golang.org/protobuf/encoding/protojson"
+	"google.golang.org/protobuf/proto"
+
+	"xmall/pkg/encoding"
+)
+
+// Name is the name registered for the json codec.
+const Name = "json"
+
+var (
+	// MarshalOptions is a configurable JSON format marshaller.
+	MarshalOptions = protojson.MarshalOptions{
+		EmitUnpopulated: true,
+	}
+	// UnmarshalOptions is a configurable JSON format parser.
+	UnmarshalOptions = protojson.UnmarshalOptions{
+		DiscardUnknown: true,
+	}
+)
+
+func init() {
+	encoding.RegisterCodec(codec{})
+}
+
+// codec is a Codec implementation with json.
+type codec struct{}
+
+// Marshal object to data
+func (codec) Marshal(v interface{}) ([]byte, error) {
+	switch m := v.(type) {
+	case json.Marshaler:
+		return m.MarshalJSON()
+	case proto.Message:
+		return MarshalOptions.Marshal(m)
+	default:
+		return json.Marshal(m)
+	}
+}
+
+// Unmarshal data to bytes
+func (codec) Unmarshal(data []byte, v interface{}) error {
+	switch m := v.(type) {
+	case json.Unmarshaler:
+		return m.UnmarshalJSON(data)
+	case proto.Message:
+		return UnmarshalOptions.Unmarshal(data, m)
+	default:
+		rv := reflect.ValueOf(v)
+		for rv := rv; rv.Kind() == reflect.Ptr; {
+			if rv.IsNil() {
+				rv.Set(reflect.New(rv.Type().Elem()))
+			}
+			rv = rv.Elem()
+		}
+		if m, ok := reflect.Indirect(rv).Interface().(proto.Message); ok {
+			return UnmarshalOptions.Unmarshal(data, m)
+		}
+		return json.Unmarshal(data, m)
+	}
+}
+
+// Name get name
+func (codec) Name() string {
+	return Name
+}

encoding/json/json_test.go

+package json
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/protobuf/compiler/protogen"
+	"google.golang.org/protobuf/reflect/protoreflect"
+	"google.golang.org/protobuf/types/pluginpb"
+)
+
+type obj struct {
+	ID   uint64 `json:"id" example:"101"`
+	Name string `json:"name"`
+}
+
+func TestJSON(t *testing.T) {
+	c := codec{}
+
+	name := c.Name()
+	assert.Equal(t, Name, name)
+
+	data, err := c.Marshal(&obj{ID: 1, Name: "foo"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.NotNil(t, data)
+
+	o := new(obj)
+	err = c.Unmarshal(data, o)
+	assert.NoError(t, err)
+	assert.Equal(t, "foo", o.Name)
+}
+
+type obj2 struct {
+	ID   uint64 `json:"id" example:"101"`
+	Name string `json:"name"`
+}
+
+func (o obj2) MarshalJSON() ([]byte, error) {
+	return []byte("test data"), nil
+}
+
+func TestJSON2(t *testing.T) {
+	c := codec{}
+	b, err := c.Marshal(&obj2{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.NotNil(t, b)
+
+	err = c.Unmarshal(b, &obj2{})
+	assert.Error(t, err)
+
+	err = c.Unmarshal(b, obj2{})
+	assert.Error(t, err)
+}
+
+type obj3 struct {
+	ID   uint64 `json:"id" example:"101"`
+	Name string `json:"name"`
+}
+
+func (o obj3) ProtoReflect() protoreflect.Message {
+	req := &pluginpb.CodeGeneratorRequest{}
+	opts := protogen.Options{}
+	gen, _ := opts.New(req)
+
+	return gen.Response().ProtoReflect()
+}
+
+func TestJSON3(t *testing.T) {
+	c := codec{}
+	b, err := c.Marshal(&obj3{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.NotNil(t, b)
+
+	err = c.Unmarshal(b, &obj3{})
+	assert.NoError(t, err)
+}
+
+type obj4 struct {
+	ID   uint64 `json:"id" example:"101"`
+	Name string `json:"name"`
+}
+
+func (o obj4) UnmarshalJSON(bytes []byte) error {
+	return nil
+}
+
+func TestJSON4(t *testing.T) {
+	c := codec{}
+	err := c.Unmarshal(nil, &obj4{})
+	assert.NoError(t, err)
+}

encoding/json_encoding.go

+package encoding
+
+import (
+	"bytes"
+	"compress/gzip"
+	"encoding/json"
+	"fmt"
+	"io"
+
+	"github.com/golang/snappy"
+)
+
+// JSONEncoding json format
+type JSONEncoding struct{}
+
+// Marshal json encode
+func (j JSONEncoding) Marshal(v interface{}) ([]byte, error) {
+	buf, err := json.Marshal(v)
+	return buf, err
+}
+
+// Unmarshal json decode
+func (j JSONEncoding) Unmarshal(data []byte, value interface{}) error {
+	err := json.Unmarshal(data, value)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// JSONGzipEncoding json and gzip
+type JSONGzipEncoding struct{}
+
+// Marshal json encode and gzip
+func (jz JSONGzipEncoding) Marshal(v interface{}) ([]byte, error) {
+	buf, err := json.Marshal(v)
+	if err != nil {
+		return nil, err
+	}
+	// var bufSizeBefore = len(buf)
+
+	buf, err = GzipEncode(buf)
+	// log.Infof("gzip_json_compress_ratio=%d/%d=%.2f", bufSizeBefore, len(buf), float64(bufSizeBefore)/float64(len(buf)))
+	return buf, err
+}
+
+// Unmarshal json encode and gzip
+func (jz JSONGzipEncoding) Unmarshal(data []byte, value interface{}) error {
+	jsonData, err := GzipDecode(data)
+	if err != nil {
+		return err
+	}
+
+	err = json.Unmarshal(jsonData, value)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// GzipEncode encoding
+func GzipEncode(in []byte) ([]byte, error) {
+	var (
+		buffer bytes.Buffer
+		out    []byte
+		err    error
+	)
+	writer, err := gzip.NewWriterLevel(&buffer, gzip.BestCompression)
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = writer.Write(in)
+	if err != nil {
+		err = writer.Close()
+		if err != nil {
+			return out, err
+		}
+		return out, err
+	}
+	err = writer.Close()
+	if err != nil {
+		return out, err
+	}
+
+	return buffer.Bytes(), nil
+}
+
+// GzipDecode decode
+func GzipDecode(in []byte) ([]byte, error) {
+	reader, err := gzip.NewReader(bytes.NewReader(in))
+	if err != nil {
+		var out []byte
+		return out, err
+	}
+	defer func() {
+		err = reader.Close()
+		if err != nil {
+			fmt.Printf("reader close err: %+v", err)
+		}
+	}()
+
+	return io.ReadAll(reader)
+}
+
+// JSONSnappyEncoding json format and snappy compression
+type JSONSnappyEncoding struct{}
+
+// Marshal serialization
+func (s JSONSnappyEncoding) Marshal(v interface{}) (data []byte, err error) {
+	b, err := json.Marshal(v)
+	if err != nil {
+		return nil, err
+	}
+	d := snappy.Encode(nil, b)
+	return d, nil
+}
+
+// Unmarshal deserialization
+func (s JSONSnappyEncoding) Unmarshal(data []byte, value interface{}) error {
+	b, err := snappy.Decode(nil, data)
+	if err != nil {
+		return err
+	}
+
+	return json.Unmarshal(b, value)
+}

encoding/msgpack_encoding.go

+package encoding
+
+import "github.com/vmihailenco/msgpack"
+
+// MsgPackEncoding msgpack format
+type MsgPackEncoding struct{}
+
+// Marshal msgpack encode
+func (mp MsgPackEncoding) Marshal(v interface{}) ([]byte, error) {
+	buf, err := msgpack.Marshal(v)
+	return buf, err
+}
+
+// Unmarshal msgpack decode
+func (mp MsgPackEncoding) Unmarshal(data []byte, value interface{}) error {
+	err := msgpack.Unmarshal(data, value)
+	if err != nil {
+		return err
+	}
+	return nil
+}

encoding/proto/proto.go

+// Package proto is a protobuf encoding and decoding.
+package proto
+
+import (
+	"fmt"
+
+	"google.golang.org/protobuf/proto"
+
+	"xmall/pkg/encoding"
+)
+
+// Name is the name registered for the proto compressor.
+const Name = "proto"
+
+func init() {
+	encoding.RegisterCodec(codec{})
+}
+
+// codec is a Codec implementation with protobuf. It is the default codec for gRPC.
+type codec struct{}
+
+func (codec) Marshal(v interface{}) ([]byte, error) {
+	vv, ok := v.(proto.Message)
+	if !ok {
+		return nil, fmt.Errorf("failed to marshal, message is %T, want proto.Message", v)
+	}
+	return proto.Marshal(vv)
+}
+
+func (codec) Unmarshal(data []byte, v interface{}) error {
+	vv, ok := v.(proto.Message)
+	if !ok {
+		return fmt.Errorf("failed to unmarshal, message is %T, want proto.Message", v)
+	}
+	return proto.Unmarshal(data, vv)
+}
+
+func (codec) Name() string {
+	return Name
+}

encoding/proto/proto_test.go

+package proto
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/protobuf/compiler/protogen"
+	"google.golang.org/protobuf/types/pluginpb"
+)
+
+func TestProto(t *testing.T) {
+	c := codec{}
+
+	name := c.Name()
+	assert.Equal(t, Name, name)
+
+	req := &pluginpb.CodeGeneratorRequest{}
+	opts := protogen.Options{}
+	gen, err := opts.New(req)
+	o1 := gen.Response()
+
+	b, err := c.Marshal(o1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.NotNil(t, b)
+
+	o2 := new(pluginpb.CodeGeneratorRequest)
+	err = c.Unmarshal(b, o2)
+	assert.NoError(t, err)
+}
+
+func TestProtoError(t *testing.T) {
+	c := codec{}
+
+	_, err := c.Marshal(nil)
+	assert.Error(t, err)
+	err = c.Unmarshal(nil, nil)
+	assert.Error(t, err)
+}

errcode/http_error.go

+// Package errcode is used for http and grpc error codes, include system-level error codes and business-level error codes
+package errcode
+
+import (
+	"fmt"
+	"net/http"
+	"strconv"
+	"strings"
+)
+
+// ToHTTPCodeLabel need to convert to standard http code label
+const ToHTTPCodeLabel = "[standard http code]"
+
+var errCodes = map[int]*Error{}
+var httpErrCodes = map[int]string{}
+
+// Error error
+type Error struct {
+	code    int
+	msg     string
+	details []string
+
+	// if true, need to convert to standard http code
+	// use ErrToHTTP and ParseError will set this to true
+	needHTTPCode bool
+}
+
+// NewError create a new error message
+func NewError(code int, msg string, details ...string) *Error {
+	if v, ok := errCodes[code]; ok {
+		panic(fmt.Sprintf(`http error code = %d already exists, please define a new error code,
+msg1 = %s
+msg2 = %s
+`, code, v.Msg(), msg))
+	}
+
+	httpErrCodes[code] = msg
+	e := &Error{code: code, msg: msg, details: details}
+	errCodes[code] = e
+	return e
+}
+
+// Err convert to standard error,
+// if there is a parameter 'msg', it will replace the original message.
+func (e *Error) Err(msg ...string) error {
+	message := e.msg
+	if len(msg) > 0 {
+		message = strings.Join(msg, ", ")
+	}
+
+	if len(e.details) == 0 {
+		return fmt.Errorf("code = %d, msg = %s", e.code, message)
+	}
+	return fmt.Errorf("code = %d, msg = %s, details = %v", e.code, message, e.details)
+}
+
+// ErrToHTTP convert to standard error add ToHTTPCodeLabel to error message,
+// use it if you need to convert to standard HTTP status code,
+// if there is a parameter 'msg', it will replace the original message.
+// Tips: you can call the GetErrorCode function to get the standard HTTP status code.
+func (e *Error) ErrToHTTP(msg ...string) error {
+	message := e.msg
+	if len(msg) > 0 {
+		message = strings.Join(msg, ", ")
+	}
+
+	if len(e.details) == 0 {
+		return fmt.Errorf("code = %d, msg = %s%s", e.code, message, ToHTTPCodeLabel)
+	}
+	return fmt.Errorf("code = %d, msg = %s, details = %v%s", e.code, message, strings.Join(e.details, ", "), ToHTTPCodeLabel)
+}
+
+// Code get error code
+func (e *Error) Code() int {
+	return e.code
+}
+
+// Msg get error code message
+func (e *Error) Msg() string {
+	return e.msg
+}
+
+func (e *Error) Error() string {
+	return e.msg
+}
+
+func (e *Error) ToXerror() (int, string) {
+	return e.code, e.msg
+}
+
+// NeedHTTPCode need to convert to standard http code
+func (e *Error) NeedHTTPCode() bool {
+	return e.needHTTPCode
+}
+
+// Details get error code details
+func (e *Error) Details() []string {
+	return e.details
+}
+
+// WithDetails add error details
+func (e *Error) WithDetails(details ...string) *Error {
+	newError := &Error{code: e.code, msg: e.msg}
+	newError.msg += ", " + strings.Join(details, ", ")
+	return newError
+}
+
+// RewriteMsg rewrite error message
+func (e *Error) RewriteMsg(msg string) *Error {
+	return &Error{code: e.code, msg: msg}
+}
+
+// WithOutMsg out error message
+// Deprecated: use RewriteMsg instead
+func (e *Error) WithOutMsg(msg string) *Error {
+	return &Error{code: e.code, msg: msg}
+}
+
+// WithOutMsgI18n out error message i18n
+// langMsg example:
+//
+//	map[int]map[string]string{
+//			20010: {
+//				"en-US": "login failed",
+//				"zh-CN": "登录失败",
+//			},
+//		}
+//
+// lang BCP 47 code https://learn.microsoft.com/en-us/openspecs/office_standards/ms-oe376/6c085406-a698-4e12-9d4d-c3b0ee3dbc4a
+func (e *Error) WithOutMsgI18n(langMsg map[int]map[string]string, lang string) *Error {
+	if i18nMsg, ok := langMsg[e.Code()]; ok {
+		if msg, ok2 := i18nMsg[lang]; ok2 {
+			return &Error{code: e.code, msg: msg}
+		}
+	}
+
+	return &Error{code: e.code, msg: e.msg}
+}
+
+// ToHTTPCode convert to http error code
+func (e *Error) ToHTTPCode() int {
+	switch e.Code() {
+	case Success.Code():
+		return http.StatusOK
+	case InternalServerError.Code():
+		return http.StatusInternalServerError
+	case InvalidParams.Code():
+		return http.StatusBadRequest
+	case Unauthorized.Code(), PermissionDenied.Code():
+		return http.StatusUnauthorized
+	case TooManyRequests.Code(), LimitExceed.Code():
+		return http.StatusTooManyRequests
+	case Forbidden.Code(), AccessDenied.Code():
+		return http.StatusForbidden
+	case NotFound.Code():
+		return http.StatusNotFound
+	case Conflict.Code(), AlreadyExists.Code():
+		return http.StatusConflict
+	case TooEarly.Code():
+		return http.StatusTooEarly
+	case Timeout.Code(), DeadlineExceeded.Code():
+		return http.StatusRequestTimeout
+	case MethodNotAllowed.Code():
+		return http.StatusMethodNotAllowed
+	case ServiceUnavailable.Code():
+		return http.StatusServiceUnavailable
+	case Unimplemented.Code():
+		return http.StatusNotImplemented
+	case StatusBadGateway.Code():
+		return http.StatusBadGateway
+	}
+
+	return http.StatusInternalServerError
+}
+
+// ParseError parsing out error codes from error messages
+func ParseError(err error) *Error {
+	if err == nil {
+		return Success
+	}
+
+	outError := &Error{
+		code: -1,
+		msg:  "unknown error",
+	}
+
+	splits := strings.Split(err.Error(), ", msg = ")
+	codeStr := strings.ReplaceAll(splits[0], "code = ", "")
+	code, er := strconv.Atoi(codeStr)
+	if er != nil {
+		return outError
+	}
+
+	if e, ok := errCodes[code]; ok {
+		if len(splits) > 1 {
+			outError.code = code
+			outError.msg = splits[1]
+			outError.needHTTPCode = strings.Contains(err.Error(), ToHTTPCodeLabel)
+			return outError
+		}
+		return e
+	}
+
+	return outError
+}
+
+// GetErrorCode get Error code from error returned by http invoke
+func GetErrorCode(err error) int {
+	e := ParseError(err)
+	if e.needHTTPCode {
+		return e.ToHTTPCode()
+	}
+	return e.Code()
+}
+
+// ListHTTPErrCodes list http error codes
+func ListHTTPErrCodes() []ErrInfo {
+	return getErrorInfo(httpErrCodes)
+}
+
+func IsSysDefinedError(code int) bool {
+	_, ok := errCodes[code]
+	return ok
+}

errcode/http_sevice_code.go

+package errcode
+
+// HCode Generate an error code between 200000 and 300000 according to the number
+//
+// http service level error code, Err prefix, example.
+//
+// var (
+// ErrUserCreate = NewError(HCode(1)+1, "failed to create user")		// 200101
+// ErrUserDelete = NewError(HCode(1)+2, "failed to delete user")		// 200102
+// ErrUserUpdate = NewError(HCode(1)+3, "failed to update user")		// 200103
+// ErrUserGet    = NewError(HCode(1)+4, "failed to get user details")	// 200104
+// )
+func HCode(num int) int {
+	if num > 999 || num < 1 {
+		panic("num range must be between 0 to 1000")
+	}
+	return 200000 + num*100
+}

errcode/http_system_code.go

+package errcode
+
+// http system level error code, error code range 10000~20000
+var (
+	Success             = NewError(1, "ok")
+	InvalidParams       = NewError(100001, "Invalid Parameter")
+	Unauthorized        = NewError(100002, "Unauthorized")
+	InternalServerError = NewError(100003, "Internal Server Error")
+	NotFound            = NewError(100004, "Not Found")
+	Timeout             = NewError(100006, "Request Timeout")
+	TooManyRequests     = NewError(100007, "Too Many Requests")
+	Forbidden           = NewError(100008, "Forbidden")
+	LimitExceed         = NewError(100009, "Limit Exceed")
+	DeadlineExceeded    = NewError(100010, "Deadline Exceeded")
+	AccessDenied        = NewError(100011, "Access Denied")
+	MethodNotAllowed    = NewError(100012, "Method Not Allowed")
+	ServiceUnavailable  = NewError(100013, "Service Unavailable")
+
+	Canceled           = NewError(100014, "Canceled")
+	Unknown            = NewError(100015, "Unknown")
+	PermissionDenied   = NewError(100016, "Permission Denied")
+	ResourceExhausted  = NewError(100017, "Resource Exhausted")
+	FailedPrecondition = NewError(100018, "Failed Precondition")
+	Aborted            = NewError(100019, "Aborted")
+	OutOfRange         = NewError(100020, "Out Of Range")
+	Unimplemented      = NewError(100021, "Unimplemented")
+	DataLoss           = NewError(100022, "Data Loss")
+
+	StatusBadGateway  = NewError(100023, "Bad Gateway")
+	IDsNotFoundRecord = NewError(100024, "IDs not found record")
+
+	// Deprecated: use Conflict instead
+	AlreadyExists = NewError(100005, "Already Exists")
+	Conflict      = NewError(100409, "Conflict")
+	TooEarly      = NewError(100425, "Too Early")
+)

errcode/response.go

+package errcode
+
+import (
+	"errors"
+	"net/http"
+	"strconv"
+	"strings"
+
+	"github.com/gin-gonic/gin"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// SkipResponse skip response
+var SkipResponse = errors.New("skip response") //nolint
+
+// Responser response interface
+type Responser interface {
+	Success(ctx *gin.Context, data interface{})
+	ParamError(ctx *gin.Context, err error)
+	Error(ctx *gin.Context, err error) bool
+}
+
+// NewResponser creates a new responser, if isFromRPC=true, it means return from rpc, otherwise default return from http
+func NewResponser(isFromRPC bool, httpErrors []*Error, rpcStatus []*RPCStatus) Responser {
+	httpErrorsMap := make(map[int]*Error)
+	rpcStatusMap := make(map[int]*RPCStatus)
+
+	for _, httpError := range httpErrors {
+		if httpError == nil {
+			continue
+		}
+		httpErrorsMap[httpError.Code()] = httpError
+	}
+	for _, statusError := range rpcStatus {
+		if statusError == nil || statusError.status == nil {
+			continue
+		}
+		rpcStatusMap[int(statusError.ToRPCCode())] = statusError
+		rpcStatusMap[int(statusError.status.Code())] = statusError
+	}
+
+	return &defaultResponse{
+		isFromRPC:  isFromRPC,
+		httpErrors: httpErrorsMap,
+		rpcStatus:  rpcStatusMap,
+	}
+}
+
+type defaultResponse struct {
+	isFromRPC  bool // error comes from grpc, if not, default is from http
+	httpErrors map[int]*Error
+	rpcStatus  map[int]*RPCStatus
+}
+
+func (resp *defaultResponse) response(c *gin.Context, respStatus, code int, msg string, data interface{}) {
+	c.JSON(respStatus, map[string]interface{}{
+		"code": code,
+		"msg":  msg,
+		"data": data,
+	})
+}
+
+// Success response success information
+func (resp *defaultResponse) Success(c *gin.Context, data interface{}) {
+	resp.response(c, http.StatusOK, 0, "ok", data)
+}
+
+// ParamError response parameter error information, does not return an error message
+func (resp *defaultResponse) ParamError(c *gin.Context, _ error) {
+	resp.response(c, http.StatusOK, InvalidParams.Code(), InvalidParams.Msg(), struct{}{})
+}
+
+// Error response error information, if return true, means that the error code is converted to a standard http code,
+// otherwise the return http code is always 200
+func (resp *defaultResponse) Error(c *gin.Context, err error) bool {
+	if resp.isFromRPC {
+		// error from rpc and response the corresponding http code
+		return resp.handleRPCError(c, err)
+	}
+
+	// error from http and response http code
+	return resp.handleHTTPError(c, err)
+}
+
+// error from grpc
+func (resp *defaultResponse) handleRPCError(c *gin.Context, err error) bool {
+	st, _ := status.FromError(err)
+
+	// user defined err, response 200
+	if st.Code() == codes.Unknown {
+		code, msg := parseCodeAndMsg(st.String())
+		if code == -1 {
+			// non-conforming err
+			resp.response(c, http.StatusOK, -1, "unknown error", struct{}{})
+		} else {
+			// err created using NewRPCStatus
+			resp.response(c, http.StatusOK, code, msg, struct{}{})
+		}
+		return false
+	}
+
+	// default error code to http
+	switch st.Code() {
+	case codes.Internal, StatusInternalServerError.status.Code():
+		resp.response(c, http.StatusInternalServerError, http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError), struct{}{})
+		return true
+	case codes.Unavailable, StatusServiceUnavailable.status.Code():
+		resp.response(c, http.StatusServiceUnavailable, http.StatusServiceUnavailable, http.StatusText(http.StatusServiceUnavailable), struct{}{})
+		return true
+	}
+
+	// check if you need to return the standard http code
+	if strings.Contains(st.Message(), ToHTTPCodeLabel) {
+		code := convertToHTTPCode(st.Code())
+		msg := strings.ReplaceAll(st.Message(), ToHTTPCodeLabel, "")
+		resp.response(c, code, int(st.Code()), msg, struct{}{})
+		return true
+	}
+
+	// user defined error code to http
+	if resp.isUserDefinedRPCErrorCode(c, int(st.Code())) {
+		return true
+	}
+
+	// response 200
+	resp.response(c, http.StatusOK, int(st.Code()), st.Message(), struct{}{})
+
+	return false
+}
+
+// error from http
+func (resp *defaultResponse) handleHTTPError(c *gin.Context, err error) bool {
+	e := ParseError(err)
+
+	// default error code to http
+	switch e.Code() {
+	case InternalServerError.Code(), http.StatusInternalServerError:
+		resp.response(c, http.StatusInternalServerError, http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError), struct{}{})
+		return true
+	case ServiceUnavailable.Code(), http.StatusServiceUnavailable:
+		resp.response(c, http.StatusServiceUnavailable, http.StatusServiceUnavailable, http.StatusText(http.StatusServiceUnavailable), struct{}{})
+		return true
+	}
+
+	// user requests to return standard HTTP code, if e.ToHTTPCode() not match, will return of 500
+	if e.needHTTPCode {
+		msg := strings.ReplaceAll(e.msg, ToHTTPCodeLabel, "")
+		resp.response(c, e.ToHTTPCode(), e.code, msg, struct{}{})
+		return true
+	}
+
+	// user defined error code to http
+	if resp.isUserDefinedHTTPErrorCode(c, e.Code()) {
+		return true
+	}
+
+	// response 200
+	resp.response(c, http.StatusOK, e.code, e.msg, struct{}{})
+	return false
+}
+
+func (resp *defaultResponse) isUserDefinedRPCErrorCode(c *gin.Context, errCode int) bool {
+	if v, ok := resp.rpcStatus[errCode]; ok {
+		httpCode := ToHTTPErr(v.status).ToHTTPCode()
+		msg := http.StatusText(httpCode)
+		if msg == "" {
+			msg = "unknown error"
+		}
+		resp.response(c, httpCode, httpCode, msg, struct{}{})
+		return true
+	}
+	return false
+}
+
+func (resp *defaultResponse) isUserDefinedHTTPErrorCode(c *gin.Context, errCode int) bool {
+	if v, ok := resp.httpErrors[errCode]; ok {
+		httpCode := v.ToHTTPCode()
+		msg := http.StatusText(httpCode)
+		if msg == "" {
+			msg = "unknown error"
+		}
+		resp.response(c, httpCode, httpCode, msg, struct{}{})
+		return true
+	}
+	return false
+}
+
+// ToHTTPErr converted to http error
+func ToHTTPErr(st *status.Status) *Error { //nolint
+	switch st.Code() {
+	case StatusSuccess.status.Code(), codes.OK:
+		return Success
+	case StatusInvalidParams.status.Code(), codes.InvalidArgument:
+		return InvalidParams
+	case StatusInternalServerError.status.Code(), codes.Internal:
+		return InternalServerError
+	case StatusUnimplemented.status.Code(), codes.Unimplemented:
+		return Unimplemented
+	case StatusPermissionDenied.status.Code(), codes.PermissionDenied:
+		return PermissionDenied
+	}
+
+	switch st.Code() {
+	case StatusCanceled.status.Code(), codes.Canceled:
+		return Canceled
+	case StatusUnknown.status.Code(), codes.Unknown:
+		return Unknown
+	case StatusDeadlineExceeded.status.Code(), codes.DeadlineExceeded:
+		return DeadlineExceeded
+	case StatusNotFound.status.Code(), codes.NotFound:
+		return NotFound
+	case StatusAlreadyExists.status.Code(), codes.AlreadyExists, StatusConflict.status.Code():
+		return Conflict
+	case StatusResourceExhausted.status.Code(), codes.ResourceExhausted:
+		return ResourceExhausted
+	case StatusFailedPrecondition.status.Code(), codes.FailedPrecondition:
+		return FailedPrecondition
+	case StatusAborted.status.Code(), codes.Aborted:
+		return Aborted
+	case StatusOutOfRange.status.Code(), codes.OutOfRange:
+		return OutOfRange
+	case StatusServiceUnavailable.status.Code(), codes.Unavailable:
+		return ServiceUnavailable
+	case StatusDataLoss.status.Code(), codes.DataLoss:
+		return DataLoss
+	case StatusUnauthorized.status.Code(), codes.Unauthenticated:
+		return Unauthorized
+
+	case StatusAccessDenied.status.Code():
+		return AccessDenied
+	case StatusLimitExceed.status.Code():
+		return LimitExceed
+	case StatusMethodNotAllowed.status.Code():
+		return MethodNotAllowed
+	}
+
+	return &Error{
+		code: int(st.Code()),
+		msg:  st.Message(),
+	}
+}
+
+func parseCodeAndMsg(errStr string) (int, string) {
+	if errStr != "" {
+		ss := strings.Split(errStr, "desc = ")
+		cm := strings.Split(ss[len(ss)-1], "msg = ")
+		if len(cm) == 2 {
+			codeStr := strings.ReplaceAll(cm[0], "code = ", "")
+			codeStr = strings.ReplaceAll(codeStr, ", ", "")
+			code, _ := strconv.Atoi(codeStr)
+			msg := cm[1]
+			return code, msg
+		}
+	}
+	return -1, errStr
+}

errcode/rpc_error.go

+package errcode
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"sort"
+	"strings"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+var grpcErrCodes = map[int]string{}
+
+// RPCStatus rpc status
+type RPCStatus struct {
+	status *status.Status
+}
+
+var statusCodes = map[codes.Code]string{}
+
+// NewRPCStatus create a new rpc status
+func NewRPCStatus(code codes.Code, msg string) *RPCStatus {
+	if v, ok := statusCodes[code]; ok {
+		panic(fmt.Sprintf(`grpc status code = %d already exists, please define a new error code,
+msg1 = %s
+msg2 = %s
+`, code, v, msg))
+	}
+
+	grpcErrCodes[int(code)] = msg
+	statusCodes[code] = msg
+	return &RPCStatus{
+		status: status.New(code, msg),
+	}
+}
+
+// Detail error details
+type Detail struct {
+	key string
+	val interface{}
+}
+
+// String detail key-value
+func (d *Detail) String() string {
+	return fmt.Sprintf("%s: %v", d.key, d.val)
+}
+
+// Any type key value
+func Any(key string, val interface{}) Detail {
+	return Detail{
+		key: key,
+		val: val,
+	}
+}
+
+// Code get code
+func (s *RPCStatus) Code() codes.Code {
+	return s.status.Code()
+}
+
+// Msg get message
+func (s *RPCStatus) Msg() string {
+	return s.status.Message()
+}
+
+// Err return error
+// if there is a parameter 'desc', it will replace the original message
+func (s *RPCStatus) Err(desc ...string) error {
+	if len(desc) > 0 {
+		return status.Errorf(s.status.Code(), "%s", strings.Join(desc, ", "))
+	}
+	return status.Errorf(s.status.Code(), "%s", s.status.Message())
+}
+
+// ErrToHTTP convert to standard error add ToHTTPCodeLabel to error message,
+// usually used when HTTP calls the GRPC API,
+// if there is a parameter 'desc', it will replace the original message.
+func (s *RPCStatus) ErrToHTTP(desc ...string) error {
+	message := s.status.Message()
+	if len(desc) > 0 {
+		message = strings.Join(desc, ", ")
+	}
+	return status.Errorf(s.status.Code(), "%s%s", message, ToHTTPCodeLabel)
+}
+
+// ToRPCErr converted to standard RPC error,
+// use it if you need to convert to standard RPC errors,
+// if there is a parameter 'desc', it will replace the original message.
+func (s *RPCStatus) ToRPCErr(desc ...string) error {
+	switch s.status.Code() {
+	case StatusInvalidParams.status.Code():
+		return toRPCErr(codes.InvalidArgument, desc...)
+	case StatusInternalServerError.status.Code():
+		return toRPCErr(codes.Internal, desc...)
+	}
+
+	switch s.status.Code() {
+	case StatusCanceled.status.Code():
+		return toRPCErr(codes.Canceled, desc...)
+	case StatusUnknown.status.Code():
+		return toRPCErr(codes.Unknown, desc...)
+	case StatusDeadlineExceeded.status.Code():
+		return toRPCErr(codes.DeadlineExceeded, desc...)
+	case StatusNotFound.status.Code():
+		return toRPCErr(codes.NotFound, desc...)
+	case StatusAlreadyExists.status.Code(), StatusConflict.status.Code():
+		return toRPCErr(codes.AlreadyExists, desc...)
+	case StatusPermissionDenied.status.Code():
+		return toRPCErr(codes.PermissionDenied, desc...)
+	case StatusResourceExhausted.status.Code():
+		return toRPCErr(codes.ResourceExhausted, desc...)
+	case StatusFailedPrecondition.status.Code():
+		return toRPCErr(codes.FailedPrecondition, desc...)
+	case StatusAborted.status.Code():
+		return toRPCErr(codes.Aborted, desc...)
+	case StatusOutOfRange.status.Code():
+		return toRPCErr(codes.OutOfRange, desc...)
+	case StatusUnimplemented.status.Code():
+		return toRPCErr(codes.Unimplemented, desc...)
+	case StatusServiceUnavailable.status.Code():
+		return toRPCErr(codes.Unavailable, desc...)
+	case StatusDataLoss.status.Code():
+		return toRPCErr(codes.DataLoss, desc...)
+	case StatusUnauthorized.status.Code():
+		return toRPCErr(codes.Unauthenticated, desc...)
+	case StatusAccessDenied.status.Code():
+		return toRPCErr(codes.PermissionDenied, desc...)
+	case StatusLimitExceed.status.Code():
+		return toRPCErr(codes.ResourceExhausted, desc...)
+	case StatusMethodNotAllowed.status.Code():
+		return toRPCErr(codes.Unimplemented, desc...)
+	}
+
+	return s.status.Err()
+}
+
+func toRPCErr(code codes.Code, descs ...string) error {
+	var desc string
+	if len(descs) > 0 {
+		desc = strings.Join(descs, ", ")
+	} else {
+		desc = code.String()
+	}
+	return status.New(code, desc).Err()
+}
+
+// ToRPCCode converted to standard RPC error code
+func (s *RPCStatus) ToRPCCode() codes.Code {
+	switch s.status.Code() {
+	case StatusInvalidParams.status.Code():
+		return codes.InvalidArgument
+	case StatusInternalServerError.status.Code():
+		return codes.Internal
+	case StatusUnimplemented.status.Code():
+		return codes.Unimplemented
+	}
+
+	switch s.status.Code() {
+	case StatusPermissionDenied.status.Code():
+		return codes.PermissionDenied
+	case StatusCanceled.status.Code():
+		return codes.Canceled
+	case StatusUnknown.status.Code():
+		return codes.Unknown
+	case StatusDeadlineExceeded.status.Code():
+		return codes.DeadlineExceeded
+	case StatusNotFound.status.Code():
+		return codes.NotFound
+	case StatusAlreadyExists.status.Code(), StatusConflict.status.Code():
+		return codes.AlreadyExists
+	case StatusResourceExhausted.status.Code():
+		return codes.ResourceExhausted
+	case StatusFailedPrecondition.status.Code():
+		return codes.FailedPrecondition
+	case StatusAborted.status.Code():
+		return codes.Aborted
+	case StatusOutOfRange.status.Code():
+		return codes.OutOfRange
+	case StatusServiceUnavailable.status.Code():
+		return codes.Unavailable
+	case StatusDataLoss.status.Code():
+		return codes.DataLoss
+	case StatusUnauthorized.status.Code():
+		return codes.Unauthenticated
+	case StatusAccessDenied.status.Code():
+		return codes.PermissionDenied
+	case StatusLimitExceed.status.Code():
+		return codes.ResourceExhausted
+	case StatusMethodNotAllowed.status.Code():
+		return codes.Unimplemented
+	}
+
+	return s.status.Code()
+}
+
+// converted grpc code to http code
+func convertToHTTPCode(code codes.Code) int {
+	switch code {
+	case StatusSuccess.status.Code():
+		return http.StatusOK
+	case codes.InvalidArgument, StatusInvalidParams.status.Code():
+		return http.StatusBadRequest
+	case codes.Internal, StatusInternalServerError.status.Code():
+		return http.StatusInternalServerError
+	case codes.Unimplemented, StatusUnimplemented.status.Code():
+		return http.StatusNotImplemented
+	case codes.NotFound, StatusNotFound.status.Code():
+		return http.StatusNotFound
+	case StatusForbidden.status.Code(), StatusAccessDenied.status.Code():
+		return http.StatusForbidden
+	}
+
+	switch code {
+	case StatusTimeout.status.Code():
+		return http.StatusRequestTimeout
+	case StatusTooManyRequests.status.Code(), StatusLimitExceed.status.Code():
+		return http.StatusTooManyRequests
+	case codes.FailedPrecondition, StatusFailedPrecondition.status.Code():
+		return http.StatusPreconditionFailed
+	case codes.Unavailable, StatusServiceUnavailable.status.Code():
+		return http.StatusServiceUnavailable
+	case codes.Unauthenticated, StatusUnauthorized.status.Code():
+		return http.StatusUnauthorized
+	case codes.PermissionDenied, StatusPermissionDenied.status.Code():
+		return http.StatusUnauthorized
+	case StatusLimitExceed.status.Code():
+		return http.StatusTooManyRequests
+	case StatusMethodNotAllowed.status.Code():
+		return http.StatusMethodNotAllowed
+	case StatusConflict.status.Code():
+		return http.StatusConflict
+	}
+
+	return http.StatusInternalServerError
+}
+
+// GetStatusCode get status code from error returned by RPC invoke
+func GetStatusCode(err error) codes.Code {
+	st, _ := status.FromError(err)
+	return st.Code()
+}
+
+// ErrInfo error info
+type ErrInfo struct {
+	Code int    `json:"code"`
+	Msg  string `json:"msg"`
+}
+
+func getErrorInfo(codeInfo map[int]string) []ErrInfo {
+	var keys []int
+	for key := range codeInfo {
+		keys = append(keys, key)
+	}
+
+	sort.Ints(keys)
+	eis := []ErrInfo{}
+	for _, key := range keys {
+		eis = append(eis, ErrInfo{
+			Code: key,
+			Msg:  codeInfo[key],
+		})
+	}
+
+	return eis
+}
+
+// ListGRPCErrCodes list grpc error codes, http handle func
+func ListGRPCErrCodes(w http.ResponseWriter, _ *http.Request) {
+	eis := getErrorInfo(grpcErrCodes)
+
+	jsonData, err := json.Marshal(&eis)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusOK)
+	_, err = w.Write(jsonData)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+}
+
+// ShowConfig show config info
+// @Summary show config info
+// @Description show config info
+// @Tags system
+// @Accept  json
+// @Produce  json
+// @Router /config [get]
+func ShowConfig(jsonData []byte) func(w http.ResponseWriter, r *http.Request) {
+	return func(w http.ResponseWriter, r *http.Request) {
+		//w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusOK)
+		_, err := w.Write(jsonData)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+	}
+}

errcode/rpc_error_test.go

+package errcode
+
+import (
+	"fmt"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/grpc/status"
+
+	"xmall/pkg/utils"
+)
+
+var rpcStatus = []*RPCStatus{
+	StatusSuccess,
+	StatusCanceled,
+	StatusUnknown,
+	StatusInvalidParams,
+	StatusDeadlineExceeded,
+	StatusNotFound,
+	StatusAlreadyExists,
+	StatusPermissionDenied,
+	StatusResourceExhausted,
+	StatusFailedPrecondition,
+	StatusAborted,
+	StatusOutOfRange,
+	StatusUnimplemented,
+	StatusInternalServerError,
+	StatusServiceUnavailable,
+	StatusDataLoss,
+	StatusUnauthorized,
+	StatusTimeout,
+	StatusTooManyRequests,
+	StatusForbidden,
+	StatusLimitExceed,
+	StatusMethodNotAllowed,
+	StatusAccessDenied,
+	StatusConflict,
+}
+
+func TestRPCStatus(t *testing.T) {
+	st := NewRPCStatus(401101, "something is wrong")
+	err := st.Err()
+	assert.Error(t, err)
+	err = st.Err("another thing is wrong")
+
+	s, ok := status.FromError(err)
+	assert.True(t, ok)
+	assert.Equal(t, s.Code(), st.Code())
+	assert.Equal(t, s.Message(), "another thing is wrong")
+
+	code := st.Code()
+	assert.Equal(t, int(code), 401101)
+	msg := st.Msg()
+	assert.Equal(t, msg, "something is wrong")
+
+	defer func() {
+		if e := recover(); e != nil {
+			t.Log(e)
+		}
+	}()
+	NewRPCStatus(401101, "something is wrong 2")
+}
+
+func TestToRPCCode(t *testing.T) {
+	var codes []string
+	for _, s := range rpcStatus {
+		codes = append(codes, s.ToRPCCode().String())
+	}
+	t.Log(codes)
+
+	var errors []error
+	for i, s := range rpcStatus {
+		if i%2 == 0 {
+			errors = append(errors, s.ToRPCErr())
+			continue
+		}
+		errors = append(errors, s.ToRPCErr(s.status.Message()))
+	}
+	t.Log(errors)
+
+	codeInt := []int{}
+	for _, s := range rpcStatus {
+		codeInt = append(codeInt, ToHTTPErr(s.status).code)
+	}
+	t.Log(codeInt)
+}
+
+func TestConvertToHTTPCode(t *testing.T) {
+	var codes []int
+	for _, s := range rpcStatus {
+		codes = append(codes, convertToHTTPCode(s.Code()))
+	}
+	t.Log(codes)
+
+}
+
+func TestGetStatusCode(t *testing.T) {
+	t.Log(GetStatusCode(fmt.Errorf("reason for error")))
+
+	for _, s := range rpcStatus {
+		t.Log(s.Code(), "|",
+			GetStatusCode(s.Err()),
+			GetStatusCode(s.Err("reason for error")), "|",
+
+			GetStatusCode(s.ToRPCErr()),
+			GetStatusCode(s.ToRPCErr("reason for error")), "|",
+
+			GetStatusCode(s.ErrToHTTP()),
+			GetStatusCode(s.ErrToHTTP("reason for error")),
+		)
+	}
+}
+
+func TestRCode(t *testing.T) {
+	code := RCode(1)
+	t.Log("error code is", int(code))
+
+	defer func() {
+		recover()
+	}()
+	code = RCode(1001)
+	t.Log("error code is", int(code))
+}
+
+func TestHandlers(t *testing.T) {
+	serverAddr, requestAddr := utils.GetLocalHTTPAddrPairs()
+
+	gin.SetMode(gin.ReleaseMode)
+	r := gin.New()
+	r.GET("/codes", gin.WrapF(ListGRPCErrCodes))
+	r.GET("/config", gin.WrapF(ShowConfig([]byte(`{"foo": "bar"}`))))
+
+	go func() {
+		_ = r.Run(serverAddr)
+	}()
+
+	time.Sleep(time.Millisecond * 200)
+	resp, err := http.Get(requestAddr + "/codes")
+	assert.NoError(t, err)
+	assert.NotNil(t, resp)
+	resp, err = http.Get(requestAddr + "/config")
+	assert.NoError(t, err)
+	assert.NotNil(t, resp)
+	time.Sleep(time.Second)
+}

errcode/rpc_sevice_code.go

+package errcode
+
+import "google.golang.org/grpc/codes"
+
+// RCode Generate an error code between 400000 and 500000 according to the number
+//
+// rpc service level error code, status prefix, example.
+//
+//	var (
+//		StatusUserCreate = NewRPCStatus(RCode(1)+1, "failed to create user")		// 400101
+//		StatusUserDelete = NewRPCStatus(RCode(1)+2, "failed to delete user")		// 400102
+//		StatusUserUpdate = NewRPCStatus(RCode(1)+3, "failed to update user")		// 400103
+//		StatusUserGet    = NewRPCStatus(RCode(1)+4, "failed to get user details")	// 400104
+//	)
+func RCode(num int) codes.Code {
+	if num > 999 || num < 1 {
+		panic("NO range must be between 0 to 1000")
+	}
+	return codes.Code(400000 + num*100)
+}

errcode/rpc_system_code.go

+package errcode
+
+// rpc system level error code with status prefix, error code range 30000~40000
+var (
+	StatusSuccess = NewRPCStatus(0, "ok")
+
+	StatusCanceled            = NewRPCStatus(300001, "Canceled")
+	StatusUnknown             = NewRPCStatus(300002, "Unknown")
+	StatusInvalidParams       = NewRPCStatus(300003, "Invalid Parameter")
+	StatusDeadlineExceeded    = NewRPCStatus(300004, "Deadline Exceeded")
+	StatusNotFound            = NewRPCStatus(300005, "Not Found")
+	StatusAlreadyExists       = NewRPCStatus(300006, "Already Exists")
+	StatusPermissionDenied    = NewRPCStatus(300007, "Permission Denied")
+	StatusResourceExhausted   = NewRPCStatus(300008, "Resource Exhausted")
+	StatusFailedPrecondition  = NewRPCStatus(300009, "Failed Precondition")
+	StatusAborted             = NewRPCStatus(300010, "Aborted")
+	StatusOutOfRange          = NewRPCStatus(300011, "Out Of Range")
+	StatusUnimplemented       = NewRPCStatus(300012, "Unimplemented")
+	StatusInternalServerError = NewRPCStatus(300013, "Internal Server Error")
+	StatusServiceUnavailable  = NewRPCStatus(300014, "Service Unavailable")
+	StatusDataLoss            = NewRPCStatus(300015, "Data Loss")
+	StatusUnauthorized        = NewRPCStatus(300016, "Unauthorized")
+
+	StatusTimeout          = NewRPCStatus(300017, "Request Timeout")
+	StatusTooManyRequests  = NewRPCStatus(300018, "Too Many Requests")
+	StatusForbidden        = NewRPCStatus(300019, "Forbidden")
+	StatusLimitExceed      = NewRPCStatus(300020, "Limit Exceed")
+	StatusMethodNotAllowed = NewRPCStatus(300021, "Method Not Allowed")
+	StatusAccessDenied     = NewRPCStatus(300022, "Access Denied")
+	StatusConflict         = NewRPCStatus(300023, "Conflict")
+)

etcdcli/etcdcli.go

+// Package etcdcli is use for connecting to the etcd service
+package etcdcli
+
+import (
+	"fmt"
+	"time"
+	"xmall/pkg/xerrors/xerror"
+
+	clientv3 "go.etcd.io/etcd/client/v3"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+// Init connecting to the etcd service
+// Note: If the WithConfig(*clientv3.Config) parameter is set, the endpoints parameter is ignored!
+func Init(endpoints []string, opts ...Option) (*clientv3.Client, error) {
+	o := defaultOptions()
+	o.apply(opts...)
+
+	if o.config != nil {
+		return clientv3.New(*o.config)
+	}
+
+	if len(endpoints) == 0 {
+		return nil, fmt.Errorf("etcd endpoints cannot be empty")
+	}
+
+	conf := clientv3.Config{
+		Endpoints:            endpoints,
+		DialTimeout:          o.dialTimeout,
+		DialKeepAliveTime:    20 * time.Second,
+		DialKeepAliveTimeout: 10 * time.Second,
+		AutoSyncInterval:     o.autoSyncInterval,
+		Logger:               o.logger,
+		Username:             o.username,
+		Password:             o.password,
+	}
+
+	if !o.isSecure {
+		conf.DialOptions = append(conf.DialOptions, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	} else {
+		cred, err := credentials.NewClientTLSFromFile(o.certFile, o.serverNameOverride)
+		if err != nil {
+			return nil, fmt.Errorf("NewClientTLSFromFile error: %v", err)
+		}
+		conf.DialOptions = append(conf.DialOptions, grpc.WithTransportCredentials(cred))
+	}
+
+	cli, err := clientv3.New(conf)
+	if err != nil {
+		return nil, xerror.Errorf("connecting to the etcd service error: %v", err)
+	}
+
+	return cli, nil
+}

etcdcli/etcdcli_test.go

+package etcdcli
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.uber.org/zap"
+)
+
+func TestInit(t *testing.T) {
+	endpoints := []string{"192.168.3.37:2379"}
+	cli, err := Init(endpoints,
+		WithDialTimeout(time.Second*2),
+		WithAuth("", ""),
+		WithAutoSyncInterval(0),
+		WithLog(zap.NewNop()),
+	)
+	t.Log(err, cli)
+
+	cli, err = Init(nil, WithConfig(&clientv3.Config{
+		Endpoints:   endpoints,
+		DialTimeout: time.Second * 2,
+		Username:    "",
+		Password:    "",
+	}))
+	t.Log(err, cli)
+
+	// test error
+	_, err = Init(endpoints,
+		WithDialTimeout(time.Second),
+		WithSecure("foo", "notfound.crt"))
+	assert.Error(t, err)
+	endpoints = nil
+	_, err = Init(endpoints)
+	assert.Error(t, err)
+}

etcdcli/option.go

+package etcdcli
+
+import (
+	"time"
+
+	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.uber.org/zap"
+)
+
+// Option set the etcd client options.
+type Option func(*options)
+
+type options struct {
+	dialTimeout time.Duration // connection timeout, unit(second)
+
+	username string
+	password string
+
+	isSecure           bool
+	serverNameOverride string // etcd domain
+	certFile           string // path to certificate file
+
+	autoSyncInterval time.Duration // automatic synchronization of member list intervals
+	logger           *zap.Logger
+
+	// if you set this parameter, all fields above are invalid
+	config *clientv3.Config
+}
+
+func defaultOptions() *options {
+	return &options{
+		dialTimeout: time.Second * 5,
+	}
+}
+
+func (o *options) apply(opts ...Option) {
+	for _, opt := range opts {
+		opt(o)
+	}
+}
+
+// WithDialTimeout set dial timeout
+func WithDialTimeout(duration time.Duration) Option {
+	return func(o *options) {
+		o.dialTimeout = duration
+	}
+}
+
+// WithAuth set authentication
+func WithAuth(username string, password string) Option {
+	return func(o *options) {
+		o.username = username
+		o.password = password
+	}
+}
+
+// WithSecure set tls
+func WithSecure(serverNameOverride string, certFile string) Option {
+	return func(o *options) {
+		o.isSecure = true
+		o.serverNameOverride = serverNameOverride
+		o.certFile = certFile
+	}
+}
+
+// WithAutoSyncInterval set auto sync interval value
+func WithAutoSyncInterval(duration time.Duration) Option {
+	return func(o *options) {
+		o.autoSyncInterval = duration
+	}
+}
+
+// WithLog set logger
+func WithLog(l *zap.Logger) Option {
+	return func(o *options) {
+		o.logger = l
+	}
+}
+
+// WithConfig set etcd client config
+func WithConfig(c *clientv3.Config) Option {
+	return func(o *options) {
+		o.config = c
+	}
+}

eventbus/eventbus.go

+package eventbus
+
+import (
+	"fmt"
+	"reflect"
+	"sync"
+)
+
+type BusSubscriber interface {
+	Subscribe(topic string, fn interface{}) error
+	SubscribeAsync(topic string, fn interface{}, transactional bool) error
+	SubscribeOnce(topic string, fn interface{}) error
+	SubscribeOnceAsync(topic string, fn interface{}) error
+	Unsubscribe(topic string, handler interface{}) error
+}
+
+type BusPublisher interface {
+	Publish(topic string, args ...interface{})
+}
+
+type BusController interface {
+	HasCallback(topic string) bool
+	WaitAsync()
+}
+
+type Bus interface {
+	BusController
+	BusSubscriber
+	BusPublisher
+}
+
+type EventBus struct {
+	handlers map[string][]*eventHandler
+	lock     sync.Mutex
+	wg       sync.WaitGroup
+}
+
+type eventHandler struct {
+	callBack      reflect.Value
+	flagOnce      bool
+	async         bool
+	transactional bool
+	sync.Mutex
+}
+
+func New() Bus {
+	b := &EventBus{
+		make(map[string][]*eventHandler),
+		sync.Mutex{},
+		sync.WaitGroup{},
+	}
+	return Bus(b)
+}
+
+func (bus *EventBus) doSubscribe(topic string, fn interface{}, handler *eventHandler) error {
+	bus.lock.Lock()
+	defer bus.lock.Unlock()
+	if !(reflect.TypeOf(fn).Kind() == reflect.Func) {
+		return fmt.Errorf("%s is not of type reflect.Func", reflect.TypeOf(fn).Kind())
+	}
+	bus.handlers[topic] = append(bus.handlers[topic], handler)
+	return nil
+}
+
+func (bus *EventBus) Subscribe(topic string, fn interface{}) error {
+	return bus.doSubscribe(topic, fn, &eventHandler{
+		reflect.ValueOf(fn), false, false, false, sync.Mutex{},
+	})
+}
+
+func (bus *EventBus) SubscribeAsync(topic string, fn interface{}, transactional bool) error {
+	return bus.doSubscribe(topic, fn, &eventHandler{
+		reflect.ValueOf(fn), false, true, transactional, sync.Mutex{},
+	})
+}
+
+func (bus *EventBus) SubscribeOnce(topic string, fn interface{}) error {
+	return bus.doSubscribe(topic, fn, &eventHandler{
+		reflect.ValueOf(fn), true, false, false, sync.Mutex{},
+	})
+}
+
+func (bus *EventBus) SubscribeOnceAsync(topic string, fn interface{}) error {
+	return bus.doSubscribe(topic, fn, &eventHandler{
+		reflect.ValueOf(fn), true, true, false, sync.Mutex{},
+	})
+}
+
+func (bus *EventBus) HasCallback(topic string) bool {
+	bus.lock.Lock()
+	defer bus.lock.Unlock()
+	_, ok := bus.handlers[topic]
+	if ok {
+		return len(bus.handlers[topic]) > 0
+	}
+	return false
+}
+
+func (bus *EventBus) Unsubscribe(topic string, handler interface{}) error {
+	bus.lock.Lock()
+	defer bus.lock.Unlock()
+	if _, ok := bus.handlers[topic]; ok && len(bus.handlers[topic]) > 0 {
+		bus.removeHandler(topic, bus.findHandlerIdx(topic, reflect.ValueOf(handler)))
+		return nil
+	}
+	return fmt.Errorf("topic %s doesn't exist", topic)
+}
+
+func (bus *EventBus) Publish(topic string, args ...interface{}) {
+	bus.lock.Lock()
+	defer bus.lock.Unlock()
+	if handlers, ok := bus.handlers[topic]; ok && 0 < len(handlers) {
+		copyHandlers := make([]*eventHandler, len(handlers))
+		copy(copyHandlers, handlers)
+		for i, handler := range copyHandlers {
+			if handler.flagOnce {
+				bus.removeHandler(topic, i)
+			}
+			if !handler.async {
+				bus.doPublish(handler, args...)
+			} else {
+				bus.wg.Add(1)
+				if handler.transactional {
+					bus.lock.Unlock()
+					handler.Lock()
+					bus.lock.Lock()
+				}
+				go func() {
+					bus.doPublishAsync(handler, args...)
+				}()
+			}
+		}
+	}
+}
+
+func (bus *EventBus) doPublish(handler *eventHandler, args ...interface{}) {
+	passedArguments := bus.setUpPublish(handler, args...)
+	handler.callBack.Call(passedArguments)
+}
+
+func (bus *EventBus) doPublishAsync(handler *eventHandler, args ...interface{}) {
+	defer bus.wg.Done()
+	if handler.transactional {
+		defer handler.Unlock()
+	}
+	bus.doPublish(handler, args...)
+}
+
+func (bus *EventBus) removeHandler(topic string, idx int) {
+	if _, ok := bus.handlers[topic]; !ok {
+		return
+	}
+	l := len(bus.handlers[topic])
+
+	if !(0 <= idx && idx < l) {
+		return
+	}
+
+	copy(bus.handlers[topic][idx:], bus.handlers[topic][idx+1:])
+	bus.handlers[topic][l-1] = nil
+	bus.handlers[topic] = bus.handlers[topic][:l-1]
+}
+
+func (bus *EventBus) findHandlerIdx(topic string, callback reflect.Value) int {
+	if _, ok := bus.handlers[topic]; ok {
+		for idx, handler := range bus.handlers[topic] {
+			if handler.callBack.Type() == callback.Type() &&
+				handler.callBack.Pointer() == callback.Pointer() {
+				return idx
+			}
+		}
+	}
+	return -1
+}
+
+func (bus *EventBus) setUpPublish(callback *eventHandler, args ...interface{}) []reflect.Value {
+	funcType := callback.callBack.Type()
+	passedArguments := make([]reflect.Value, len(args))
+	for i, v := range args {
+		if v == nil {
+			passedArguments[i] = reflect.New(funcType.In(i)).Elem()
+		} else {
+			passedArguments[i] = reflect.ValueOf(v)
+		}
+	}
+
+	return passedArguments
+}
+
+func (bus *EventBus) WaitAsync() {
+	bus.wg.Wait()
+}

gin/frontend/frontend.go

+// Package frontend embeds the frontend static file and adds routing.
+package frontend
+
+import (
+	"embed"
+	"fmt"
+	"io/fs"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"runtime"
+	"strings"
+	"time"
+
+	"github.com/gin-gonic/gin"
+)
+
+// Note: the front-end static files must all be in the same directory, i.e. htmlDir
+
+// when customizing customAddr, read static resources and save them to this directory.
+var targetDir = "frontend"
+
+// FrontEnd is the frontend router configuration
+type FrontEnd struct {
+	// embed.FS static resources.
+	staticFS embed.FS
+	// must be set to false if cross-host access is required, otherwise true
+	isUseEmbedFS bool
+	// directory where index.html is located, e.g. web/html
+	htmlDir string
+	// configuration file in which js requests the address of the back-end api request, e.g. config.js
+	configFile string
+	// modify config
+	modifyConfigFn func(content []byte) []byte
+}
+
+// New create a new frontend
+func New(staticFS embed.FS, isUseEmbedFS bool, htmlDir string, configFile string, modifyConfigFn func(content []byte) []byte) *FrontEnd {
+	htmlDir = strings.Trim(htmlDir, "/")
+	return &FrontEnd{
+		staticFS:       staticFS,
+		isUseEmbedFS:   isUseEmbedFS,
+		htmlDir:        htmlDir,
+		configFile:     configFile,
+		modifyConfigFn: modifyConfigFn,
+	}
+}
+
+// SetRouter set frontend router
+func (f *FrontEnd) SetRouter(r *gin.Engine) error {
+	routerPath := fmt.Sprintf("%s/index.html", f.htmlDir)
+	// solve vue using history route 404 problem
+	r.NoRoute(browserRefreshFS(f.staticFS, routerPath))
+
+	relativePath := fmt.Sprintf("/%s/*filepath", f.htmlDir)
+
+	// use embed file
+	if f.isUseEmbedFS {
+		r.GET(relativePath, func(c *gin.Context) {
+			staticServer := http.FileServer(http.FS(f.staticFS))
+			staticServer.ServeHTTP(c.Writer, c.Request)
+		})
+		return nil
+	}
+
+	// use local file
+	err := f.saveFSToLocal()
+	if err != nil {
+		return err
+	}
+	r.GET(relativePath, func(c *gin.Context) {
+		localFileDir := filepath.Join(targetDir, f.htmlDir)
+		filePath := c.Param("filepath")
+		c.File(localFileDir + filePath)
+	})
+
+	return nil
+}
+
+func (f *FrontEnd) saveFSToLocal() error {
+	_ = os.RemoveAll(filepath.Join(targetDir, f.htmlDir))
+	time.Sleep(time.Millisecond * 10)
+
+	// Walk through the embedded filesystem
+	return fs.WalkDir(f.staticFS, ".", func(path string, d fs.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+
+		// Create the corresponding directory structure locally
+		localPath := filepath.Join(targetDir, path)
+		if d.IsDir() {
+			err := os.MkdirAll(localPath, 0755)
+			if err != nil {
+				return err
+			}
+		} else {
+			// Read the file from the embedded filesystem
+			content, err := fs.ReadFile(f.staticFS, path)
+			if err != nil {
+				return err
+			}
+
+			// replace config address
+			if path == f.configFile {
+				content = f.modifyConfigFn(content)
+			}
+
+			// Save the content to the local file
+			err = os.WriteFile(localPath, content, 0644)
+			if err != nil {
+				return err
+			}
+		}
+
+		return nil
+	})
+}
+
+// solve vue using history route 404 problem, for embed.FS
+func browserRefreshFS(efs embed.FS, path string) func(c *gin.Context) {
+	return func(c *gin.Context) {
+		accept := c.Request.Header.Get("Accept")
+		flag := strings.Contains(accept, "text/html")
+		if flag {
+			content, err := efs.ReadFile(path)
+			if err != nil {
+				c.Writer.WriteHeader(404)
+				_, _ = c.Writer.WriteString("Not Found")
+				return
+			}
+			c.Writer.WriteHeader(200)
+			c.Writer.Header().Add("Accept", "text/html")
+			_, _ = c.Writer.Write(content)
+			c.Writer.Flush()
+		}
+	}
+}
+
+// AutoOpenBrowser auto open browser
+func AutoOpenBrowser(visitURL string) error {
+	var cmd string
+	var args []string
+
+	switch runtime.GOOS {
+	case "windows":
+		cmd = "cmd"
+		args = []string{"/c", "start"}
+	case "darwin":
+		cmd = "open"
+	default: // "linux", "freebsd", "openbsd", "netbsd"
+		cmd = "xdg-open"
+	}
+
+	args = append(args, visitURL)
+	return exec.Command(cmd, args...).Start()
+}

gin/handlerfunc/common.go

+// Package handlerfunc is used for public http request handler.
+package handlerfunc
+
+import (
+	"embed"
+	"net/http"
+	"os"
+	"strings"
+
+	"github.com/gin-gonic/gin"
+
+	"xmall/pkg/errcode"
+	"xmall/pkg/utils"
+)
+
+// CheckHealthReply check health result
+type CheckHealthReply struct {
+	Status   string `json:"status"`
+	Hostname string `json:"hostname"`
+}
+
+// CheckHealth
+// @Summary check health 健康检查
+// @Description check health 健康检查
+// @Tags system
+// @Accept  json
+// @Produce  json
+// @Success 200 {object} CheckHealthReply{}
+// @Router /health [get]
+func CheckHealth(c *gin.Context) {
+	c.JSON(http.StatusOK, CheckHealthReply{Status: "UP", Hostname: utils.GetHostname()})
+}
+
+// Ping ping
+// @Summary ping
+// @Description ping
+// @Tags system
+// @Accept  json
+// @Produce  json
+// @Router /ping [get]
+func Ping(c *gin.Context) {
+	c.JSON(http.StatusOK, gin.H{})
+}
+
+// ListCodes
+// @Summary 错误码信息
+// @Description 错误码信息
+// @Tags system
+// @Accept  json
+// @Produce  json
+// @Router /codes [get]
+func ListCodes(c *gin.Context) {
+	c.JSON(http.StatusOK, errcode.ListHTTPErrCodes())
+}
+
+// BrowserRefresh solve vue using history route 404 problem, for system file
+func BrowserRefresh(path string) func(c *gin.Context) {
+	return func(c *gin.Context) {
+		accept := c.Request.Header.Get("Accept")
+		flag := strings.Contains(accept, "text/html")
+		if flag {
+			content, err := os.ReadFile(path)
+			if err != nil {
+				c.Writer.WriteHeader(404)
+				_, _ = c.Writer.WriteString("Not Found")
+				return
+			}
+			c.Writer.WriteHeader(200)
+			c.Writer.Header().Add("Accept", "text/html")
+			_, _ = c.Writer.Write(content)
+			c.Writer.Flush()
+		}
+	}
+}
+
+// BrowserRefreshFS solve vue using history route 404 problem, for embed.FS
+func BrowserRefreshFS(fs embed.FS, path string) func(c *gin.Context) {
+	return func(c *gin.Context) {
+		accept := c.Request.Header.Get("Accept")
+		flag := strings.Contains(accept, "text/html")
+		if flag {
+			content, err := fs.ReadFile(path)
+			if err != nil {
+				c.Writer.WriteHeader(404)
+				_, _ = c.Writer.WriteString("Not Found")
+				return
+			}
+			c.Writer.WriteHeader(200)
+			c.Writer.Header().Add("Accept", "text/html")
+			_, _ = c.Writer.Write(content)
+			c.Writer.Flush()
+		}
+	}
+}

gin/middleware/auth.go

+// Package middleware is gin middleware plugin.
+package middleware
+
+import (
+	"github.com/gin-gonic/gin"
+
+	"xmall/pkg/errcode"
+	"xmall/pkg/gin/response"
+	ctxUtil "xmall/pkg/gin/xctx"
+	"xmall/pkg/jwt"
+	"xmall/pkg/logger"
+)
+
+const (
+	// HeaderAuthorizationKey http header authorization key
+	HeaderAuthorizationKey = "Authorization"
+)
+
+type jwtOptions struct {
+	isSwitchHTTPCode bool
+	verify           VerifyFn // verify function, only use in Auth
+}
+
+// JwtOption set the jwt options.
+type JwtOption func(*jwtOptions)
+
+func (o *jwtOptions) apply(opts ...JwtOption) {
+	for _, opt := range opts {
+		opt(o)
+	}
+}
+
+func defaultJwtOptions() *jwtOptions {
+	return &jwtOptions{
+		isSwitchHTTPCode: false,
+		verify:           nil,
+	}
+}
+
+// WithSwitchHTTPCode switch to http code
+func WithSwitchHTTPCode() JwtOption {
+	return func(o *jwtOptions) {
+		o.isSwitchHTTPCode = true
+	}
+}
+
+// WithVerify set verify function
+func WithVerify(verify VerifyFn) JwtOption {
+	return func(o *jwtOptions) {
+		o.verify = verify
+	}
+}
+
+func responseUnauthorized(c *gin.Context, isSwitchHTTPCode bool) {
+	if isSwitchHTTPCode {
+		response.Out(c, errcode.Unauthorized)
+	} else {
+		response.ErrorE(c, errcode.Unauthorized)
+	}
+}
+
+// -------------------------------------------------------------------------------------------
+
+// VerifyFn verify function, tokenTail10 is the last 10 characters of the token.
+type VerifyFn func(claims *jwt.Claims, tokenTail10 string, c *gin.Context) error
+
+// Auth authorization
+func Auth(opts ...JwtOption) gin.HandlerFunc {
+	o := defaultJwtOptions()
+	o.apply(opts...)
+
+	return func(c *gin.Context) {
+		authorization := c.GetHeader(HeaderAuthorizationKey)
+		if authorization == "" || len(authorization) < 10 {
+			responseUnauthorized(c, o.isSwitchHTTPCode)
+			c.Abort()
+			return
+		}
+		// if len(authorization) < 150 {
+		// 	logger.Warn("authorization is illegal")
+		// 	responseUnauthorized(c, o.isSwitchHTTPCode)
+		// 	c.Abort()
+		// 	return
+		// }
+
+		token := authorization[7:] // remove Bearer prefix
+		claims, err := jwt.ParseToken(token)
+		if err != nil {
+			logger.Warn("ParseToken error", logger.Err(err))
+			responseUnauthorized(c, o.isSwitchHTTPCode)
+			c.Abort()
+			return
+		}
+
+		if o.verify != nil {
+			tokenTail10 := token[len(token)-10:]
+			if err = o.verify(claims, tokenTail10, c); err != nil {
+				logger.Warn("verify error", logger.Err(err), logger.String("uid", claims.UID), logger.String("name", claims.Name))
+				responseUnauthorized(c, o.isSwitchHTTPCode)
+				c.Abort()
+				return
+			}
+		} else {
+			c.Set(ctxUtil.KeyUID, claims.UID)
+			c.Set(ctxUtil.KeyUName, claims.Name)
+		}
+
+		c.Next()
+	}
+}
+
+// -------------------------------------------------------------------------------------------
+
+// VerifyCustomFn verify custom function, tokenTail10 is the last 10 characters of the token.
+type VerifyCustomFn func(claims *jwt.CustomClaims, tokenTail10 string, c *gin.Context) error
+
+// AuthCustom custom authentication
+func AuthCustom(verify VerifyCustomFn, opts ...JwtOption) gin.HandlerFunc {
+	o := defaultJwtOptions()
+	o.apply(opts...)
+
+	return func(c *gin.Context) {
+		authorization := c.GetHeader(HeaderAuthorizationKey)
+		if len(authorization) < 150 {
+			logger.Warn("authorization is illegal")
+			responseUnauthorized(c, o.isSwitchHTTPCode)
+			c.Abort()
+			return
+		}
+
+		token := authorization[7:] // remove Bearer prefix
+		claims, err := jwt.ParseCustomToken(token)
+		if err != nil {
+			logger.Warn("ParseToken error", logger.Err(err))
+			responseUnauthorized(c, o.isSwitchHTTPCode)
+			c.Abort()
+			return
+		}
+
+		tokenTail10 := token[len(token)-10:]
+		if err = verify(claims, tokenTail10, c); err != nil {
+			logger.Warn("verify error", logger.Err(err), logger.Any("fields", claims.Fields))
+			responseUnauthorized(c, o.isSwitchHTTPCode)
+			c.Abort()
+			return
+		}
+
+		c.Next()
+	}
+}