Repository: zeromicro/go-zero
Pull request #2472 (2022-09-30 06:55:23)

feat(redis):add timeout method to extend blpop

Suyghur: please code review @kevwan 

Patch:
diff --git a/core/mapping/unmarshaler.go b/core/mapping/unmarshaler.go
index d8c71b7b..a4cdf3f6 100644
--- a/core/mapping/unmarshaler.go
+++ b/core/mapping/unmarshaler.go
@@ -205,10 +205,10 @@ func (u *Unmarshaler) processFieldNotFromString(field reflect.StructField, value
 		return u.processFieldStruct(field, value, mapValue, fullName)
 	case valueKind == reflect.Map && typeKind == reflect.Map:
 		return u.fillMap(field, value, mapValue)
-	case valueKind == reflect.String && typeKind == reflect.Slice:
-		return u.fillSliceFromString(fieldType, value, mapValue)
 	case valueKind == reflect.String && typeKind == reflect.Map:
 		return u.fillMapFromString(value, mapValue)
+	case valueKind == reflect.String && typeKind == reflect.Slice:
+		return u.fillSliceFromString(fieldType, value, mapValue)
 	case valueKind == reflect.String && derefedFieldType == durationType:
 		return fillDurationValue(fieldType.Kind(), value, mapValue.(string))
 	default:
@@ -444,6 +444,27 @@ func (u *Unmarshaler) fillMap(field reflect.StructField, value reflect.Value, ma
 	return nil
 }
 
+func (u *Unmarshaler) fillMapFromString(value reflect.Value, mapValue interface{}) error {
+	if !value.CanSet() {
+		return errValueNotSettable
+	}
+
+	switch v := mapValue.(type) {
+	case fmt.Stringer:
+		if err := jsonx.UnmarshalFromString(v.String(), value.Addr().Interface()); err != nil {
+			return err
+		}
+	case string:
+		if err := jsonx.UnmarshalFromString(v, value.Addr().Interface()); err != nil {
+			return err
+		}
+	default:
+		return errUnsupportedType
+	}
+
+	return nil
+}
+
 func (u *Unmarshaler) fillSlice(fieldType reflect.Type, value reflect.Value, mapValue interface{}) error {
 	if !value.CanSet() {
 		return errValueNotSettable
@@ -532,23 +553,6 @@ func (u *Unmarshaler) fillSliceFromString(fieldType reflect.Type, value reflect.
 	return nil
 }
 
-func (u *Unmarshaler) fillMapFromString(value reflect.Value, mapValue interface{}) error {
-	switch v := mapValue.(type) {
-	case fmt.Stringer:
-		if err := jsonx.UnmarshalFromString(v.String(), value.Addr().Interface()); err != nil {
-			return err
-		}
-	case string:
-		if err := jsonx.UnmarshalFromString(v, value.Addr().Interface()); err != nil {
-			return err
-		}
-	default:
-		return errUnsupportedType
-	}
-
-	return nil
-}
-
 func (u *Unmarshaler) fillSliceValue(slice reflect.Value, index int,
 	baseKind reflect.Kind, value interface{}) error {
 	ithVal := slice.Index(index)
diff --git a/core/stores/redis/redis.go b/core/stores/redis/redis.go
index c4d9c401..535f523a 100644
--- a/core/stores/redis/redis.go
+++ b/core/stores/redis/redis.go
@@ -267,6 +267,31 @@ func (s *Redis) BlpopExCtx(ctx context.Context, node RedisNode, key string) (str
 	return vals[1], true, nil
 }
 
+// BlpopWithTimeout uses passed in redis connection to execute blpop command.
+// Control blocking query timeout
+func (s *Redis) BlpopWithTimeout(node RedisNode, timeout time.Duration, key string) (string, error) {
+	return s.BlpopWithTimeoutCtx(context.Background(), node, timeout, key)
+}
+
+// BlpopWithTimeoutCtx uses passed in redis connection to execute blpop command.
+// Control blocking query timeout
+func (s *Redis) BlpopWithTimeoutCtx(ctx context.Context, node RedisNode, timeout time.Duration, key string) (string, error) {
+	if node == nil {
+		return "", ErrNilNode
+	}
+
+	vals, err := node.BLPop(ctx, timeout, key).Result()
+	if err != nil {
+		return "", err
+	}
+
+	if len(vals) < 2 {
+		return "", fmt.Errorf("no value on key: %s", key)
+	}
+
+	return vals[1], nil
+}
+
 // Decr is the implementation of redis decr command.
 func (s *Redis) Decr(key string) (int64, error) {
 	return s.DecrCtx(context.Background(), key)


Test patch:
diff --git a/core/mapping/unmarshaler_test.go b/core/mapping/unmarshaler_test.go
index 9e91432d..fe4c4375 100644
--- a/core/mapping/unmarshaler_test.go
+++ b/core/mapping/unmarshaler_test.go
@@ -2,6 +2,7 @@ package mapping
 
 import (
 	"encoding/json"
+	"fmt"
 	"strconv"
 	"strings"
 	"testing"
@@ -467,6 +468,85 @@ func TestUnmarshalIntSliceFromString(t *testing.T) {
 	ast.Equal(2, v.Values[1])
 }
 
+func TestUnmarshalIntMapFromString(t *testing.T) {
+	var v struct {
+		Sort map[string]int `key:"sort"`
+	}
+	m := map[string]interface{}{
+		"sort": `{"value":12345,"zeroVal":0,"nullVal":null}`,
+	}
+
+	ast := assert.New(t)
+	ast.Nil(UnmarshalKey(m, &v))
+	ast.Equal(3, len(v.Sort))
+	ast.Equal(12345, v.Sort["value"])
+	ast.Equal(0, v.Sort["zeroVal"])
+	ast.Equal(0, v.Sort["nullVal"])
+}
+
+func TestUnmarshalBoolMapFromString(t *testing.T) {
+	var v struct {
+		Sort map[string]bool `key:"sort"`
+	}
+	m := map[string]interface{}{
+		"sort": `{"value":true,"zeroVal":false,"nullVal":null}`,
+	}
+
+	ast := assert.New(t)
+	ast.Nil(UnmarshalKey(m, &v))
+	ast.Equal(3, len(v.Sort))
+	ast.Equal(true, v.Sort["value"])
+	ast.Equal(false, v.Sort["zeroVal"])
+	ast.Equal(false, v.Sort["nullVal"])
+}
+
+type CustomStringer string
+
+type UnsupportedStringer string
+
+func (c CustomStringer) String() string {
+	return fmt.Sprintf("{%s}", string(c))
+}
+
+func TestUnmarshalStringMapFromStringer(t *testing.T) {
+	var v struct {
+		Sort map[string]string `key:"sort"`
+	}
+	m := map[string]interface{}{
+		"sort": CustomStringer(`"value":"ascend","emptyStr":""`),
+	}
+
+	ast := assert.New(t)
+	ast.Nil(UnmarshalKey(m, &v))
+	ast.Equal(2, len(v.Sort))
+	ast.Equal("ascend", v.Sort["value"])
+	ast.Equal("", v.Sort["emptyStr"])
+}
+
+func TestUnmarshalStringMapFromUnsupportedType(t *testing.T) {
+	var v struct {
+		Sort map[string]string `key:"sort"`
+	}
+	m := map[string]interface{}{
+		"sort": UnsupportedStringer(`{"value":"ascend","emptyStr":""}`),
+	}
+
+	ast := assert.New(t)
+	ast.NotNil(UnmarshalKey(m, &v))
+}
+
+func TestUnmarshalStringMapFromNotSettableValue(t *testing.T) {
+	var v struct {
+		sort map[string]string `key:"sort"`
+	}
+	m := map[string]interface{}{
+		"sort": `{"value":"ascend","emptyStr":""}`,
+	}
+
+	ast := assert.New(t)
+	ast.NotNil(UnmarshalKey(m, &v))
+}
+
 func TestUnmarshalStringMapFromString(t *testing.T) {
 	var v struct {
 		Sort map[string]string `key:"sort"`
@@ -482,6 +562,32 @@ func TestUnmarshalStringMapFromString(t *testing.T) {
 	ast.Equal("", v.Sort["emptyStr"])
 }
 
+func TestUnmarshalStructMapFromString(t *testing.T) {
+	var v struct {
+		Filter map[string]struct {
+			Field1 bool     `json:"field1"`
+			Field2 int64    `json:"field2,string"`
+			Field3 string   `json:"field3"`
+			Field4 *string  `json:"field4"`
+			Field5 []string `json:"field5"`
+		} `key:"filter"`
+	}
+	m := map[string]interface{}{
+		"filter": `{"obj":{"field1":true,"field2":"1573570455447539712","field3":"this is a string",
+			"field4":"this is a string pointer","field5":["str1","str2"]}}`,
+	}
+
+	ast := assert.New(t)
+	ast.Nil(UnmarshalKey(m, &v))
+	ast.Equal(1, len(v.Filter))
+	ast.NotNil(v.Filter["obj"])
+	ast.Equal(true, v.Filter["obj"].Field1)
+	ast.Equal(int64(1573570455447539712), v.Filter["obj"].Field2)
+	ast.Equal("this is a string", v.Filter["obj"].Field3)
+	ast.Equal("this is a string pointer", *v.Filter["obj"].Field4)
+	ast.ElementsMatch([]string{"str1", "str2"}, v.Filter["obj"].Field5)
+}
+
 func TestUnmarshalStringSliceMapFromString(t *testing.T) {
 	var v struct {
 		Filter map[string][]string `key:"filter"`
diff --git a/core/stores/redis/redis_test.go b/core/stores/redis/redis_test.go
index 4f960909..3dd2a576 100644
--- a/core/stores/redis/redis_test.go
+++ b/core/stores/redis/redis_test.go
@@ -1117,6 +1117,17 @@ func TestRedisBlpopEx(t *testing.T) {
 	})
 }
 
+func TestRedisBlpopWithTimeout(t *testing.T) {
+	runOnRedis(t, func(client *Redis) {
+		client.Ping()
+		var node mockedNode
+		_, err := client.BlpopWithTimeout(nil, 10*time.Second, "foo")
+		assert.NotNil(t, err)
+		_, err = client.BlpopWithTimeout(node, 10*time.Second, "foo")
+		assert.NotNil(t, err)
+	})
+}
+
 func TestRedisGeo(t *testing.T) {
 	runOnRedis(t, func(client *Redis) {
 		client.Ping()