Repository: lf-edge/ekuiper
Pull request #1411 (2022-09-23 09:15:11)

feat(lookup): redis lookup source
Signed-off-by: Jiyong Huang <huangjy@emqx.io>


Patch:
diff --git a/etc/sources/redis.json b/etc/sources/redis.json
new file mode 100644
index 00000000..116052e8
--- /dev/null
+++ b/etc/sources/redis.json
@@ -0,0 +1,96 @@
+{
+  "about": {
+    "trial": true,
+    "author": {
+      "name": "EMQ",
+      "email": "contact@emqx.io",
+      "company": "EMQ Technologies Co., Ltd",
+      "website": "https://www.emqx.io"
+    },
+    "helpUrl": {
+      "en_US": "https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/rules/sources/builtin/redis.md",
+      "zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/rules/sources/builtin/redis.md"
+    },
+    "description": {
+      "en_US": "Read message from Redis",
+      "zh_CN": "从 Redis 中读取消息"
+    }
+  },
+  "libs": [],
+  "properties": {
+    "default": [
+      {
+        "name": "addr",
+        "default": "10.122.48.17:6379",
+        "optional": false,
+        "control": "text",
+        "type": "string",
+        "hint": {
+          "en_US": "The addr of the Redis,example: 10.122.48.17:6379",
+          "zh_CN": "Redis的地址, 例如: 10.122.48.17:6379"
+        },
+        "label": {
+          "en_US": "Addr",
+          "zh_CN": "地址"
+        }
+      },
+      {
+        "name": "username",
+        "default": "",
+        "optional": true,
+        "control": "text",
+        "type": "string",
+        "hint": {
+          "en_US": "The Redis user name",
+          "zh_CN": "Redis 用户名"
+        },
+        "label": {
+          "en_US": "Username",
+          "zh_CN": "用户名"
+        }
+      },
+      {
+        "name": "password",
+        "default": "",
+        "optional": true,
+        "control": "text",
+        "type": "string",
+        "hint": {
+          "en_US": "The Redis login password",
+          "zh_CN": "Redis 登陆密码"
+        },
+        "label": {
+          "en_US": "Password",
+          "zh_CN": "密码"
+        }
+      },
+      {
+        "name": "dataType",
+        "default": "string",
+        "optional": false,
+        "control": "select",
+        "type": "string",
+        "values": [
+          "string",
+          "list"
+        ],
+        "hint": {
+          "en_US": "The Redis data type, could be string or list. The default is string.",
+          "zh_CN": "Redis 数据的类型，可以为 string 或者 list， 默认是 string。"
+        },
+        "label": {
+          "en_US": "data type",
+          "zh_CN": "数据类型"
+        }
+      }
+    ]
+  },
+  "node": {
+    "category": ["Protocol","Protocol1"],
+    "icon": "iconPath",
+    "label": {
+      "en_US": "Redis",
+      "zh_CN": "Redis"
+    }
+  }
+}
diff --git a/etc/sources/redis.yaml b/etc/sources/redis.yaml
new file mode 100644
index 00000000..ae797dd8
--- /dev/null
+++ b/etc/sources/redis.yaml
@@ -0,0 +1,7 @@
+default:
+  # the redis host address
+  addr: "127.0.0.1:6379"
+  # currently supports string and list only
+  datatype: "string"
+#  username: ""
+#  password: ""
\ No newline at end of file
diff --git a/internal/binder/io/ext_redis.go b/internal/binder/io/ext_redis.go
new file mode 100644
index 00000000..43c68798
--- /dev/null
+++ b/internal/binder/io/ext_redis.go
@@ -0,0 +1,27 @@
+// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build redisdb || !core
+
+package io
+
+import (
+	"github.com/lf-edge/ekuiper/internal/topo/redis"
+	"github.com/lf-edge/ekuiper/pkg/api"
+)
+
+func init() {
+	lookupSources["redis"] = func() api.LookupSource { return redis.GetLookupSource() }
+	sinks["redis"] = func() api.Sink { return redis.GetSink() }
+}
diff --git a/internal/topo/redis/lookup.go b/internal/topo/redis/lookup.go
new file mode 100644
index 00000000..196e5965
--- /dev/null
+++ b/internal/topo/redis/lookup.go
@@ -0,0 +1,123 @@
+// Copyright 2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build redisdb || !core
+
+package redis
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"github.com/go-redis/redis/v7"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/cast"
+)
+
+type conf struct {
+	// host:port address.
+	Addr     string `json:"addr,omitempty"`
+	Username string `json:"username,omitempty"`
+	// Optional password. Must match the password specified in the
+	Password string `json:"password,omitempty"`
+	DataType string `json:"dataType,omitempty"`
+}
+
+type lookupSource struct {
+	c   *conf
+	db  int
+	cli *redis.Client
+}
+
+func (s *lookupSource) Configure(datasource string, props map[string]interface{}) error {
+	db, err := cast.ToInt(datasource, cast.CONVERT_ALL)
+	if err != nil {
+		return fmt.Errorf("invalid datasource, it must be an integer but got %s", datasource)
+	}
+	s.db = db
+	cfg := &conf{}
+	err = cast.MapToStruct(props, cfg)
+	if err != nil {
+		return err
+	}
+	if cfg.Addr == "" {
+		return errors.New("redis addr is null")
+	}
+	if cfg.DataType != "string" && cfg.DataType != "list" {
+		return errors.New("redis dataType must be string or list")
+	}
+	s.c = cfg
+	return nil
+}
+
+func (s *lookupSource) Open(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("Opening redis lookup source with conf %v", s.c)
+	s.cli = redis.NewClient(&redis.Options{
+		Addr:     s.c.Addr,
+		Username: s.c.Username,
+		Password: s.c.Password,
+		DB:       s.db,
+	})
+	return nil
+}
+
+func (s *lookupSource) Lookup(ctx api.StreamContext, keys []string, values []interface{}) ([]api.SourceTuple, error) {
+	ctx.GetLogger().Debugf("Lookup redis %v", keys)
+	if len(keys) != 1 {
+		return nil, fmt.Errorf("redis lookup only support one key, but got %v", keys)
+	}
+	v := fmt.Sprintf("%v", values[0])
+	if s.c.DataType == "string" {
+		res, err := s.cli.Get(v).Result()
+		if err != nil {
+			if err == redis.Nil {
+				return []api.SourceTuple{}, nil
+			}
+			return nil, err
+		}
+		m := make(map[string]interface{})
+		err = json.Unmarshal([]byte(res), &m)
+		if err != nil {
+			return nil, err
+		}
+		return []api.SourceTuple{api.NewDefaultSourceTuple(m, nil)}, nil
+	} else {
+		res, err := s.cli.LRange(v, 0, -1).Result()
+		if err != nil {
+			if err == redis.Nil {
+				return []api.SourceTuple{}, nil
+			}
+			return nil, err
+		}
+		ret := make([]api.SourceTuple, 0, len(res))
+		for _, r := range res {
+			m := make(map[string]interface{})
+			err = json.Unmarshal([]byte(r), &m)
+			if err != nil {
+				return nil, err
+			}
+			ret = append(ret, api.NewDefaultSourceTuple(m, nil))
+		}
+		return ret, nil
+	}
+}
+
+func (s *lookupSource) Close(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("Closing redis lookup source")
+	return s.cli.Close()
+}
+
+func GetLookupSource() api.LookupSource {
+	return &lookupSource{}
+}
diff --git a/internal/topo/redis/sink.go b/internal/topo/redis/sink.go
new file mode 100644
index 00000000..01ce832f
--- /dev/null
+++ b/internal/topo/redis/sink.go
@@ -0,0 +1,213 @@
+// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build redisdb || !core
+
+package redis
+
+import (
+	"errors"
+	"fmt"
+	"github.com/lf-edge/ekuiper/pkg/errorx"
+	"time"
+
+	"github.com/go-redis/redis/v7"
+
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/cast"
+)
+
+type RedisSink struct {
+	// host:port address.
+	addr     string
+	username string
+	// Optional password. Must match the password specified in the
+	password string
+	// Database to be selected after connecting to the server.
+	db int
+
+	// key of field
+	field string
+
+	// key define
+	key string
+
+	dataType string
+
+	expiration time.Duration
+
+	sendSingle bool
+
+	cli *redis.Client
+}
+
+func (r *RedisSink) Configure(props map[string]interface{}) error {
+	if i, ok := props["addr"]; ok {
+		if i, ok := i.(string); ok {
+			r.addr = i
+		}
+	} else {
+		return errors.New("redis addr is null")
+	}
+
+	if i, ok := props["password"]; ok {
+		if i, ok := i.(string); ok {
+			r.password = i
+		}
+	}
+
+	r.db = 0
+	if i, ok := props["db"]; ok {
+		if t, err := cast.ToInt(i, cast.STRICT); err == nil {
+			r.db = t
+		}
+	}
+
+	if i, ok := props["key"]; ok {
+		if i, ok := i.(string); ok {
+			r.key = i
+		}
+	} else {
+		return errors.New("not config data key for redis")
+	}
+
+	if i, ok := props["field"]; ok {
+		if i, ok := i.(string); ok {
+			r.field = i
+		}
+	}
+
+	r.sendSingle = true
+	if i, ok := props["sendSingle"]; ok {
+		if i, ok := i.(bool); ok {
+			r.sendSingle = i
+		}
+	}
+
+	r.dataType = "string"
+	if i, ok := props["dataType"]; ok {
+		if i, ok := i.(string); ok {
+			r.dataType = i
+		}
+	}
+
+	r.expiration = -1
+	if i, ok := props["expiration"]; ok {
+		if t, err := cast.ToInt(i, cast.STRICT); err == nil {
+			r.expiration = time.Duration(t)
+		}
+	}
+
+	return nil
+}
+
+func (r *RedisSink) Open(ctx api.StreamContext) (err error) {
+	logger := ctx.GetLogger()
+	logger.Debug("Opening redis sink")
+
+	r.cli = redis.NewClient(&redis.Options{
+		Addr:     r.addr,
+		Username: r.username,
+		Password: r.password,
+		DB:       r.db, // use default DB
+	})
+
+	return nil
+}
+
+func (r *RedisSink) Collect(ctx api.StreamContext, data interface{}) error {
+	logger := ctx.GetLogger()
+	v, _, err := ctx.TransformOutput(data)
+	if err != nil {
+		logger.Error(err)
+		return err
+	}
+	if r.field != "" {
+		switch out := data.(type) {
+		case []map[string]interface{}:
+			for _, m := range out {
+				key := r.field
+				k, err := cast.ToString(m[key], cast.CONVERT_ALL)
+				if err != nil {
+					return fmt.Errorf("key must be string or convertible to string, but got %v", m[key])
+				}
+
+				if r.dataType == "list" {
+					err := r.cli.LPush(k, v).Err()
+					if err != nil {
+						logger.Error(err)
+						return fmt.Errorf("%s:%s", errorx.IOErr, err.Error())
+					}
+					logger.Debugf("send redis list success, key:%s data: %s", k, string(v))
+				} else {
+					err := r.cli.Set(k, v, r.expiration*time.Second).Err()
+					if err != nil {
+						logger.Error(err)
+						return fmt.Errorf("%s:%s", errorx.IOErr, err.Error())
+					}
+					logger.Debugf("send redis string success, key:%s data: %s", k, string(v))
+				}
+			}
+		case map[string]interface{}:
+			key := r.field
+			k, err := cast.ToString(out[key], cast.CONVERT_ALL)
+			if err != nil {
+				return fmt.Errorf("key must be string or convertible to string, but got %v", out[key])
+			}
+
+			if r.dataType == "list" {
+				err := r.cli.LPush(k, v).Err()
+				if err != nil {
+					logger.Error(err)
+					return fmt.Errorf("%s:%s", errorx.IOErr, err.Error())
+				}
+				logger.Debugf("send redis list success, key:%s data: %s", k, string(v))
+			} else {
+				err := r.cli.Set(k, v, r.expiration*time.Second).Err()
+				if err != nil {
+					logger.Error(err)
+					return fmt.Errorf("%s:%s", errorx.IOErr, err.Error())
+				}
+				logger.Debugf("send redis string success, key:%s data: %s", k, string(v))
+			}
+		}
+	} else if r.key != "" {
+		if r.dataType == "list" {
+			err := r.cli.LPush(r.key, v).Err()
+			if err != nil {
+				logger.Error(err)
+				return fmt.Errorf("%s:%s", errorx.IOErr, err.Error())
+			}
+			logger.Debugf("send redis list success, key:%s data: %s", r.key, string(v))
+		} else {
+			err := r.cli.Set(r.key, v, r.expiration*time.Second).Err()
+			if err != nil {
+				logger.Error(err)
+				return fmt.Errorf("%s:%s", errorx.IOErr, err.Error())
+			}
+			logger.Debugf("send redis string success, key:%s data: %s", r.key, string(v))
+		}
+	}
+	logger.Debug("insert success %v", data)
+	return nil
+}
+
+func (r *RedisSink) Close(ctx api.StreamContext) error {
+	err := r.cli.Close()
+	return err
+}
+
+func GetSink() api.Sink {
+	return &RedisSink{}
+}


