feat: restore redis io

Signed-off-by: Jiyong Huang <[email protected]>

internal/binder/io/ext_redis.go

@@ -0,0 +1,29 @@
+// Copyright 2021-2024 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/v2/internal/io/redis"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+)
+
+func init() {
+	modules.RegisterLookupSource("redis", redis.GetLookupSource)
+	modules.RegisterSink("redis", redis.GetSink)
+	modules.RegisterSink("redisPub", redis.RedisPub)
+	modules.RegisterSource("redisSub", redis.RedisSub)
+}

internal/io/redis/lookup.go

@@ -0,0 +1,141 @@
+// Copyright 2022-2024 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.
+
+package redis
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strconv"
+
+	"github.com/redis/go-redis/v9"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/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"`
+	DB       string `json:"datasource,omitempty"`
+}
+
+type lookupSource struct {
+	c   *conf
+	db  int
+	cli *redis.Client
+}
+
+func (s *lookupSource) Provision(ctx api.StreamContext, props map[string]any) error {
+	return s.Validate(props)
+}
+
+func (s *lookupSource) Connect(ctx api.StreamContext) error {
+	logger := ctx.GetLogger()
+	logger.Debug("Opening redis lookup source")
+
+	s.cli = redis.NewClient(&redis.Options{
+		Addr:     s.c.Addr,
+		Username: s.c.Username,
+		Password: s.c.Password,
+		DB:       s.db, // use default DB
+	})
+	_, err := s.cli.Ping(ctx).Result()
+	return err
+}
+
+func (s *lookupSource) Lookup(ctx api.StreamContext, _ []string, keys []string, values []any) ([]map[string]any, 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(ctx, v).Result()
+		if err != nil {
+			if err == redis.Nil {
+				return []map[string]any{}, nil
+			}
+			return nil, err
+		}
+		m := make(map[string]any)
+		err = json.Unmarshal(cast.StringToBytes(res), &m)
+		if err != nil {
+			return nil, err
+		}
+		return []map[string]any{m}, nil
+	} else {
+		res, err := s.cli.LRange(ctx, v, 0, -1).Result()
+		if err != nil {
+			if err == redis.Nil {
+				return []map[string]any{}, nil
+			}
+			return nil, err
+		}
+		ret := make([]map[string]any, 0, len(res))
+		for _, r := range res {
+			m := make(map[string]any)
+			err = json.Unmarshal(cast.StringToBytes(r), &m)
+			if err != nil {
+				return nil, err
+			}
+			ret = append(ret, m)
+		}
+		return ret, nil
+	}
+}
+
+func (s *lookupSource) Validate(props map[string]any) error {
+	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.db, err = strconv.Atoi(cfg.DB)
+	if err != nil {
+		return fmt.Errorf("datasource %s is invalid", cfg.DB)
+	}
+	if s.db < 0 || s.db > 15 {
+		return fmt.Errorf("redis lookup source db should be in range 0-15")
+	}
+	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)
+	return nil
+}
+
+func (s *lookupSource) Close(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("Closing redis lookup source")
+	return s.cli.Close()
+}
+
+func GetLookupSource() api.Source {
+	return &lookupSource{}
+}
+
+var _ api.LookupSource = &lookupSource{}

internal/io/redis/lookup_test.go

@@ -0,0 +1,136 @@
+// Copyright 2022-2024 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.
+
+package redis
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+)
+
+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"}`)
+	mr = s
+}
+
+// TestSingle test lookup value of a single map
+func TestSingle(t *testing.T) {
+	ctx := mockContext.NewMockContext("test", "tt")
+	ls := GetLookupSource()
+	err := ls.Provision(ctx, map[string]any{"addr": addr, "datatype": "string", "datasource": "0"})
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	err = ls.Connect(ctx)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	tests := []struct {
+		value  int
+		result []map[string]any
+	}{
+		{
+			value: 1,
+			result: []map[string]any{
+				{"id": float64(1), "name": "John", "address": float64(34), "mobile": "334433"},
+			},
+		}, {
+			value: 2,
+			result: []map[string]any{
+				{"id": float64(2), "name": "Susan", "address": float64(22), "mobile": "666433"},
+			},
+		}, {
+			value:  3,
+			result: []map[string]any{},
+		},
+	}
+	for i, tt := range tests {
+		t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
+			actual, err := ls.(api.LookupSource).Lookup(ctx, []string{}, []string{"id"}, []any{tt.value})
+			assert.NoError(t, err)
+			assert.Equal(t, tt.result, actual)
+		})
+	}
+}
+
+func TestList(t *testing.T) {
+	ctx := mockContext.NewMockContext("test", "tt")
+	ls := GetLookupSource()
+	err := ls.Provision(ctx, map[string]any{"addr": addr, "datatype": "list", "datasource": "0"})
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	err = ls.Connect(ctx)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	tests := []struct {
+		value  string
+		result []map[string]any
+	}{
+		{
+			value: "group1",
+			result: []map[string]any{
+				{"id": float64(2), "name": "Susan"},
+				{"id": float64(1), "name": "John"},
+			},
+		}, {
+			value: "group2",
+			result: []map[string]any{
+				{"id": float64(3), "name": "Nancy"},
+			},
+		}, {
+			value:  "group4",
+			result: []map[string]any{},
+		},
+	}
+	for i, tt := range tests {
+		t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
+			actual, err := ls.(api.LookupSource).Lookup(ctx, []string{}, []string{"id"}, []any{tt.value})
+			assert.NoError(t, err)
+			assert.Equal(t, tt.result, actual)
+		})
+	}
+}
+
+func TestLookupSourceDB(t *testing.T) {
+	ctx := mockContext.NewMockContext("test", "tt")
+	s := &lookupSource{}
+	err := s.Provision(ctx, map[string]any{"addr": addr, "datatype": "string", "datasource": "199"})
+	require.Error(t, err)
+	require.Equal(t, "redis lookup source db should be in range 0-15", err.Error())
+}