Test patch:
diff --git a/internal/plugin/portable/test/portable_rule_test.go b/internal/plugin/portable/test/portable_rule_test.go
index e239809c..d1afe981 100644
--- a/internal/plugin/portable/test/portable_rule_test.go
+++ b/internal/plugin/portable/test/portable_rule_test.go
@@ -24,7 +24,7 @@ import (
 	"github.com/lf-edge/ekuiper/internal/plugin/portable/runtime"
 	"github.com/lf-edge/ekuiper/internal/processor"
 	"github.com/lf-edge/ekuiper/internal/topo"
-	"github.com/lf-edge/ekuiper/internal/topo/memory"
+	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
 	"github.com/lf-edge/ekuiper/internal/topo/planner"
 	"github.com/lf-edge/ekuiper/internal/topo/topotest"
 	"github.com/lf-edge/ekuiper/pkg/api"
@@ -111,7 +111,7 @@ func TestSourceAndFunc(t *testing.T) {
 			continue
 		}
 		var mm [][]map[string]interface{}
-		ch := memory.CreateSub("cache", nil, fmt.Sprintf("cache%d", i+1), 10)
+		ch := pubsub.CreateSub("cache", nil, fmt.Sprintf("cache%d", i+1), 10)
 		ctx, cancel := context.WithTimeout(context.Background(), 10000*time.Second)
 		go func(ctx context.Context) {
 			select {
diff --git a/internal/topo/redis/lookup_test.go b/internal/topo/redis/lookup_test.go
new file mode 100644
index 00000000..80080506
--- /dev/null
+++ b/internal/topo/redis/lookup_test.go
@@ -0,0 +1,138 @@
+// Copyright 2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build redisdb || !core
+
+package redis
+
+import (
+	"github.com/alicebob/miniredis/v2"
+	econf "github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/topo/context"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"reflect"
+	"testing"
+)
+
+var addr string
+
+func init() {
+	s, err := miniredis.Run()
+	if err != nil {
+		panic(err)
+	}
+	addr = "localhost:" + s.Port()
+	// Mock id key data
+	s.Set("1", `{"id":1,"name":"John","address":34,"mobile":"334433"}`)
+	s.Set("2", `{"id":2,"name":"Susan","address":22,"mobile":"666433"}`)
+	// Mock group key list data
+	s.Lpush("group1", `{"id":1,"name":"John"}`)
+	s.Lpush("group1", `{"id":2,"name":"Susan"}`)
+	s.Lpush("group2", `{"id":3,"name":"Nancy"}`)
+	s.Lpush("group3", `{"id":4,"name":"Tom"}`)
+}
+
+// TestSingle test lookup value of a single map
+func TestSingle(t *testing.T) {
+	contextLogger := econf.Log.WithField("rule", "test")
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+	ls := GetLookupSource()
+	err := ls.Configure("0", map[string]interface{}{"addr": addr, "datatype": "string"})
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	err = ls.Open(ctx)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	var tests = []struct {
+		value  int
+		result []api.SourceTuple
+	}{
+		{
+			value: 1,
+			result: []api.SourceTuple{
+				api.NewDefaultSourceTuple(map[string]interface{}{"id": float64(1), "name": "John", "address": float64(34), "mobile": "334433"}, nil),
+			},
+		}, {
+			value: 2,
+			result: []api.SourceTuple{
+				api.NewDefaultSourceTuple(map[string]interface{}{"id": float64(2), "name": "Susan", "address": float64(22), "mobile": "666433"}, nil),
+			},
+		}, {
+			value:  3,
+			result: []api.SourceTuple{},
+		},
+	}
+	for i, tt := range tests {
+		actual, err := ls.Lookup(ctx, []string{"id"}, []interface{}{tt.value})
+		if err != nil {
+			t.Errorf("Test %d: %v", i, err)
+			continue
+		}
+		if !reflect.DeepEqual(actual, tt.result) {
+			t.Errorf("Test %d: expected %v, actual %v", i, tt.result, actual)
+			continue
+		}
+	}
+}
+
+func TestList(t *testing.T) {
+	contextLogger := econf.Log.WithField("rule", "test")
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+	ls := GetLookupSource()
+	err := ls.Configure("0", map[string]interface{}{"addr": addr, "datatype": "list"})
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	err = ls.Open(ctx)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	var tests = []struct {
+		value  string
+		result []api.SourceTuple
+	}{
+		{
+			value: "group1",
+			result: []api.SourceTuple{
+				api.NewDefaultSourceTuple(map[string]interface{}{"id": float64(2), "name": "Susan"}, nil),
+				api.NewDefaultSourceTuple(map[string]interface{}{"id": float64(1), "name": "John"}, nil),
+			},
+		}, {
+			value: "group2",
+			result: []api.SourceTuple{
+				api.NewDefaultSourceTuple(map[string]interface{}{"id": float64(3), "name": "Nancy"}, nil),
+			},
+		}, {
+			value:  "group4",
+			result: []api.SourceTuple{},
+		},
+	}
+	for i, tt := range tests {
+		actual, err := ls.Lookup(ctx, []string{"id"}, []interface{}{tt.value})
+		if err != nil {
+			t.Errorf("Test %d: %v", i, err)
+			continue
+		}
+		if !reflect.DeepEqual(actual, tt.result) {
+			t.Errorf("Test %d: expected %v, actual %v", i, tt.result, actual)
+			continue
+		}
+	}
+}