Repository: lf-edge/ekuiper
Pull request #3004 (2024-07-12 10:01:38)

feat(converter): delimiter supports header

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3004?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
Attention: Patch coverage is `98.00000%` with `1 line` in your changes missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@3e78c26`). [Learn more](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#section-missing-base-commit) about missing BASE report.

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3004?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [internal/io/file/sink.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3004?src=pr&el=tree&filepath=internal%2Fio%2Ffile%2Fsink.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vZmlsZS9zaW5rLmdv) | 95.24% | [1 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3004?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff              @@
##             master-v2    #3004   +/-   ##
============================================
  Coverage             ?   68.61%           
============================================
  Files                ?      348           
  Lines                ?    33436           
  Branches             ?        0           
============================================
  Hits                 ?    22941           
  Misses               ?     8449           
  Partials             ?     2046           
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3004?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).


Patch:
diff --git a/etc/sources/edgex.yaml b/etc/sources/edgex.yaml
index ac823aa7..9f375a31 100644
--- a/etc/sources/edgex.yaml
+++ b/etc/sources/edgex.yaml
@@ -56,5 +56,4 @@ share_conf: #Conf_key
   server: 127.0.0.1
   port: 1883
   topic: events
-  type: mqtt
-  connectionSelector: edgex.redisMsgBus
+  connectionSelector: redisMsgBus
diff --git a/go.mod b/go.mod
index c866d8db..98082c60 100644
--- a/go.mod
+++ b/go.mod
@@ -14,15 +14,16 @@ require (
 	github.com/eclipse/paho.mqtt.golang v1.4.3
 	github.com/edgexfoundry/go-mod-core-contracts/v3 v3.1.0
 	github.com/edgexfoundry/go-mod-messaging/v3 v3.1.0
+	github.com/fxamacker/cbor/v2 v2.6.0
 	github.com/gdexlab/go-render v1.0.1
 	github.com/go-sql-driver/mysql v1.8.1
 	github.com/golang-collections/collections v0.0.0-20130729185459-604e922904d3
 	github.com/golang-jwt/jwt/v5 v5.2.1
 	github.com/golang/protobuf v1.5.4
-	github.com/google/pprof v0.0.0-20240528025155-186aa0362fba
 	github.com/google/uuid v1.6.0
 	github.com/gorilla/handlers v1.5.2
 	github.com/gorilla/mux v1.8.1
+	github.com/gorilla/websocket v1.5.1
 	github.com/jhump/protoreflect v1.15.6
 	github.com/jinzhu/now v1.1.5
 	github.com/keepeye/logrus-filename v0.0.0-20190711075016-ce01a4391dd1
@@ -80,7 +81,6 @@ require (
 	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/fsnotify/fsnotify v1.7.0 // indirect
-	github.com/fxamacker/cbor/v2 v2.6.0 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.3 // indirect
 	github.com/go-kit/kit v0.10.0 // indirect
 	github.com/go-ole/go-ole v1.3.0 // indirect
@@ -89,7 +89,7 @@ require (
 	github.com/go-playground/validator/v10 v10.19.0 // indirect
 	github.com/go-redis/redis/v7 v7.4.1 // indirect
 	github.com/go-sourcemap/sourcemap v2.1.4+incompatible // indirect
-	github.com/gorilla/websocket v1.5.1 // indirect
+	github.com/google/pprof v0.0.0-20240528025155-186aa0362fba // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/hashicorp/golang-lru v0.5.4 // indirect
@@ -127,6 +127,7 @@ require (
 	github.com/tklauser/go-sysconf v0.3.13 // indirect
 	github.com/tklauser/numcpus v0.7.0 // indirect
 	github.com/u2takey/go-utils v0.3.1 // indirect
+	github.com/x448/float16 v0.8.4 // indirect
 	github.com/yuin/gopher-lua v1.1.0 // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	go.opentelemetry.io/otel v1.7.0 // indirect
diff --git a/internal/binder/io/ext_edgex.go b/internal/binder/io/ext_edgex.go
new file mode 100644
index 00000000..a795ac05
--- /dev/null
+++ b/internal/binder/io/ext_edgex.go
@@ -0,0 +1,29 @@
+// Copyright 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 edgex || full
+
+package io
+
+import (
+	"github.com/lf-edge/ekuiper/v2/internal/io/edgex"
+	edgexCon "github.com/lf-edge/ekuiper/v2/internal/io/edgex/client"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+)
+
+func init() {
+	modules.RegisterConnection("edgex", edgexCon.GetConnection)
+	modules.RegisterSource("edgex", edgex.GetSource)
+	modules.RegisterSink("edgex", edgex.GetSink)
+}
diff --git a/internal/converter/delimited/converter.go b/internal/converter/delimited/converter.go
index b70f024e..65730d46 100644
--- a/internal/converter/delimited/converter.go
+++ b/internal/converter/delimited/converter.go
@@ -16,6 +16,7 @@ package delimited
 
 import (
 	"bytes"
+	"encoding/binary"
 	"fmt"
 	"sort"
 	"strconv"
@@ -29,7 +30,8 @@ import (
 
 type Converter struct {
 	Delimiter string   `json:"delimiter"`
-	Cols      []string `json:"cols"`
+	Cols      []string `json:"fields"`
+	HasHeader bool     `json:"hasHeader"`
 }
 
 func NewConverter(props map[string]any) (message.Converter, error) {
@@ -56,7 +58,7 @@ func (c *Converter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
 		}
 	}()
 	switch m := d.(type) {
-	case map[string]interface{}:
+	case map[string]any:
 		sb := &bytes.Buffer{}
 		if len(c.Cols) == 0 {
 			keys := make([]string, 0, len(m))
@@ -65,8 +67,14 @@ func (c *Converter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
 			}
 			sort.Strings(keys)
 			c.Cols = keys
+			if len(c.Cols) > 0 && c.HasHeader {
+				hb := []byte(strings.Join(c.Cols, c.Delimiter))
+				sb.WriteString(c.Delimiter)
+				_ = binary.Write(sb, binary.BigEndian, uint32(len(hb)))
+				sb.Write(hb)
+				ctx.GetLogger().Infof("delimiter header %s", hb)
+			}
 		}
-
 		for i, v := range c.Cols {
 			if i > 0 {
 				sb.WriteString(c.Delimiter)
@@ -74,6 +82,34 @@ func (c *Converter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
 			fmt.Fprintf(sb, "%v", m[v])
 		}
 		return sb.Bytes(), nil
+	case []map[string]any:
+		sb := &bytes.Buffer{}
+		var cols []string
+		for i, mm := range m {
+			if i > 0 {
+				sb.WriteString("\n")
+			}
+			if len(cols) == 0 {
+				keys := make([]string, 0, len(mm))
+				for k := range mm {
+					keys = append(keys, k)
+				}
+				sort.Strings(keys)
+				cols = keys
+				if len(cols) > 0 && c.HasHeader {
+					hb := []byte(strings.Join(cols, c.Delimiter))
+					sb.Write(hb)
+					sb.WriteString("\n")
+				}
+			}
+			for j, v := range cols {
+				if j > 0 {
+					sb.WriteString(c.Delimiter)
+				}
+				fmt.Fprintf(sb, "%v", mm[v])
+			}
+		}
+		return sb.Bytes(), nil
 	default:
 		return nil, fmt.Errorf("unsupported type %v, must be a map", d)
 	}
diff --git a/internal/io/edgex/client/edgex.go b/internal/io/edgex/client/edgex.go
new file mode 100644
index 00000000..7006d809
--- /dev/null
+++ b/internal/io/edgex/client/edgex.go
@@ -0,0 +1,189 @@
+// 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 client
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/edgexfoundry/go-mod-messaging/v3/messaging"
+	"github.com/edgexfoundry/go-mod-messaging/v3/pkg/types"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+)
+
+type Client struct {
+	mbconf types.MessageBusConfig
+	client messaging.MessageClient
+}
+
+func (es *Client) Ping(_ api.StreamContext) error {
+	if es.client != nil {
+		return nil
+	}
+	return fmt.Errorf("client is nil")
+}
+
+func (es *Client) DetachSub(ctx api.StreamContext, props map[string]any) {
+	topic, ok := props["topic"]
+	ctx.GetLogger().Infof("detach edgex sub %v", topic)
+	if ok {
+		err := es.client.Unsubscribe(topic.(string))
+		if err != nil {
+			ctx.GetLogger().Error(err)
+		}
+	}
+}
+
+func (es *Client) Close(ctx api.StreamContext) error {
+	if es.client != nil {
+		return es.client.Disconnect()
+	}
+	return nil
+}
+
+var _ modules.Connection = &Client{}
+
+func GetConnection(ctx api.StreamContext, props map[string]any) (modules.Connection, error) {
+	ctx.GetLogger().Infof("connect to edgex")
+	c := &Client{}
+	err := c.CfgValidate(props)
+	if err != nil {
+		return nil, err
+	}
+	client, err := messaging.NewMessageClient(c.mbconf)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := client.Connect(); err != nil {
+		conf.Log.Errorf("The connection to edgex messagebus failed.")
+		return nil, fmt.Errorf("Failed to connect to edgex message bus: " + err.Error())
+	}
+	conf.Log.Infof("The connection to edgex messagebus is established successfully.")
+
+	c.client = client
+	return c, nil
+}
+
+type EdgexConf struct {
+	Protocol string            `json:"protocol"`
+	Server   string            `json:"server"`
+	Host     string            `json:"host"`
+	Port     int               `json:"port"`
+	Type     string            `json:"type"`
+	Optional map[string]string `json:"optional"`
+}
+
+// Modify the copied conf to print no password.
+func printConf(mbconf types.MessageBusConfig) {
+	printableOptional := make(map[string]string)
+	for k, v := range mbconf.Optional {
+		if strings.EqualFold(k, "password") {
+			printableOptional[k] = "*"
+		} else {
+			printableOptional[k] = v
+		}
+	}
+	mbconf.Optional = printableOptional
+	conf.Log.Infof("Use configuration for edgex messagebus %v", mbconf)
+}
+
+func (es *Client) CfgValidate(props map[string]interface{}) error {
+	edgeAddr := "localhost"
+	c := &EdgexConf{
+		Protocol: "redis",
+		Port:     6379,
+		Type:     messaging.Redis,
+		Optional: nil,
+	}
+
+	if o, ok := props["optional"]; ok {
+		switch ot := o.(type) {
+		case map[string]string:
+			c.Optional = ot
+		case map[string]interface{}:
+			c.Optional = make(map[string]string)
+			for k, v := range ot {
+				c.Optional[k] = fmt.Sprintf("%v", v)
+			}
+		default:
+			return fmt.Errorf("invalid optional config %v, must be a map", o)
+		}
+		delete(props, "optional")
+	}
+
+	err := cast.MapToStruct(props, c)
+	if err != nil {
+		return fmt.Errorf("map config map to struct fail with error: %v", err)
+	}
+
+	if c.Host != "" {
+		edgeAddr = c.Host
+	} else if c.Server != "" {
+		edgeAddr = c.Server
+	}
+
+	if c.Type != messaging.MQTT && c.Type != messaging.Redis &&
+		c.Type != messaging.NatsCore && c.Type != messaging.NatsJetStream {
+		return fmt.Errorf("specified wrong type value %s", c.Type)
+	}
+	if c.Port < 0 {
+		return fmt.Errorf("specified wrong port value, expect positive integer but got %d", c.Port)
+	}
+
+	mbconf := types.MessageBusConfig{
+		Broker: types.HostInfo{
+			Host:     edgeAddr,
+			Port:     c.Port,
+			Protocol: c.Protocol,
+		},
+		Type: c.Type,
+	}
+	mbconf.Optional = c.Optional
+	es.mbconf = mbconf
+
+	printConf(mbconf)
+
+	return nil
+}
+
+func (es *Client) Publish(env types.MessageEnvelope, topic string) error {
+	if err := es.client.Publish(env, topic); err != nil {
+		conf.Log.Errorf("Publish to topic %s has error : %s.", topic, err.Error())
+		return fmt.Errorf("Failed to publish to edgex message bus: " + err.Error())
+	}
+	return nil
+}
+
+func (es *Client) Subscribe(msg chan types.MessageEnvelope, topic string, err chan error) error {
+	topics := []types.TopicChannel{{Topic: topic, Messages: msg}}
+	if err := es.client.Subscribe(topics, err); err != nil {
+		conf.Log.Errorf("Failed to subscribe to edgex messagebus with topic %s has error : %s.", topic, err.Error())
+		return err
+	}
+	return nil
+}
+
+func (es *Client) Disconnect() error {
+	conf.Log.Infof("Closing the connection to edgex messagebus.")
+	if e := es.client.Disconnect(); e != nil {
+		return e
+	}
+	return nil
+}
diff --git a/internal/io/edgex/sink.go b/internal/io/edgex/sink.go
new file mode 100644
index 00000000..7b74ad1b
--- /dev/null
+++ b/internal/io/edgex/sink.go
@@ -0,0 +1,684 @@
+// 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.
+
+package edgex
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"reflect"
+
+	v3 "github.com/edgexfoundry/go-mod-core-contracts/v3/common"
+	"github.com/edgexfoundry/go-mod-core-contracts/v3/dtos"
+	"github.com/edgexfoundry/go-mod-core-contracts/v3/dtos/requests"
+	"github.com/edgexfoundry/go-mod-messaging/v3/pkg/types"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/edgex/client"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
+	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
+)
+
+type SinkConf struct {
+	Topic        string      `json:"topic"`
+	TopicPrefix  string      `json:"topicPrefix"`
+	MessageType  messageType `json:"messageType"`
+	ContentType  string      `json:"contentType"`
+	DeviceName   string      `json:"deviceName"`
+	ProfileName  string      `json:"profileName"`
+	SourceName   string      `json:"sourceName"`
+	Metadata     string      `json:"metadata"`
+	DataTemplate string      `json:"dataTemplate"`
+	Fields       []string    `json:"fields"`
+	DataField    string      `json:"dataField"`
+}
+
+type EdgexMsgBusSink struct {
+	c *SinkConf
+
+	config map[string]any
+	topic  string
+
+	cli        *client.Client
+	sendParams map[string]any
+}
+
+func (ems *EdgexMsgBusSink) Provision(ctx api.StreamContext, ps map[string]any) error {
+	c := &SinkConf{
+		MessageType: MessageTypeEvent,
+		ContentType: "application/json",
+		DeviceName:  "ekuiper",
+		ProfileName: "ekuiperProfile",
+	}
+
+	err := cast.MapToStruct(ps, c)
+	if err != nil {
+		return fmt.Errorf("read properties %v fail with error: %v", ps, err)
+	}
+
+	if c.MessageType != MessageTypeEvent && c.MessageType != MessageTypeRequest {
+		return fmt.Errorf("specified wrong messageType value %s", c.MessageType)
+	}
+
+	if c.MessageType == MessageTypeEvent && c.ContentType != "application/json" {
+		return fmt.Errorf("specified wrong contentType value %s: only 'application/json' is supported if messageType is event", c.ContentType)
+	}
+
+	if c.Topic != "" && c.TopicPrefix != "" {
+		return fmt.Errorf("not allow to specify both topic and topicPrefix, please set one only")
+	}
+	ems.c = c
+	ems.config = ps
+	ems.sendParams = map[string]any{
+		"contentType": ems.c.ContentType,
+	}
+	if ems.c.SourceName == "" {
+		ems.c.SourceName = ctx.GetRuleId()
+	}
+
+	if ems.c.Topic == "" && ems.c.TopicPrefix == "" {
+		ems.topic = "application"
+	} else if ems.c.Topic != "" {
+		ems.topic = ems.c.Topic
+	} else if ems.c.Metadata == "" { // If meta data are static, the "dynamic" topic is static
+		ems.topic = fmt.Sprintf("%s/%s/%s/%s", ems.c.TopicPrefix, ems.c.ProfileName, ems.c.DeviceName, ems.c.SourceName)
+	} else {
+		ems.topic = "" // calculate dynamically
+	}
+	return nil
+}
+
+func (ems *EdgexMsgBusSink) Connect(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("Connecting to edgex server")
+	var err error
+	id := fmt.Sprintf("%s-%s-%d-edgex-sink", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId())
+	conn, err := connection.FetchConnection(ctx, id, "edgex", ems.config)
+	if err != nil {
+		return err
+	}
+	c, ok := conn.(*client.Client)
+	if !ok {
+		return fmt.Errorf("connection %s should be edgex connection", id)
+	}
+	ems.cli = c
+	return err
+}
+
+func (ems *EdgexMsgBusSink) produceEvents(ctx api.StreamContext, item any) (*dtos.Event, error) {
+	var m []map[string]any
+	switch payload := item.(type) {
+	case map[string]any:
+		m = []map[string]any{payload}
+	case []map[string]any:
+		m = payload
+	default:
+		// impossible
+		return nil, fmt.Errorf("receive invalid data %v", item)
+	}
+	m1 := ems.getMeta(m)
+	event := m1.createEvent()
+	// Override the devicename if user specified the value
+	if event.DeviceName == "" {
+		event.DeviceName = ems.c.DeviceName
+	}
+	if event.ProfileName == "" {
+		event.ProfileName = ems.c.ProfileName
+	}
+	if event.SourceName == "" {
+		event.SourceName = ems.c.SourceName
+	}
+	for _, v := range m {
+		for k1, v1 := range v {
+			// Ignore nil values
+			if k1 == ems.c.Metadata || v1 == nil {
+				continue
+			} else {
+				var (
+					vt  string
+					vv  any
+					err error
+				)
+				mm1 := m1.readingMeta(ctx, k1)
+				if mm1 != nil && mm1.valueType != nil {
+					vt = *mm1.valueType
+					vv, err = getValueByType(v1, vt)
+				} else {
+					vt, vv, err = getValueType(v1)
+				}
+				if err != nil {
+					ctx.GetLogger().Errorf("%v", err)
+					continue
+				}
+				switch vt {
+				case v3.ValueTypeBinary:
+					// default media type
+					event.AddBinaryReading(k1, vv.([]byte), "application/text")
+				case v3.ValueTypeObject:
+					event.AddObjectReading(k1, vv)
+				default:
+					err = event.AddSimpleReading(k1, vt, vv)
+				}
+
+				if err != nil {
+					ctx.GetLogger().Errorf("%v", err)
+					continue
+				}
+				r := event.Readings[len(event.Readings)-1]
+				if mm1 != nil {
+					event.Readings[len(event.Readings)-1] = mm1.decorate(&r)
+				}
+			}
+		}
+	}
+	return event, nil
+}
+
+func getValueType(v any) (string, any, error) {
+	vt := reflect.TypeOf(v)
+	if vt == nil {
+		return "", nil, fmt.Errorf("unsupported value nil")
+	}
+	k := vt.Kind()
+	switch k {
+	case reflect.Bool:
+		return v3.ValueTypeBool, v, nil
+	case reflect.String:
+		return v3.ValueTypeString, v, nil
+	case reflect.Uint8:
+		return v3.ValueTypeUint8, v, nil
+	case reflect.Uint16:
+		return v3.ValueTypeUint16, v, nil
+	case reflect.Uint32:
+		return v3.ValueTypeUint32, v, nil
+	case reflect.Uint64:
+		return v3.ValueTypeUint64, v, nil
+	case reflect.Uint:
+		return v3.ValueTypeUint64, uint64(v.(uint)), nil
+	case reflect.Int8:
+		return v3.ValueTypeInt8, v, nil
+	case reflect.Int16:
+		return v3.ValueTypeInt16, v, nil
+	case reflect.Int32:
+		return v3.ValueTypeInt32, v, nil
+	case reflect.Int64:
+		return v3.ValueTypeInt64, v, nil
+	case reflect.Int:
+		return v3.ValueTypeInt64, int64(v.(int)), nil
+	case reflect.Float32:
+		return v3.ValueTypeFloat32, v, nil
+	case reflect.Float64:
+		return v3.ValueTypeFloat64, v, nil
+	case reflect.Slice:
+		switch arrayValue := v.(type) {
+		case []any:
+			if len(arrayValue) > 0 {
+				kt := reflect.TypeOf(arrayValue[0])
+				if kt == nil {
+					return "", nil, fmt.Errorf("unsupported value %v(%s), the first element is nil", v, k)
+				}
+				switch kt.Kind() {
+				case reflect.Bool:
+					result := make([]bool, len(arrayValue))
+					for i, av := range arrayValue {
+						temp, ok := av.(bool)
+						if !ok {
+							return "", nil, fmt.Errorf("unable to cast value to []bool for %v", v)
+						}
+						result[i] = temp
+					}
+					return v3.ValueTypeBoolArray, result, nil
+				case reflect.String:
+					result := make([]string, len(arrayValue))
+					for i, av := range arrayValue {
+						temp, ok := av.(string)
+						if !ok {
+							return "", nil, fmt.Errorf("unable to cast value to []string for %v", v)
+						}
+						result[i] = temp
+					}
+					return v3.ValueTypeStringArray, result, nil
+				case reflect.Int8:
+					result := make([]int8, len(arrayValue))
+					for i, av := range arrayValue {
+						temp, ok := av.(int8)
+						if !ok {
+							return "", nil, fmt.Errorf("unable to cast value to []int8 for %v", v)
+						}
+						result[i] = temp
+					}
+					return v3.ValueTypeInt8Array, result, nil
+				case reflect.Int16:
+					result := make([]int16, len(arrayValue))
+					for i, av := range arrayValue {
+						temp, ok := av.(int16)
+						if !ok {
+							return "", nil, fmt.Errorf("unable to cast value to []int16 for %v", v)
+						}
+						result[i] = temp
+					}
+					return v3.ValueTypeInt16Array, result, nil
+				case reflect.Int32:
+					result := make([]int32, len(arrayValue))
+					for i, av := range arrayValue {
+						temp, ok := av.(int32)
+						if !ok {
+							return "", nil, fmt.Errorf("unable to cast value to []int32 for %v", v)
+						}
+						result[i] = temp
+					}
+					return v3.ValueTypeInt32Array, result, nil
+				case reflect.Int64, reflect.Int:
+					result := make([]int64, len(arrayValue))
+					for i, av := range arrayValue {
+						temp, ok := av.(int64)
+						if !ok {
+							return "", nil, fmt.Errorf("unable to cast value to []int64 for %v", v)
+						}
+						result[i] = temp
+					}
+					return v3.ValueTypeInt64Array, result, nil
+				case reflect.Uint8:
+					result := make([]uint8, len(arrayValue))
+					for i, av := range arrayValue {
+						temp, ok := av.(uint8)
+						if !ok {
+							return "", nil, fmt.Errorf("unable to cast value to []uint8 for %v", v)
+						}
+						result[i] = temp
+					}
+					return v3.ValueTypeUint8Array, result, nil
+				case reflect.Uint16:
+					result := make([]uint16, len(arrayValue))
+					for i, av := range arrayValue {
+						temp, ok := av.(uint16)
+						if !ok {
+							return "", nil, fmt.Errorf("unable to cast value to []uint16 for %v", v)
+						}
+						result[i] = temp
+					}
+					return v3.ValueTypeUint16Array, result, nil
+				case reflect.Uint32:
+					result := make([]uint32, len(arrayValue))
+					for i, av := range arrayValue {
+						temp, ok := av.(uint32)
+						if !ok {
+							return "", nil, fmt.Errorf("unable to cast value to []uint32 for %v", v)
+						}
+						result[i] = temp
+					}
+					return v3.ValueTypeUint32Array, result, nil
+				case reflect.Uint64, reflect.Uint:
+					result := make([]uint64, len(arrayValue))
+					for i, av := range arrayValue {
+						temp, ok := av.(uint64)
+						if !ok {
+							return "", nil, fmt.Errorf("unable to cast value to []uint64 for %v", v)
+						}
+						result[i] = temp
+					}
+					return v3.ValueTypeUint64Array, result, nil
+				case reflect.Float32:
+					result := make([]float32, len(arrayValue))
+					for i, av := range arrayValue {
+						temp, ok := av.(float32)
+						if !ok {
+							return "", nil, fmt.Errorf("unable to cast value to []float32 for %v", v)
+						}
+						result[i] = temp
+					}
+					return v3.ValueTypeFloat64Array, result, nil
+				case reflect.Float64:
+					result := make([]float64, len(arrayValue))
+					for i, av := range arrayValue {
+						temp, ok := av.(float64)
+						if !ok {
+							return "", nil, fmt.Errorf("unable to cast value to []float64 for %v", v)
+						}
+						result[i] = temp
+					}
+					return v3.ValueTypeFloat64Array, result, nil
+				}
+			} else { // default to string array
+				return v3.ValueTypeStringArray, []string{}, nil
+			}
+		case []byte:
+			return v3.ValueTypeBinary, v, nil
+		}
+	}
+	return v3.ValueTypeObject, v, nil
+}
+
+func getValueByType(v any, vt string) (any, error) {
+	switch vt {
+	case v3.ValueTypeBool:
+		return cast.ToBool(v, cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeInt8:
+		return cast.ToInt8(v, cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeInt16:
+		return cast.ToInt16(v, cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeInt32:
+		return cast.ToInt32(v, cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeInt64:
+		return cast.ToInt64(v, cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeUint8:
+		return cast.ToUint8(v, cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeUint16:
+		return cast.ToUint16(v, cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeUint32:
+		return cast.ToUint32(v, cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeUint64:
+		return cast.ToUint64(v, cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeFloat32:
+		return cast.ToFloat32(v, cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeFloat64:
+		return cast.ToFloat64(v, cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeString:
+		return cast.ToString(v, cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeBoolArray:
+		return cast.ToBoolSlice(v, cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeInt8Array:
+		return cast.ToTypedSlice(v, func(input any, sn cast.Strictness) (any, error) {
+			return cast.ToInt8(input, sn)
+		}, "int8", cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeInt16Array:
+		return cast.ToTypedSlice(v, func(input any, sn cast.Strictness) (any, error) {
+			return cast.ToInt16(input, sn)
+		}, "int16", cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeInt32Array:
+		return cast.ToTypedSlice(v, func(input any, sn cast.Strictness) (any, error) {
+			return cast.ToInt32(input, sn)
+		}, "int32", cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeInt64Array:
+		return cast.ToInt64Slice(v, cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeUint8Array:
+		return cast.ToTypedSlice(v, func(input any, sn cast.Strictness) (any, error) {
+			return cast.ToUint8(input, sn)
+		}, "uint8", cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeUint16Array:
+		return cast.ToTypedSlice(v, func(input any, sn cast.Strictness) (any, error) {
+			return cast.ToUint16(input, sn)
+		}, "uint16", cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeUint32Array:
+		return cast.ToTypedSlice(v, func(input any, sn cast.Strictness) (any, error) {
+			return cast.ToUint32(input, sn)
+		}, "uint32", cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeUint64Array:
+		return cast.ToUint64Slice(v, cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeFloat32Array:
+		return cast.ToTypedSlice(v, func(input any, sn cast.Strictness) (any, error) {
+			return cast.ToFloat32(input, sn)
+		}, "float32", cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeFloat64Array:
+		return cast.ToFloat64Slice(v, cast.CONVERT_SAMEKIND, cast.FORCE_CONVERT)
+	case v3.ValueTypeStringArray:
+		return cast.ToStringSlice(v, cast.CONVERT_SAMEKIND)
+	case v3.ValueTypeBinary:
+		var (
+			bv  []byte
+			err error
+		)
+		switch vv := v.(type) {
+		case string:
+			if bv, err = base64.StdEncoding.DecodeString(vv); err != nil {
+				return nil, fmt.Errorf("fail to decode binary value from %s: %v", vv, err)
+			}
+		case []byte:
+			bv = vv
+		default:
+			return nil, fmt.Errorf("fail to decode binary value from %v: not binary type", vv)
+		}
+		return bv, nil
+	case v3.ValueTypeObject:
+		return v, nil
+	default:
+		return nil, fmt.Errorf("unsupported type %v", vt)
+	}
+}
+
+func (ems *EdgexMsgBusSink) getMeta(result []map[string]any) *meta {
+	if ems.c.Metadata == "" {
+		return newMetaFromMap(nil)
+	}
+	// Try to get the meta field
+	for _, v := range result {
+		if m, ok := v[ems.c.Metadata]; ok {
+			if m1, ok1 := m.(map[string]any); ok1 {
+				return newMetaFromMap(m1)
+			} else {
+				conf.Log.Infof("Specified a meta field, but the field does not contains any EdgeX metadata.")
+			}
+		}
+	}
+	return newMetaFromMap(nil)
+}
+
+func (ems *EdgexMsgBusSink) Collect(ctx api.StreamContext, data api.MessageTuple) error {
+	return ems.doCollect(ctx, data.ToMap())
+}
+
+func (ems *EdgexMsgBusSink) CollectList(ctx api.StreamContext, data api.MessageTupleList) error {
+	return ems.doCollect(ctx, data.ToMaps())
+}
+
+func (ems *EdgexMsgBusSink) doCollect(ctx api.StreamContext, item any) error {
+	evt, err := ems.produceEvents(ctx, item)
+	if err != nil {
+		return fmt.Errorf("Failed to convert to EdgeX event: %s.", err.Error())
+	}
+	var (
+		data  []byte
+		topic string
+	)
+	if ems.c.MessageType == MessageTypeRequest {
+		req := requests.NewAddEventRequest(*evt)
+		data, _, err = req.Encode()
+		if err != nil {
+			return fmt.Errorf("unexpected error encode event %v", err)
+		}
+	} else {
+		data, err = json.Marshal(evt)
+		if err != nil {
+			return fmt.Errorf("unexpected error MarshalEvent %v", err)
+		}
+	}
+
+	if ems.topic == "" { // dynamic topic
+		topic = fmt.Sprintf("%s/%s/%s/%s", ems.c.TopicPrefix, evt.ProfileName, evt.DeviceName, evt.SourceName)
+	} else {
+		topic = ems.topic
+	}
+
+	env := types.NewMessageEnvelope(data, ctx)
+	env.ContentType = "application/json"
+	if pk, ok := ems.sendParams["contentType"]; ok {
+		if v, ok := pk.(string); ok {
+			env.ContentType = v
+		}
+	}
+	e := ems.cli.Publish(env, topic)
+	if e != nil {
+		ctx.GetLogger().Errorf("%s: found error %s when publish to EdgeX message bus.\n", e.Error(), e.Error())
+		return errorx.NewIOErr(e.Error())
+	}
+	ctx.GetLogger().Debugf("Published %+v to EdgeX message bus topic %s", evt, topic)
+	return nil
+}
+
+func (ems *EdgexMsgBusSink) Close(ctx api.StreamContext) error {
+	logger := ctx.GetLogger()
+	logger.Infof("Closing edgex sink")
+	if ems.cli != nil {
+		id := fmt.Sprintf("%s-%s-%d-edgex-sink", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId())
+		err := connection.DetachConnection(ctx, id, ems.config)
+		return err
+	}
+	return nil
+}
+
+type eventMeta struct {
+	id          *string
+	deviceName  string
+	profileName string
+	sourceName  string
+	origin      *int64
+	tags        map[string]any
+}
+
+type readingMeta struct {
+	id           *string
+	deviceName   *string
+	profileName  *string
+	resourceName *string
+	origin       *int64
+	valueType    *string
+	mediaType    *string
+}
+
+func (m *readingMeta) decorate(r *dtos.BaseReading) dtos.BaseReading {
+	if m.id != nil {
+		r.Id = *m.id
+	}
+	if m.deviceName != nil {
+		r.DeviceName = *m.deviceName
+	}
+	if m.profileName != nil {
+		r.ProfileName = *m.profileName
+	}
+	if m.origin != nil {
+		r.Origin = *m.origin
+	}
+	if m.valueType != nil {
+		r.ValueType = *m.valueType
+	}
+	if m.mediaType != nil {
+		r.MediaType = *m.mediaType
+	}
+	return *r
+}
+
+type meta struct {
+	eventMeta
+	readingMetas map[string]any
+}
+
+func newMetaFromMap(m1 map[string]any) *meta {
+	result := &meta{
+		eventMeta: eventMeta{},
+	}
+	for k, v := range m1 {
+		switch k {
+		case "id":
+			if v1, ok := v.(string); ok {
+				result.id = &v1
+			}
+		case "deviceName":
+			if v1, ok := v.(string); ok {
+				result.deviceName = v1
+			}
+		case "profileName":
+			if v1, ok := v.(string); ok {
+				result.profileName = v1
+			}
+		case "sourceName":
+			if v1, ok := v.(string); ok {
+				result.sourceName = v1
+			}
+		case "origin":
+			if v1, ok := v.(float64); ok {
+				temp := int64(v1)
+				result.origin = &temp
+			}
+		case "tags":
+			if v1, ok1 := v.(map[string]any); ok1 {
+				result.tags = v1
+			}
+		default:
+			if result.readingMetas == nil {
+				result.readingMetas = make(map[string]any)
+			}
+			result.readingMetas[k] = v
+		}
+	}
+	return result
+}
+
+func (m *meta) readingMeta(ctx api.StreamContext, readingName string) *readingMeta {
+	vi, ok := m.readingMetas[readingName]
+	if !ok {
+		return nil
+	}
+	m1, ok := vi.(map[string]any)
+	if !ok {
+		ctx.GetLogger().Errorf("reading %s meta is not a map, but %v", readingName, vi)
+		return nil
+	}
+	result := &readingMeta{}
+	for k, v := range m1 {
+		switch k {
+		case "id":
+			if v1, ok := v.(string); ok {
+				result.id = &v1
+			}
+		case "deviceName":
+			if v1, ok := v.(string); ok {
+				result.deviceName = &v1
+			}
+		case "profileName":
+			if v1, ok := v.(string); ok {
+				result.profileName = &v1
+			}
+		case "resourceName":
+			if v1, ok := v.(string); ok {
+				result.resourceName = &v1
+			}
+		case "origin":
+			if v1, ok := v.(float64); ok {
+				temp := int64(v1)
+				result.origin = &temp
+			}
+		case "valueType":
+			if v1, ok := v.(string); ok {
+				result.valueType = &v1
+			}
+		case "mediaType":
+			if v1, ok := v.(string); ok {
+				result.mediaType = &v1
+			}
+		default:
+			ctx.GetLogger().Warnf("reading %s meta got unknown field %s of value %v", readingName, k, v)
+		}
+	}
+	return result
+}
+
+func (m *meta) createEvent() *dtos.Event {
+	event := dtos.NewEvent(m.profileName, m.deviceName, m.sourceName)
+	if m.id != nil {
+		event.Id = *m.id
+	}
+	if m.origin != nil {
+		event.Origin = *m.origin
+	}
+	if m.tags != nil {
+		event.Tags = m.tags
+	}
+	return &event
+}
+
+func GetSink() api.Sink {
+	return &EdgexMsgBusSink{}
+}
diff --git a/internal/io/edgex/source.go b/internal/io/edgex/source.go
new file mode 100644
index 00000000..7464714f
--- /dev/null
+++ b/internal/io/edgex/source.go
@@ -0,0 +1,331 @@
+// 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.
+
+package edgex
+
+import (
+	"encoding/json"
+	"fmt"
+	"strconv"
+	"strings"
+
+	v3 "github.com/edgexfoundry/go-mod-core-contracts/v3/common"
+	"github.com/edgexfoundry/go-mod-core-contracts/v3/dtos"
+	"github.com/edgexfoundry/go-mod-core-contracts/v3/dtos/requests"
+	"github.com/edgexfoundry/go-mod-messaging/v3/pkg/types"
+	"github.com/fxamacker/cbor/v2"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/io/edgex/client"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+type Source struct {
+	cli *client.Client
+
+	config      map[string]any
+	topic       string
+	messageType messageType
+	buflen      int
+}
+
+type SourceConf struct {
+	Topic       string      `json:"topic"`
+	MessageType messageType `json:"messageType"`
+	BufferLen   int         `json:"bufferLength"`
+}
+
+type SubConf struct {
+	Topic string `json:"topic"`
+}
+
+type messageType string
+
+const (
+	MessageTypeEvent   messageType = "event"
+	MessageTypeRequest messageType = "request"
+)
+
+func (es *Source) Provision(_ api.StreamContext, props map[string]any) error {
+	c := &SourceConf{
+		MessageType: MessageTypeEvent,
+	}
+	err := cast.MapToStruct(props, c)
+	if err != nil {
+		return fmt.Errorf("read properties %v fail with error: %v", props, err)
+	}
+	if c.BufferLen <= 0 {
+		c.BufferLen = 1024
+	}
+	es.buflen = c.BufferLen
+	es.messageType = c.MessageType
+	es.topic = c.Topic
+	es.config = props
+	return nil
+}
+
+func (es *Source) Connect(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("Connecting to edgex server")
+	var cli *client.Client
+	var err error
+	id := fmt.Sprintf("%s-%s-%d-edgex-source", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId())
+	conn, err := connection.FetchConnection(ctx, id, "edgex", es.config)
+	if err != nil {
+		return err
+	}
+	cli = conn.(*client.Client)
+	es.cli = cli
+	return err
+}
+
+func (es *Source) SubId(props map[string]any) string {
+	sc := &SubConf{}
+	err := cast.MapToStruct(props, sc)
+	if err != nil {
+		return ""
+	}
+	return sc.Topic
+}
+
+func (es *Source) Subscribe(ctx api.StreamContext, ingest api.TupleIngest, ingestErr api.ErrorIngest) error {
+	log := ctx.GetLogger()
+
+	message := make(chan types.MessageEnvelope, es.buflen)
+	errChan := make(chan error)
+
+	if e := es.cli.Subscribe(message, es.topic, errChan); e != nil {
+		log.Errorf("Failed to subscribe to edgex messagebus topic %s.", e)
+		return e
+	} else {
+		log.Infof("Successfully subscribed to edgex messagebus topic %s.", es.topic)
+		for {
+			select {
+			case <-ctx.Done():
+				log.Infof("Exit subscription to edgex messagebus topic %s.", es.topic)
+				return nil
+			case e1 := <-errChan:
+				log.Errorf("Subscription to edgex messagebus received error %v.", e1)
+				ingestErr(ctx, e1)
+			case env, ok := <-message:
+				rcvTime := timex.GetNow()
+				if !ok { // the source is closed
+					log.Infof("Exit subscription to edgex messagebus topic %s.", es.topic)
+					return nil
+				}
+
+				var r any
+				switch es.messageType {
+				case MessageTypeEvent:
+					r = &dtos.Event{}
+				case MessageTypeRequest:
+					r = &requests.AddEventRequest{}
+				}
+
+				if strings.EqualFold(env.ContentType, "application/json") {
+					if err := json.Unmarshal(env.Payload, r); err != nil {
+						l := len(env.Payload)
+						if l > 200 {
+							l = 200
+						}
+						log.Warnf("payload %s unmarshal fail: %v", env.Payload[0:(l-1)], err)
+						break
+					}
+				} else if strings.EqualFold(env.ContentType, "application/cbor") {
+					if err := cbor.Unmarshal(env.Payload, r); err != nil {
+						l := len(env.Payload)
+						if l > 200 {
+							l = 200
+						}
+						log.Warnf("payload %s unmarshal fail: %v", env.Payload[0:(l-1)], err)
+						break
+					}
+				} else {
+					log.Errorf("Unsupported data type %s.", env.ContentType)
+					break
+				}
+
+				result := make(map[string]any)
+				meta := make(map[string]any)
+				var e *dtos.Event
+				switch t := r.(type) {
+				case *dtos.Event:
+					e = t
+				case *requests.AddEventRequest:
+					e = &t.Event
+				}
+
+				log.Debugf("receive message %s from device %s", env.Payload, e.DeviceName)
+				for _, r := range e.Readings {
+					if r.ResourceName != "" {
+						if v, err := es.getValue(r, log); err != nil {
+							log.Warnf("fail to get value for %s: %v", r.ResourceName, err)
+						} else {
+							result[r.ResourceName] = v
+						}
+						rMeta := map[string]any{}
+						rMeta["id"] = r.Id
+						// r_meta["created"] = r.Created
+						// r_meta["modified"] = r.Modified
+						rMeta["origin"] = r.Origin
+						// r_meta["pushed"] = r.Pushed
+						rMeta["deviceName"] = r.DeviceName
+						rMeta["profileName"] = r.ProfileName
+						rMeta["valueType"] = r.ValueType
+						if r.MediaType != "" {
+							rMeta["mediaType"] = r.MediaType
+						}
+						meta[r.ResourceName] = rMeta
+					} else {
+						log.Warnf("The name of readings should not be empty!")
+					}
+				}
+				if len(result) > 0 {
+					meta["id"] = e.Id
+					// meta["pushed"] = e.Pushed
+					meta["deviceName"] = e.DeviceName
+					meta["profileName"] = e.ProfileName
+					meta["sourceName"] = e.SourceName
+					// meta["created"] = e.Created
+					// meta["modified"] = e.Modified
+					meta["origin"] = e.Origin
+					meta["tags"] = e.Tags
+					meta["correlationid"] = env.CorrelationID
+
+					ingest(ctx, result, meta, rcvTime)
+				} else {
+					log.Warnf("No readings are processed for the event, so ignore it.")
+				}
+			}
+		}
+	}
+}
+
+func (es *Source) getValue(r dtos.BaseReading, logger api.Logger) (any, error) {
+	t := r.ValueType
+	logger.Debugf("name %s with type %s", r.ResourceName, r.ValueType)
+	v := r.Value
+	switch t {
+	case v3.ValueTypeBool:
+		if r, err := strconv.ParseBool(v); err != nil {
+			return nil, err
+		} else {
+			return r, nil
+		}
+	case v3.ValueTypeInt8, v3.ValueTypeInt16, v3.ValueTypeInt32, v3.ValueTypeInt64, v3.ValueTypeUint8, v3.ValueTypeUint16, v3.ValueTypeUint32:
+		if r, err := strconv.Atoi(v); err != nil {
+			return nil, err
+		} else {
+			return r, nil
+		}
+	case v3.ValueTypeUint64:
+		if u64, err := strconv.ParseUint(v, 10, 64); err != nil {
+			return nil, err
+		} else {
+			return u64, nil
+		}
+	case v3.ValueTypeFloat32:
+		if r, err := strconv.ParseFloat(v, 32); err != nil {
+			return nil, err
+		} else {
+			return r, nil
+		}
+	case v3.ValueTypeFloat64:
+		if r, err := strconv.ParseFloat(v, 64); err != nil {
+			return nil, err
+		} else {
+			return r, nil
+		}
+	case v3.ValueTypeString:
+		return v, nil
+	case v3.ValueTypeBoolArray:
+		var val []bool
+		if e := json.Unmarshal(cast.StringToBytes(v), &val); e == nil {
+			return val, nil
+		} else {
+			return nil, e
+		}
+	case v3.ValueTypeInt8Array, v3.ValueTypeInt16Array, v3.ValueTypeInt32Array, v3.ValueTypeInt64Array, v3.ValueTypeUint8Array, v3.ValueTypeUint16Array, v3.ValueTypeUint32Array:
+		var val []int
+		if e := json.Unmarshal(cast.StringToBytes(v), &val); e == nil {
+			return val, nil
+		} else {
+			return nil, e
+		}
+	case v3.ValueTypeUint64Array:
+		var val []uint64
+		if e := json.Unmarshal(cast.StringToBytes(v), &val); e == nil {
+			return val, nil
+		} else {
+			return nil, e
+		}
+	case v3.ValueTypeFloat32Array:
+		return convertFloatArray(v, 32)
+	case v3.ValueTypeFloat64Array:
+		return convertFloatArray(v, 64)
+	case v3.ValueTypeStringArray:
+		var val []string
+		if e := json.Unmarshal(cast.StringToBytes(v), &val); e == nil {
+			return val, nil
+		} else {
+			return nil, e
+		}
+	case v3.ValueTypeBinary:
+		return r.BinaryValue, nil
+	case v3.ValueTypeObject:
+		return r.ObjectValue, nil
+	default:
+		logger.Warnf("Not supported type %s, and processed as string value", t)
+		return v, nil
+	}
+}
+
+func convertFloatArray(v string, bitSize int) (any, error) {
+	var val1 []string
+	if e := json.Unmarshal(cast.StringToBytes(v), &val1); e == nil {
+		var ret []float64
+		for _, v := range val1 {
+			if fv, err := strconv.ParseFloat(v, bitSize); err != nil {
+				return nil, err
+			} else {
+				ret = append(ret, fv)
+			}
+		}
+		return ret, nil
+	} else {
+		var val []float64
+		if e := json.Unmarshal(cast.StringToBytes(v), &val); e == nil {
+			return val, nil
+		} else {
+			return nil, e
+		}
+	}
+}
+
+func (es *Source) Close(ctx api.StreamContext) error {
+	log := ctx.GetLogger()
+	log.Infof("EdgeX Source instance %d Done.", ctx.GetInstanceId())
+	if es.cli != nil {
+		id := fmt.Sprintf("%s-%s-%d-edgex-source", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId())
+		err := connection.DetachConnection(ctx, id, es.config)
+		es.cli.DetachSub(ctx, es.config)
+		return err
+	}
+	return nil
+}
+
+func GetSource() api.Source {
+	return &Source{}
+}
diff --git a/internal/io/file/sink.go b/internal/io/file/sink.go
index f0669e5a..07497b07 100644
--- a/internal/io/file/sink.go
+++ b/internal/io/file/sink.go
@@ -1,4 +1,4 @@
-// Copyright 2021-2023 EMQ Technologies Co., Ltd.
+// 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.
@@ -15,10 +15,11 @@
 package file
 
 import (
+	"bytes"
+	"encoding/binary"
 	"errors"
 	"fmt"
 	"path/filepath"
-	"sort"
 	"strings"
 	"sync"
 	"time"
@@ -55,6 +56,7 @@ type fileSink struct {
 	mux      sync.Mutex
 	fws      map[string]*fileWriter
 	rollHook modules.RollHook
+	headers  string
 }
 
 func (m *fileSink) Provision(ctx api.StreamContext, props map[string]interface{}) error {
@@ -166,7 +168,7 @@ func (m *fileSink) Collect(ctx api.StreamContext, tuple api.RawTuple) error {
 		}
 	}
 	ctx.GetLogger().Debugf("writing to file path %s", fn)
-	fw, err := m.GetFws(ctx, fn, item)
+	fw, item, err := m.GetFws(ctx, fn, item)
 	if err != nil {
 		return err
 	}
@@ -234,41 +236,31 @@ func (m *fileSink) roll(ctx api.StreamContext, k string, v *fileWriter) error {
 
 // GetFws returns the file writer for the given file name, if the file writer does not exist, it will create one
 // The item is used to get the csv header if needed
-func (m *fileSink) GetFws(ctx api.StreamContext, fn string, item interface{}) (*fileWriter, error) {
+func (m *fileSink) GetFws(ctx api.StreamContext, fn string, item []byte) (*fileWriter, []byte, error) {
 	m.mux.Lock()
 	defer m.mux.Unlock()
+	if m.c.FileType == CSV_TYPE && m.c.HasHeader && m.headers == "" {
+		if len(m.c.Fields) > 0 {
+			m.headers = strings.Join(m.c.Fields, m.c.Delimiter)
+		} else {
+			db := []byte(m.c.Delimiter)
+			if bytes.HasPrefix(item, db) {
+				cursor := len(db)
+				nextCursor := cursor + 4
+				hl := binary.BigEndian.Uint32(item[cursor:nextCursor])
+				cursor = nextCursor
+				nextCursor = cursor + int(hl)
+				hb := item[cursor:nextCursor]
+				m.headers = string(hb)
+				ctx.GetLogger().Infof("csv header %s", hb)
+				cursor = nextCursor
+				item = item[nextCursor:]
+			}
+		}
+	}
 	fws, ok := m.fws[fn]
 	if !ok {
 		var e error
-		// extract header for csv
-		var headers string
-		if m.c.FileType == CSV_TYPE && m.c.HasHeader {
-			var header []string
-			if len(m.c.Fields) > 0 {
-				header = m.c.Fields
-			} else {
-				switch v := item.(type) {
-				case map[string]interface{}:
-					header = make([]string, len(v))
-					i := 0
-					for k := range item.(map[string]interface{}) {
-						header[i] = k
-						i++
-					}
-				case []map[string]interface{}:
-					if len(v) > 0 {
-						header = make([]string, len(v[0]))
-						i := 0
-						for k := range v[0] {
-							header[i] = k
-							i++
-						}
-					}
-				}
-				sort.Strings(header)
-			}
-			headers = strings.Join(header, m.c.Delimiter)
-		}
 		nfn := fn
 		if m.c.RollingNamePattern != "" {
 			newFile := ""
@@ -286,13 +278,13 @@ func (m *fileSink) GetFws(ctx api.StreamContext, fn string, item interface{}) (*
 			nfn = filepath.Join(fileDir, newFile)
 		}
 
-		fws, e = m.createFileWriter(ctx, nfn, m.c.FileType, headers, m.c.Compression, m.c.Encryption)
+		fws, e = m.createFileWriter(ctx, nfn, m.c.FileType, m.headers, m.c.Compression, m.c.Encryption)
 		if e != nil {
-			return nil, e
+			return nil, item, e
 		}
 		m.fws[fn] = fws
 	}
-	return fws, nil
+	return fws, item, nil
 }
 
 func GetSink() api.Sink {
diff --git a/internal/io/neuron/source.go b/internal/io/neuron/source.go
index 569588e7..bac3a785 100644
--- a/internal/io/neuron/source.go
+++ b/internal/io/neuron/source.go
@@ -54,7 +54,7 @@ func (s *source) Ping(ctx api.StreamContext, props map[string]any) error {
 	return ping(ctx, props)
 }
 
-func (s *source) SubId(props map[string]any) string {
+func (s *source) ConnId(props map[string]any) string {
 	var url string
 	u, ok := props["url"]
 	if ok {
@@ -63,6 +63,10 @@ func (s *source) SubId(props map[string]any) string {
 	return "nng:" + PROTOCOL + url
 }
 
+func (s *source) SubId(_ map[string]any) string {
+	return "singleton"
+}
+
 func (s *source) Connect(ctx api.StreamContext) error {
 	cli, err := connect(ctx, s.c.Url, s.props)
 	if err != nil {
diff --git a/internal/pkg/def/rule.go b/internal/pkg/def/rule.go
index 2d6ea646..9fe89520 100644
--- a/internal/pkg/def/rule.go
+++ b/internal/pkg/def/rule.go
@@ -80,7 +80,7 @@ type RuleGraph struct {
 // Rule the definition of the business logic
 // Sql and Graph are mutually exclusive, at least one of them should be set
 type Rule struct {
-	Triggered bool                     `json:"triggered,omitempty" yaml:"triggered,omitempty"`
+	Triggered bool                     `json:"triggered" yaml:"triggered"`
 	Id        string                   `json:"id,omitempty" yaml:"id,omitempty"`
 	Name      string                   `json:"name,omitempty" yaml:"name,omitempty"` // The display name of a rule
 	Sql       string                   `json:"sql,omitempty" yaml:"sql,omitempty"`
diff --git a/internal/topo/node/encode_op.go b/internal/topo/node/encode_op.go
index 5ba4ae33..696d9b4e 100644
--- a/internal/topo/node/encode_op.go
+++ b/internal/topo/node/encode_op.go
@@ -35,7 +35,7 @@ type EncodeOp struct {
 }
 
 func NewEncodeOp(ctx api.StreamContext, name string, rOpt *def.RuleOption, sc *SinkConf) (*EncodeOp, error) {
-	c, err := converter.GetOrCreateConverter(ctx, sc.Format, sc.SchemaId, nil, map[string]any{"delimiter": sc.Delimiter})
+	c, err := converter.GetOrCreateConverter(ctx, sc.Format, sc.SchemaId, nil, map[string]any{"delimiter": sc.Delimiter, "hasHeader": sc.HasHeader})
 	if err != nil {
 		return nil, err
 	}
diff --git a/internal/topo/node/props.go b/internal/topo/node/props.go
index 698f10d8..6e843412 100644
--- a/internal/topo/node/props.go
+++ b/internal/topo/node/props.go
@@ -38,6 +38,7 @@ type SinkConf struct {
 	LingerInterval cast.DurationConf `json:"lingerInterval"`
 	Compression    string            `json:"compression"`
 	Encryption     string            `json:"encryption"`
+	HasHeader      bool              `json:"hasHeader"`
 	conf.SinkConf
 }
 
diff --git a/internal/topo/operator/emitter_operator.go b/internal/topo/operator/emitter_operator.go
new file mode 100644
index 00000000..9399751a
--- /dev/null
+++ b/internal/topo/operator/emitter_operator.go
@@ -0,0 +1,42 @@
+// 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.
+
+package operator
+
+import (
+	"fmt"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/xsql"
+)
+
+// EmitterOp set the emitter to the stream name
+// It is only planned after shared connection node in which the emitter is not determined in the source side
+type EmitterOp struct {
+	Emitter string
+}
+
+func (p *EmitterOp) Apply(ctx api.StreamContext, data any, _ *xsql.FunctionValuer, _ *xsql.AggregateFunctionValuer) any {
+	ctx.GetLogger().Debugf("emitter op receive %v", data)
+	switch input := data.(type) {
+	case *xsql.RawTuple:
+		input.Emitter = p.Emitter
+		return input
+	case *xsql.Tuple:
+		input.Emitter = p.Emitter
+		return input
+	default:
+		return fmt.Errorf("run emitter op error: invalid input %[1]T(%[1]v)", input)
+	}
+}
diff --git a/internal/topo/planner/planner_source.go b/internal/topo/planner/planner_source.go
index 497a7941..d31388b4 100644
--- a/internal/topo/planner/planner_source.go
+++ b/internal/topo/planner/planner_source.go
@@ -80,15 +80,27 @@ func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, option
 	)
 	// Some connection only allow one subscription. The source should implement UniqueSub to provide a subId to avoid multiple connection.
 	us, hasSubId := ss.(model.UniqueSub)
-	if sp.SelId == "" && !hasSubId {
+	conId := sp.SelId
+	// Some connection only have on
+	cs, hasConId := ss.(model.UniqueConn)
+	if hasConId {
+		conId = cs.ConnId(props)
+	}
+
+	var ops []node.OperatorNode
+	// If having unique connection id AND unique sub id for each connection, need to share the sub node
+	if conId == "" {
 		srcConnNode, err = node.NewSourceNode(ctx, string(t.name), ss, props, options)
+		if err != nil {
+			return nil, nil, 0, err
+		}
+		index++
 	} else { // connection selector is set as a one node sub_topo
-		var selName string
-		if sp.SelId != "" {
-			selName = fmt.Sprintf("%s/%s", sp.SelId, t.streamStmt.Options.DATASOURCE)
-		} else {
-			selName = us.SubId(props)
+		subId := t.streamStmt.Options.DATASOURCE
+		if hasSubId {
+			subId = us.SubId(props)
 		}
+		selName := fmt.Sprintf("%s/%s", conId, subId)
 		srcSubtopo, existed := topo.GetSubTopo(selName)
 		if !existed {
 			var scn node.DataSourceNode
@@ -99,13 +111,15 @@ func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, option
 			}
 		}
 		srcConnNode = srcSubtopo
+		if err != nil {
+			return nil, nil, 0, err
+		}
+		index++
+		// another node to set emitter
+		op := Transform(&operator.EmitterOp{Emitter: string(t.name)}, fmt.Sprintf("%d_emitter", index), options)
+		index++
+		ops = append(ops, op)
 	}
-	if err != nil {
-		return nil, nil, 0, err
-	}
-	index++
-	var ops []node.OperatorNode
-
 	// Need to check after source has provisioned, so do not put it before provision
 	featureSet, err := checkFeatures(ss, sp, props)
 	if err != nil {
diff --git a/internal/topo/rule/ruleState.go b/internal/topo/rule/ruleState.go
index 06d920d7..8ca05a20 100644
--- a/internal/topo/rule/ruleState.go
+++ b/internal/topo/rule/ruleState.go
@@ -128,6 +128,10 @@ func NewRuleState(rule *def.Rule) (rs *RuleState, err error) {
 		} else {
 			rs.Topology = tp
 		}
+		if !rs.Rule.Triggered {
+			// manually force stop rule
+			rs.Topology.Cancel()
+		}
 		return nil
 	})
 	rs.run()
diff --git a/pkg/model/io.go b/pkg/model/io.go
index 54407ce1..f522254b 100644
--- a/pkg/model/io.go
+++ b/pkg/model/io.go
@@ -45,3 +45,7 @@ type NodeInfo struct {
 type UniqueSub interface {
 	SubId(props map[string]any) string
 }
+
+type UniqueConn interface {
+	ConnId(props map[string]any) string
+}


Test patch:
diff --git a/internal/converter/delimited/converter_test.go b/internal/converter/delimited/converter_test.go
index a0c0b5fa..387fccce 100644
--- a/internal/converter/delimited/converter_test.go
+++ b/internal/converter/delimited/converter_test.go
@@ -19,6 +19,7 @@ import (
 	"reflect"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
@@ -28,17 +29,21 @@ import (
 
 func TestEncode(t *testing.T) {
 	tests := []struct {
-		m map[string]interface{}
-		r []byte
-		e string
+		name string
+		m    any
+		r    []byte
+		e    string
 	}{
 		{
+			name: "normal",
 			m: map[string]interface{}{
 				"id":   12,
 				"name": "test",
 			},
 			r: []byte(`12:test`),
-		}, {
+		},
+		{
+			name: "embedded",
 			m: map[string]interface{}{
 				"id":   7,
 				"name": "John Doe",
@@ -54,20 +59,97 @@ func TestEncode(t *testing.T) {
 			},
 			r: []byte(`22:map[indoor:[Chess] outdoor:[Basketball]]:7:John Doe`),
 		},
+		{
+			name: "list",
+			m: []map[string]interface{}{
+				{
+					"id":   12,
+					"name": "test",
+				},
+				{
+					"id":   14,
+					"name": "test2",
+				},
+			},
+			r: []byte("12:test\n14:test2"),
+		},
 	}
-	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
 	ctx := mockContext.NewMockContext("test", "op1")
-	for i, tt := range tests {
-		c, err := NewConverter(map[string]any{"delimiter": ":"})
-		if err != nil {
-			t.Fatal(err)
-		}
-		a, err := c.Encode(ctx, tt.m)
-		if !reflect.DeepEqual(tt.e, testx.Errstring(err)) {
-			t.Errorf("%d.error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.e, err)
-		} else if tt.e == "" && !reflect.DeepEqual(tt.r, a) {
-			t.Errorf("%d. \n\nresult mismatch:\n\nexp=%s\n\ngot=%s\n\n", i, tt.r, a)
-		}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c, err := NewConverter(map[string]any{"delimiter": ":"})
+			assert.NoError(t, err)
+			a, err := c.Encode(ctx, tt.m)
+			if tt.e != "" {
+				assert.EqualError(t, err, tt.e)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.r, a)
+			}
+		})
+	}
+}
+
+func TestEncodeWithHeader(t *testing.T) {
+	tests := []struct {
+		name string
+		m    any
+		r    []byte
+		e    string
+	}{
+		{
+			name: "normal",
+			m: map[string]interface{}{
+				"id":   12,
+				"name": "test",
+			},
+			r: []byte{0x3a, 0x0, 0x0, 0x0, 0x7, 0x69, 0x64, 0x3a, 0x6e, 0x61, 0x6d, 0x65, 0x31, 0x32, 0x3a, 0x74, 0x65, 0x73, 0x74},
+		},
+		{
+			name: "embedded",
+			m: map[string]interface{}{
+				"id":   7,
+				"name": "John Doe",
+				"age":  22,
+				"hobbies": map[string]interface{}{
+					"indoor": []string{
+						"Chess",
+					},
+					"outdoor": []string{
+						"Basketball",
+					},
+				},
+			},
+			r: []byte{0x3a, 0x0, 0x0, 0x0, 0x13, 0x61, 0x67, 0x65, 0x3a, 0x68, 0x6f, 0x62, 0x62, 0x69, 0x65, 0x73, 0x3a, 0x69, 0x64, 0x3a, 0x6e, 0x61, 0x6d, 0x65, 0x32, 0x32, 0x3a, 0x6d, 0x61, 0x70, 0x5b, 0x69, 0x6e, 0x64, 0x6f, 0x6f, 0x72, 0x3a, 0x5b, 0x43, 0x68, 0x65, 0x73, 0x73, 0x5d, 0x20, 0x6f, 0x75, 0x74, 0x64, 0x6f, 0x6f, 0x72, 0x3a, 0x5b, 0x42, 0x61, 0x73, 0x6b, 0x65, 0x74, 0x62, 0x61, 0x6c, 0x6c, 0x5d, 0x5d, 0x3a, 0x37, 0x3a, 0x4a, 0x6f, 0x68, 0x6e, 0x20, 0x44, 0x6f, 0x65},
+		},
+		{
+			name: "list",
+			m: []map[string]interface{}{
+				{
+					"id":   12,
+					"name": "test",
+				},
+				{
+					"id":   14,
+					"name": "test2",
+				},
+			},
+			r: []byte("id:name\n12:test\n14:test2"),
+		},
+	}
+	ctx := mockContext.NewMockContext("test", "op1")
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c, err := NewConverter(map[string]any{"delimiter": ":", "hasHeader": true})
+			assert.NoError(t, err)
+			a, err := c.Encode(ctx, tt.m)
+			if tt.e != "" {
+				assert.EqualError(t, err, tt.e)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.r, a)
+			}
+		})
 	}
 }
 
diff --git a/internal/io/edgex/client/edgex_test.go b/internal/io/edgex/client/edgex_test.go
new file mode 100644
index 00000000..3c4d43ab
--- /dev/null
+++ b/internal/io/edgex/client/edgex_test.go
@@ -0,0 +1,194 @@
+// 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 client
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/edgexfoundry/go-mod-messaging/v3/pkg/types"
+)
+
+func TestEdgex_CfgValidate(t *testing.T) {
+	tests := []struct {
+		name    string
+		expConf types.MessageBusConfig
+		props   map[string]interface{}
+		wantErr bool
+	}{
+		{
+			name: "config pass",
+			props: map[string]interface{}{
+				"protocol": "tcp",
+				"server":   "127.0.0.1",
+				"port":     1883,
+				"type":     "mqtt",
+				"optional": map[string]interface{}{
+					"ClientId":  "client1",
+					"Username":  "user1",
+					"KeepAlive": 500,
+				},
+			},
+			wantErr: false,
+			expConf: types.MessageBusConfig{
+				Broker: types.HostInfo{
+					Host:     "127.0.0.1",
+					Port:     1883,
+					Protocol: "tcp",
+				},
+				Type: "mqtt",
+				Optional: map[string]string{
+					"ClientId":  "client1",
+					"Username":  "user1",
+					"KeepAlive": "500",
+				},
+			},
+		},
+		{
+			name: "config pass",
+			props: map[string]interface{}{
+				"protocol": "redis",
+				"server":   "edgex-redis",
+				"port":     6379,
+				"type":     "redis",
+			},
+			wantErr: false,
+			expConf: types.MessageBusConfig{
+				Broker: types.HostInfo{
+					Host:     "edgex-redis",
+					Port:     6379,
+					Protocol: "redis",
+				},
+				Type: "redis",
+			},
+		},
+		{
+			name: "config not case sensitive",
+			props: map[string]interface{}{
+				"Protocol": "tcp",
+				"server":   "127.0.0.1",
+				"Port":     1883,
+				"type":     "mqtt",
+				"optional": map[string]string{
+					"ClientId": "client1",
+					"Username": "user1",
+				},
+			},
+			expConf: types.MessageBusConfig{
+				Broker: types.HostInfo{
+					Host:     "127.0.0.1",
+					Port:     1883,
+					Protocol: "tcp",
+				},
+				Type: "mqtt",
+				Optional: map[string]string{
+					"ClientId": "client1",
+					"Username": "user1",
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "config type not in mqtt/redis ",
+			props: map[string]interface{}{
+				"protocol": "tcp",
+				"server":   "127.0.0.1",
+				"port":     1883,
+				"type":     "kafka",
+				"optional": map[string]string{
+					"ClientId": "client1",
+					"Username": "user1",
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "do not have enough config items ",
+			props: map[string]interface{}{
+				"optional": map[string]string{
+					"ClientId": "client1",
+					"Username": "user1",
+				},
+			},
+			expConf: types.MessageBusConfig{
+				Broker: types.HostInfo{
+					Host:     "localhost",
+					Port:     6379,
+					Protocol: "redis",
+				},
+				Type: "redis",
+				Optional: map[string]string{
+					"ClientId": "client1",
+					"Username": "user1",
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "type is not right",
+			props: map[string]interface{}{
+				"type":     20,
+				"protocol": "redis",
+				"host":     "edgex-redis",
+				"port":     6379,
+				"optional": map[string]string{
+					"ClientId": "client1",
+					"Username": "user1",
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "port is not right",
+			props: map[string]interface{}{
+				"type":     "mqtt",
+				"protocol": "redis",
+				"host":     "edgex-redis",
+				"port":     -1,
+				"optional": map[string]string{
+					"ClientId": "client1",
+					"Username": "user1",
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "wrong type value",
+			props: map[string]interface{}{
+				"type":     "mqt",
+				"protocol": "redis",
+				"host":     "edgex-redis",
+				"port":     6379,
+				"optional": map[string]string{
+					"ClientId": "client1",
+					"Username": "user1",
+				},
+			},
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			es := &Client{}
+			if err := es.CfgValidate(tt.props); (err != nil) != tt.wantErr {
+				t.Errorf("CfgValidate() error = %v, wantErr %v", err, tt.wantErr)
+			} else {
+				if !reflect.DeepEqual(tt.expConf, es.mbconf) {
+					t.Errorf("CfgValidate() expect = %v, actual %v", tt.expConf, es.mbconf)
+				}
+			}
+		})
+	}
+}
diff --git a/internal/io/edgex/sink_test.go b/internal/io/edgex/sink_test.go
new file mode 100644
index 00000000..eb5ed38c
--- /dev/null
+++ b/internal/io/edgex/sink_test.go
@@ -0,0 +1,504 @@
+// 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.
+
+package edgex
+
+import (
+	"encoding/json"
+	"fmt"
+	"reflect"
+	"testing"
+
+	v3 "github.com/edgexfoundry/go-mod-core-contracts/v3/common"
+	"github.com/edgexfoundry/go-mod-core-contracts/v3/dtos"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/testx"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
+)
+
+var (
+	contextLogger = conf.Log.WithField("rule", "testEdgexSink")
+	ctx           = context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+)
+
+func compareEvent(expected, actual *dtos.Event) bool {
+	if (expected.Id == actual.Id || (expected.Id == "" && actual.Id != "")) && expected.ProfileName == actual.ProfileName && expected.DeviceName == actual.DeviceName && (expected.Origin == actual.Origin || (expected.Origin == 0 && actual.Origin > 0)) && reflect.DeepEqual(expected.Tags, actual.Tags) && expected.SourceName == actual.SourceName && len(expected.Readings) == len(actual.Readings) {
+		for _, r := range expected.Readings {
+			compared := false
+			for _, a := range actual.Readings {
+				if r.ResourceName == a.ResourceName {
+					compared = true
+					if !compareReading(r, a) {
+						return false
+					}
+				}
+			}
+			if !compared {
+				return false
+			}
+		}
+		return true
+	}
+	return false
+}
+
+func compareReading(expected, actual dtos.BaseReading) bool {
+	if (expected.Id == actual.Id || (expected.Id == "" && actual.Id != "")) && expected.ProfileName == actual.ProfileName && expected.DeviceName == actual.DeviceName && (expected.Origin == actual.Origin || (expected.Origin == 0 && actual.Origin > 0)) && expected.ResourceName == actual.ResourceName && expected.Value == actual.Value && expected.ValueType == actual.ValueType {
+		if expected.ValueType == v3.ValueTypeObject {
+			if !reflect.DeepEqual(expected.ObjectValue, actual.ObjectValue) {
+				return false
+			}
+		}
+		return true
+	}
+	return false
+}
+
+func TestConfigure(t *testing.T) {
+	tests := []struct {
+		conf     map[string]interface{}
+		expected *SinkConf
+		error    string
+	}{
+		{ // 0
+			conf: map[string]interface{}{
+				"metadata": "meta",
+			},
+			expected: &SinkConf{
+				MessageType: MessageTypeEvent,
+				ContentType: "application/json",
+				DeviceName:  "ekuiper",
+				ProfileName: "ekuiperProfile",
+				Metadata:    "meta",
+			},
+		},
+		{ // 1
+			conf: map[string]interface{}{
+				"type":        "redis",
+				"protocol":    "redis",
+				"host":        "edgex-redis",
+				"port":        6379,
+				"topic":       "ekuiperResult",
+				"deviceName":  "ekuiper",
+				"profileName": "ekuiper",
+				"sourceName":  "ekuiper",
+				"contentType": "application/json",
+			},
+			expected: &SinkConf{
+				MessageType: MessageTypeEvent,
+				ContentType: "application/json",
+				DeviceName:  "ekuiper",
+				ProfileName: "ekuiper",
+				SourceName:  "ekuiper",
+				Topic:       "ekuiperResult",
+			},
+		},
+		{ // 2
+			conf: map[string]interface{}{
+				"protocol":    "tcp",
+				"host":        "127.0.0.1",
+				"port":        1883,
+				"topic":       "result",
+				"type":        "mqtt",
+				"metadata":    "edgex_meta",
+				"contentType": "application/json",
+				"optional": map[string]interface{}{
+					"ClientId": "edgex_message_bus_001",
+				},
+			},
+			expected: &SinkConf{
+				MessageType: MessageTypeEvent,
+				ContentType: "application/json",
+				DeviceName:  "ekuiper",
+				ProfileName: "ekuiperProfile",
+				SourceName:  "",
+				Metadata:    "edgex_meta",
+				Topic:       "result",
+			},
+		},
+		{ // 3
+			conf: map[string]interface{}{
+				"type":        "redis",
+				"protocol":    "redis",
+				"host":        "edgex-redis",
+				"port":        6379,
+				"topicPrefix": "edgex/events/device",
+				"messageType": "request",
+				"contentType": "application/json",
+			},
+			expected: &SinkConf{
+				MessageType: MessageTypeRequest,
+				ContentType: "application/json",
+				DeviceName:  "ekuiper",
+				ProfileName: "ekuiperProfile",
+				SourceName:  "",
+				TopicPrefix: "edgex/events/device",
+			},
+		},
+		{ // 4
+			conf: map[string]interface{}{
+				"type":        "redis",
+				"protocol":    "redis",
+				"host":        "edgex-redis",
+				"port":        6379,
+				"topicPrefix": "edgex/events/device",
+				"messageType": "requests",
+				"contentType": "application/json",
+			},
+			error: "specified wrong messageType value requests",
+		},
+		{ // 5
+			conf: map[string]interface{}{
+				"protocol":    "redis",
+				"host":        "edgex-redis",
+				"port":        6379,
+				"topicPrefix": "edgex/events/device",
+				"topic":       "requests",
+				"contentType": "application/json",
+			},
+			error: "not allow to specify both topic and topicPrefix, please set one only",
+		},
+	}
+	for i, test := range tests {
+		t.Run(fmt.Sprintf("test %d", i), func(t *testing.T) {
+			ems := EdgexMsgBusSink{}
+			err := ems.Provision(ctx, test.conf)
+			if !reflect.DeepEqual(test.error, testx.Errstring(err)) {
+				t.Errorf("%d: error mismatch:\n  exp=%s\n  got=%s\n\n", i, test.error, err)
+			} else if test.error == "" {
+				assert.Equal(t, test.expected, ems.c)
+				assert.Equal(t, map[string]any{
+					"contentType": "application/json",
+				}, ems.sendParams)
+			}
+		})
+	}
+}
+
+func TestProduceEvents(t1 *testing.T) {
+	tests := []struct {
+		input    string
+		conf     map[string]interface{}
+		expected *dtos.Event
+		error    string
+	}{
+		{ // 0
+			input: `[
+						{"meta":{
+							"correlationid":"","deviceName":"demo","id":"","origin":3,
+							"humidity":{"deviceName":"test device name1","id":"12","origin":14,"valueType":"Int64"},
+							"temperature":{"deviceName":"test device name2","id":"22","origin":24}
+							}
+						},
+						{"humidity":100},
+						{"temperature":50}
+					]`,
+			conf: map[string]interface{}{
+				"metadata": "meta",
+			},
+			expected: &dtos.Event{
+				Id:          "",
+				DeviceName:  "demo",
+				ProfileName: "ekuiperProfile",
+				SourceName:  "ruleTest",
+				Origin:      3,
+				Readings: []dtos.BaseReading{
+					{
+						ResourceName:  "humidity",
+						DeviceName:    "test device name1",
+						ProfileName:   "ekuiperProfile",
+						Id:            "12",
+						Origin:        14,
+						ValueType:     v3.ValueTypeInt64,
+						SimpleReading: dtos.SimpleReading{Value: "100"},
+					},
+					{
+						ResourceName:  "temperature",
+						DeviceName:    "test device name2",
+						ProfileName:   "ekuiperProfile",
+						Id:            "22",
+						Origin:        24,
+						ValueType:     v3.ValueTypeFloat64,
+						SimpleReading: dtos.SimpleReading{Value: "5.000000e+01"},
+					},
+				},
+			},
+			error: "",
+		},
+
+		{ // 1
+			input: `[
+						{"meta":{
+							"correlationid":"","profileName":"demoProfile","deviceName":"demo","sourceName":"demoSource","id":"abc","origin":3,"tags":{"auth":"admin"},
+							"humidity":{"deviceName":"test device name1","id":"12","origin":14},
+							"temperature":{"deviceName":"test device name2","id":"22","origin":24}
+							}
+						},
+						{"h1":100},
+						{"h2":null}
+					]`,
+			conf: map[string]interface{}{
+				"metadata": "meta",
+			},
+			expected: &dtos.Event{
+				Id:          "abc",
+				DeviceName:  "demo",
+				ProfileName: "demoProfile",
+				SourceName:  "demoSource",
+				Origin:      3,
+				Tags:        map[string]interface{}{"auth": "admin"},
+				Readings: []dtos.BaseReading{
+					{
+						ResourceName:  "h1",
+						SimpleReading: dtos.SimpleReading{Value: "1.000000e+02"},
+						DeviceName:    "demo",
+						ProfileName:   "demoProfile",
+						ValueType:     v3.ValueTypeFloat64,
+					},
+				},
+			},
+			error: "",
+		},
+
+		{ // 2
+			input: `[
+						{"meta": 50,"h1":100}
+					]`,
+			conf: map[string]interface{}{
+				"sourceName": "demo",
+			},
+			expected: &dtos.Event{
+				DeviceName:  "ekuiper",
+				ProfileName: "ekuiperProfile",
+				SourceName:  "demo",
+				Readings: []dtos.BaseReading{
+					{
+						ResourceName:  "meta",
+						SimpleReading: dtos.SimpleReading{Value: "5.000000e+01"},
+						DeviceName:    "ekuiper",
+						ProfileName:   "ekuiperProfile",
+						ValueType:     v3.ValueTypeFloat64,
+					},
+					{
+						ResourceName:  "h1",
+						SimpleReading: dtos.SimpleReading{Value: "1.000000e+02"},
+						DeviceName:    "ekuiper",
+						ProfileName:   "ekuiperProfile",
+						ValueType:     v3.ValueTypeFloat64,
+					},
+				},
+			},
+			error: "",
+		},
+
+		{ // 3
+			input: `[
+						{"meta1": "newmeta"},
+						{"h1":true},
+						{"sa":["1","2","3","4"]},
+						{"fa":[1.1,2.2,3.3,4.4]}
+					]`,
+			expected: &dtos.Event{
+				DeviceName:  "ekuiper",
+				ProfileName: "ekuiperProfile",
+				SourceName:  "ruleTest",
+				Readings: []dtos.BaseReading{
+					{
+						ResourceName:  "meta1",
+						SimpleReading: dtos.SimpleReading{Value: "newmeta"},
+						DeviceName:    "ekuiper",
+						ProfileName:   "ekuiperProfile",
+						ValueType:     v3.ValueTypeString,
+					},
+					{
+						ResourceName:  "h1",
+						SimpleReading: dtos.SimpleReading{Value: "true"},
+						DeviceName:    "ekuiper",
+						ProfileName:   "ekuiperProfile",
+						ValueType:     v3.ValueTypeBool,
+					},
+					{
+						ResourceName:  "sa",
+						SimpleReading: dtos.SimpleReading{Value: "[1, 2, 3, 4]"},
+						DeviceName:    "ekuiper",
+						ProfileName:   "ekuiperProfile",
+						ValueType:     v3.ValueTypeStringArray,
+					},
+					{
+						ResourceName:  "fa",
+						SimpleReading: dtos.SimpleReading{Value: "[1.100000e+00, 2.200000e+00, 3.300000e+00, 4.400000e+00]"},
+						DeviceName:    "ekuiper",
+						ProfileName:   "ekuiperProfile",
+						ValueType:     v3.ValueTypeFloat64Array,
+					},
+				},
+			},
+			error: "",
+		},
+
+		{ // 4
+			input: `[]`,
+			conf: map[string]interface{}{
+				"deviceName":  "kuiper",
+				"profileName": "kp",
+				"topic":       "demo",
+			},
+			expected: &dtos.Event{
+				ProfileName: "kp",
+				DeviceName:  "kuiper",
+				SourceName:  "ruleTest",
+				Origin:      0,
+				Readings:    nil,
+			},
+			error: "",
+		},
+		{ // 5
+			input: `[{"sa":["1","2",3,"4"]}]`, // invalid array, return nil
+			expected: &dtos.Event{
+				DeviceName:  "ekuiper",
+				ProfileName: "ekuiperProfile",
+				SourceName:  "ruleTest",
+				Origin:      0,
+				Readings:    nil,
+			},
+		},
+		{ // 6
+			input: `[
+						{"meta1": "newmeta"},
+						{"sa":"SGVsbG8gV29ybGQ="},
+						{"meta":{
+							"correlationid":"","profileName":"demoProfile","deviceName":"demo","id":"abc","origin":3,"tags":{"auth":"admin"},
+							"sa":{"deviceName":"test device name1","id":"12","origin":14, "valueType":"Binary","mediaType":"application/css"}
+						}}
+					]`,
+			conf: map[string]interface{}{
+				"metadata":    "meta",
+				"profileName": "myprofile",
+				"sourceName":  "ds",
+			},
+			expected: &dtos.Event{
+				DeviceName:  "demo",
+				ProfileName: "demoProfile",
+				SourceName:  "ds",
+				Origin:      3,
+				Tags:        map[string]interface{}{"auth": "admin"},
+				Readings: []dtos.BaseReading{
+					{
+						DeviceName:    "demo",
+						ProfileName:   "demoProfile",
+						ResourceName:  "meta1",
+						SimpleReading: dtos.SimpleReading{Value: "newmeta"},
+						ValueType:     v3.ValueTypeString,
+					},
+					{
+						ResourceName:  "sa",
+						BinaryReading: dtos.BinaryReading{BinaryValue: []byte("Hello World"), MediaType: "application/css"},
+						ProfileName:   "demoProfile",
+						DeviceName:    "test device name1",
+						Id:            "12",
+						Origin:        14,
+						ValueType:     v3.ValueTypeBinary,
+					},
+				},
+			},
+			error: "",
+		},
+		{ // 7
+			input: `[
+						{"meta":{
+							"correlationid":"","deviceName":"demo","id":"","origin":3,
+							"obj":{"deviceName":"test device name1","id":"12","origin":14,"valueType":"Object"}
+							}
+						},
+						{"obj":{"a":1,"b":"sttt"}}
+					]`,
+			conf: map[string]interface{}{
+				"metadata": "meta",
+			},
+			expected: &dtos.Event{
+				Id:          "",
+				DeviceName:  "demo",
+				ProfileName: "ekuiperProfile",
+				SourceName:  "ruleTest",
+				Origin:      3,
+				Readings: []dtos.BaseReading{
+					{
+						ResourceName: "obj",
+						DeviceName:   "test device name1",
+						ProfileName:  "ekuiperProfile",
+						Id:           "12",
+						Origin:       14,
+						ValueType:    v3.ValueTypeObject,
+						ObjectReading: dtos.ObjectReading{ObjectValue: map[string]interface{}{
+							"a": float64(1),
+							"b": "sttt",
+						}},
+					},
+				},
+			},
+			error: "",
+		},
+		{ // 8
+			input: `[
+						{"obj":{"a":1,"b":"sttt"}}
+					]`,
+			conf: map[string]interface{}{},
+			expected: &dtos.Event{
+				Id:          "",
+				DeviceName:  "ekuiper",
+				ProfileName: "ekuiperProfile",
+				SourceName:  "ruleTest",
+				Origin:      0,
+				Readings: []dtos.BaseReading{
+					{
+						ResourceName: "obj",
+						DeviceName:   "ekuiper",
+						ProfileName:  "ekuiperProfile",
+						Id:           "",
+						Origin:       0,
+						ValueType:    v3.ValueTypeObject,
+						ObjectReading: dtos.ObjectReading{ObjectValue: map[string]interface{}{
+							"a": float64(1),
+							"b": "sttt",
+						}},
+					},
+				},
+			},
+			error: "",
+		},
+	}
+
+	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
+	for i, t := range tests {
+		ems := EdgexMsgBusSink{}
+		err := ems.Provision(ctx, t.conf)
+		if err != nil {
+			t1.Errorf("%d: configure error %v", i, err)
+			continue
+		}
+		if ems.c.SourceName == "" {
+			ems.c.SourceName = "ruleTest"
+		}
+		var payload []map[string]interface{}
+		json.Unmarshal([]byte(t.input), &payload)
+		result, err := ems.produceEvents(ctx, payload)
+		if !reflect.DeepEqual(t.error, testx.Errstring(err)) {
+			t1.Errorf("%d. %q: error mismatch:\n  exp=%s\n  got=%s\n\n", i, t.input, t.error, err)
+		} else if t.error == "" && !compareEvent(t.expected, result) {
+			t1.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, t.input, t.expected, result)
+		}
+	}
+}
diff --git a/internal/io/edgex/source_test.go b/internal/io/edgex/source_test.go
new file mode 100644
index 00000000..e1c5867b
--- /dev/null
+++ b/internal/io/edgex/source_test.go
@@ -0,0 +1,358 @@
+// 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.
+
+package edgex
+
+import (
+	"encoding/json"
+	"fmt"
+	"math"
+	"reflect"
+	"testing"
+
+	v3 "github.com/edgexfoundry/go-mod-core-contracts/v3/common"
+	"github.com/edgexfoundry/go-mod-core-contracts/v3/dtos"
+	"github.com/edgexfoundry/go-mod-core-contracts/v3/models"
+	"github.com/edgexfoundry/go-mod-messaging/v3/pkg/types"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+)
+
+var (
+	es      = &Source{}
+	typeMap = map[string]string{
+		"b1":  v3.ValueTypeBool,
+		"i1":  v3.ValueTypeInt8,
+		"i2":  v3.ValueTypeInt16,
+		"i3":  v3.ValueTypeInt32,
+		"i4":  v3.ValueTypeInt64,
+		"i5":  v3.ValueTypeUint8,
+		"i6":  v3.ValueTypeUint16,
+		"i7":  v3.ValueTypeUint32,
+		"s1":  v3.ValueTypeString,
+		"f1":  v3.ValueTypeFloat32,
+		"f2":  v3.ValueTypeFloat64,
+		"i8":  v3.ValueTypeUint64,
+		"ba":  v3.ValueTypeBoolArray,
+		"ia1": v3.ValueTypeInt8Array,
+		"ia2": v3.ValueTypeInt16Array,
+		"ia3": v3.ValueTypeInt32Array,
+		"ia4": v3.ValueTypeInt64Array,
+		"ia5": v3.ValueTypeUint8Array,
+		"ia6": v3.ValueTypeUint16Array,
+		"ia7": v3.ValueTypeUint32Array,
+		"ia8": v3.ValueTypeUint64Array,
+		"fa1": v3.ValueTypeFloat32Array,
+		"fa2": v3.ValueTypeFloat64Array,
+	}
+)
+
+func TestGetValue_IntFloat(t *testing.T) {
+	testEvent := models.Event{DeviceName: "test"}
+	for i := 1; i < 8; i++ {
+		name := fmt.Sprintf("i%d", i)
+		r1 := models.SimpleReading{
+			BaseReading: models.BaseReading{
+				ResourceName: name,
+				ValueType:    typeMap[name],
+			},
+			Value: "1",
+		}
+		testEvent.Readings = append(testEvent.Readings, r1)
+	}
+
+	dtoe := dtos.FromEventModelToDTO(testEvent)
+	for _, r := range dtoe.Readings {
+		if v, e := es.getValue(r, conf.Log); e != nil {
+			t.Errorf("%s", e)
+		} else {
+			expectOne(t, v)
+		}
+	}
+
+	r1 := dtos.BaseReading{ResourceName: "i8", ValueType: typeMap["i8"], SimpleReading: dtos.SimpleReading{Value: "10796529505058023104"}}
+	if v, e := es.getValue(r1, conf.Log); e != nil {
+		t.Errorf("%s", e)
+	} else {
+		if v1, ok := v.(uint64); ok {
+			if v1 != 10796529505058023104 {
+				t.Errorf("expected 10796529505058023104, but it's %d.", v1)
+			}
+		}
+	}
+
+	r2 := dtos.BaseReading{ResourceName: "f1", ValueType: typeMap["f1"], SimpleReading: dtos.SimpleReading{Value: "3.14"}}
+	if v, e := es.getValue(r2, conf.Log); e != nil {
+		t.Errorf("%s", e)
+	} else {
+		if v1, ok := v.(float64); ok {
+			if !almostEqual(v1, 3.14) {
+				t.Errorf("expected 3.14, but it's %f.", v1)
+			}
+		}
+	}
+}
+
+func almostEqual(a, b float64) bool {
+	return math.Abs(a-b) <= 1e-6
+}
+
+func TestGetValue_IntFloatArr(t *testing.T) {
+	testEvent := models.Event{DeviceName: "test"}
+	for i := 1; i < 8; i++ {
+		ia := []int{i, i * 2}
+		jsonValue, _ := json.Marshal(ia)
+		name := fmt.Sprintf("ia%d", i)
+		r1 := models.SimpleReading{
+			BaseReading: models.BaseReading{
+				ResourceName: name,
+				ValueType:    typeMap[name],
+			},
+			Value: string(jsonValue),
+		}
+		testEvent.Readings = append(testEvent.Readings, r1)
+	}
+
+	dtoe := dtos.FromEventModelToDTO(testEvent)
+	for i, r := range dtoe.Readings {
+		if v, e := es.getValue(r, conf.Log); e != nil {
+			t.Errorf("%s", e)
+		} else {
+			checkArray(t, i, v)
+		}
+	}
+
+	r1 := dtos.BaseReading{ResourceName: "ia8", ValueType: typeMap["ia8"], SimpleReading: dtos.SimpleReading{Value: `[10796529505058023104, 10796529505058023105]`}}
+	if v, e := es.getValue(r1, conf.Log); e != nil {
+		t.Errorf("%s", e)
+	} else {
+		if v1, ok := v.([]uint64); ok {
+			if v1[0] != 10796529505058023104 || v1[1] != 10796529505058023105 {
+				t.Errorf("Failed, the array value is not correct %v.", v1)
+			}
+		} else {
+			t.Errorf("expected uint64 array type, but it's %T.", v1)
+		}
+	}
+
+	rf_00 := dtos.BaseReading{ResourceName: "fa1", ValueType: typeMap["fa1"], SimpleReading: dtos.SimpleReading{Value: `[3.14, 2.71828]`}}
+	if v, e := es.getValue(rf_00, conf.Log); e != nil {
+		t.Errorf("%s", e)
+	} else {
+		if v1, ok := v.([]float64); ok {
+			if v1[0] != 3.14 || v1[1] != 2.71828 {
+				t.Errorf("expected 3.14 & 2.71828, but it's %v.", v1)
+			}
+		} else {
+			t.Errorf("expected float32 array type, but it's %T.", v)
+		}
+	}
+}
+
+func checkArray(t *testing.T, index int, val interface{}) {
+	if v1, ok := val.([]int); ok {
+		newIdx := index + 1
+		if v1[0] != newIdx || v1[1] != newIdx*2 {
+			t.Errorf("Failed, the array value is not correct %v.", v1)
+		}
+	} else {
+		t.Errorf("expected int array type, but it's %T.", val)
+	}
+}
+
+func expectOne(t *testing.T, expected interface{}) {
+	if v1, ok := expected.(int); ok {
+		if v1 != 1 {
+			t.Errorf("expected 1, but it's %d.", v1)
+		}
+	} else {
+		t.Errorf("expected int type, but it's %T.", expected)
+	}
+}
+
+func TestGetValue_Float(t *testing.T) {
+	testEvent := models.Event{DeviceName: "test"}
+	for i := 1; i < 3; i++ {
+		name := fmt.Sprintf("f%d", i)
+		r1 := models.SimpleReading{
+			BaseReading: models.BaseReading{
+				ResourceName: name,
+				ValueType:    typeMap[name],
+			},
+			Value: "3.14",
+		}
+		testEvent.Readings = append(testEvent.Readings, r1)
+	}
+
+	dtoe := dtos.FromEventModelToDTO(testEvent)
+	for _, r := range dtoe.Readings {
+		if v, e := es.getValue(r, conf.Log); e != nil {
+			t.Errorf("%s", e)
+		} else {
+			expectPi(t, v)
+		}
+	}
+}
+
+func expectPi(t *testing.T, expected interface{}) {
+	if v1, ok := expected.(float64); ok {
+		if !almostEqual(v1, 3.14) {
+			t.Errorf("expected 3.14, but it's %f.", v1)
+		}
+	} else {
+		t.Errorf("expected float type, but it's %T.", expected)
+	}
+}
+
+func TestGetValue_Bool(t *testing.T) {
+	///////////True
+	trues := []string{"1", "t", "T", "true", "TRUE", "True"}
+	for _, v := range trues {
+		r1 := dtos.BaseReading{ResourceName: "b1", ValueType: typeMap["b1"], SimpleReading: dtos.SimpleReading{Value: v}}
+		if v, e := es.getValue(r1, conf.Log); e != nil {
+			t.Errorf("%s", e)
+		} else {
+			expectTrue(t, v)
+		}
+	}
+
+	r1 := dtos.BaseReading{ResourceName: "b1", ValueType: typeMap["b1"], SimpleReading: dtos.SimpleReading{Value: "TRue"}}
+	if _, e := es.getValue(r1, conf.Log); e == nil {
+		t.Errorf("%s", e)
+	}
+
+	///////////False
+	falses := []string{"0", "f", "F", "false", "FALSE", "False"}
+	for _, v := range falses {
+		r1 := dtos.BaseReading{ResourceName: "b1", ValueType: typeMap["b1"], SimpleReading: dtos.SimpleReading{Value: v}}
+		if v, e := es.getValue(r1, conf.Log); e != nil {
+			t.Errorf("%s", e)
+		} else {
+			expectFalse(t, v)
+		}
+	}
+
+	r1 = dtos.BaseReading{ResourceName: "b1", ValueType: typeMap["b1"], SimpleReading: dtos.SimpleReading{Value: "FAlse"}}
+	if _, e := es.getValue(r1, conf.Log); e == nil {
+		t.Errorf("%s", e)
+	}
+}
+
+func expectTrue(t *testing.T, expected interface{}) {
+	if v1, ok := expected.(bool); ok {
+		if !v1 {
+			t.Errorf("expected true, but it's false.")
+		}
+	} else {
+		t.Errorf("expected boolean type, but it's %t.", expected)
+	}
+}
+
+func expectFalse(t *testing.T, expected interface{}) {
+	if v1, ok := expected.(bool); ok {
+		if v1 {
+			t.Errorf("expected false, but it's true.")
+		}
+	} else {
+		t.Errorf("expected boolean type, but it's %t.", expected)
+	}
+}
+
+func TestWrongType(t *testing.T) {
+	r1 := dtos.BaseReading{ResourceName: "f", ValueType: "FLOAT", SimpleReading: dtos.SimpleReading{Value: "100"}}
+	if v, _ := es.getValue(r1, conf.Log); v != "100" {
+		t.Errorf("Expected 100, but it's %s!", v)
+	}
+}
+
+func TestWrongValue(t *testing.T) {
+	testEvent := models.Event{DeviceName: "test"}
+	// 100 cannot be converted to a boolean value
+	r1 := models.SimpleReading{
+		BaseReading: models.BaseReading{
+			ResourceName: "b1",
+			ValueType:    typeMap["b1"],
+		},
+		Value: "100",
+	}
+	//'int' string cannot be converted to int value
+	r2 := models.SimpleReading{
+		BaseReading: models.BaseReading{
+			ResourceName: "i1",
+			ValueType:    typeMap["i1"],
+		},
+		Value: "int",
+	}
+	//'float' string cannot be converted to int value
+	r3 := models.SimpleReading{
+		BaseReading: models.BaseReading{
+			ResourceName: "f1",
+			ValueType:    typeMap["f1"],
+		},
+		Value: "float",
+	}
+	testEvent.Readings = append(testEvent.Readings, r1, r2, r3)
+
+	dtoe := dtos.FromEventModelToDTO(testEvent)
+	for _, v := range dtoe.Readings {
+		if _, e := es.getValue(v, conf.Log); e == nil {
+			t.Errorf("Expected an error!")
+		}
+	}
+}
+
+func TestPrintConf(t *testing.T) {
+	expMbconf := types.MessageBusConfig{Broker: types.HostInfo{Protocol: "tcp", Host: "127.0.0.1", Port: 6625}, Type: "mbus", Optional: map[string]string{
+		"proa":     "proa",
+		"Password": "fafsadfsadf=",
+		"Prob":     "Prob",
+	}}
+	mbconf := types.MessageBusConfig{Broker: types.HostInfo{Protocol: "tcp", Host: "127.0.0.1", Port: 6625}, Type: "mbus", Optional: map[string]string{
+		"proa":     "proa",
+		"Password": "fafsadfsadf=",
+		"Prob":     "Prob",
+	}}
+	if !reflect.DeepEqual(expMbconf, mbconf) {
+		t.Errorf("conf changed after printing")
+	}
+}
+
+func TestGetValue_Binary(t *testing.T) {
+	ev := []byte("Hello World")
+	r1 := dtos.BaseReading{ResourceName: "bin", ValueType: v3.ValueTypeBinary, BinaryReading: dtos.BinaryReading{MediaType: "application/text", BinaryValue: ev}}
+	if v, e := es.getValue(r1, conf.Log); e != nil {
+		t.Errorf("%s", e)
+	} else if !reflect.DeepEqual(ev, v) {
+		t.Errorf("result mismatch, expect %v, but got %v", ev, v)
+	}
+}
+
+func TestGetValue_Object(t *testing.T) {
+	ev := map[string]interface{}{
+		"a": 3,
+		"b": "string",
+		"c": 12.3,
+	}
+	r1 := dtos.NewObjectReading("p", "d", "obj", ev)
+	err := r1.Validate()
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	if v, e := es.getValue(r1, conf.Log); e != nil {
+		t.Errorf("%s", e)
+	} else if !reflect.DeepEqual(ev, v) {
+		t.Errorf("result mismatch, expect %v, but got %v", ev, v)
+	}
+}
diff --git a/internal/io/file/sink_test.go b/internal/io/file/sink_test.go
index 02e8224e..97157e29 100644
--- a/internal/io/file/sink_test.go
+++ b/internal/io/file/sink_test.go
@@ -1,4 +1,4 @@
-// Copyright 2023 EMQ Technologies Co., Ltd.
+// Copyright 2023-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.
@@ -353,6 +353,69 @@ func TestFileSink_Collect(t *testing.T) {
 	}
 }
 
+// Test single message with header defined
+func TestCSVSingMessHeader(t *testing.T) {
+	tests := []struct {
+		name    string
+		fname   string
+		content []byte
+	}{
+		{
+			name:    "csv",
+			fname:   "test_csvh",
+			content: []byte("id:name\n12:test\n11:value2"),
+		},
+	}
+
+	ctx := mockContext.NewMockContext("test1", "test")
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create a temporary file for testing
+			tmpfile, err := os.CreateTemp("", tt.fname)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer os.Remove(tmpfile.Name())
+			// Create a file sink with the temporary file path
+			sink := &fileSink{}
+			f := message.FormatDelimited
+			err = sink.Provision(ctx, map[string]interface{}{
+				"path":               tmpfile.Name(),
+				"fileType":           CSV_TYPE,
+				"hasHeader":          true,
+				"delimiter":          ":",
+				"format":             f,
+				"rollingNamePattern": "none",
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+			err = sink.Connect(ctx)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			if err := sink.Collect(ctx, &xsql.RawTuple{Rawdata: []byte{0x3a, 0x0, 0x0, 0x0, 0x7, 0x69, 0x64, 0x3a, 0x6e, 0x61, 0x6d, 0x65, 0x31, 0x32, 0x3a, 0x74, 0x65, 0x73, 0x74}}); err != nil {
+				t.Errorf("unexpected error: %s", err)
+			}
+
+			// Test collecting another map item
+			if err := sink.Collect(ctx, &xsql.RawTuple{Rawdata: []byte("11:value2")}); err != nil {
+				t.Errorf("unexpected error: %s", err)
+			}
+			if err = sink.Close(ctx); err != nil {
+				t.Errorf("unexpected close error: %s", err)
+			}
+			// Read the contents of the temporary file and check if they match the collected items
+			contents, err := os.ReadFile(tmpfile.Name())
+			if err != nil {
+				t.Fatal(err)
+			}
+			assert.Equal(t, tt.content, contents)
+		})
+	}
+}
+
 // Test file rolling by time
 func TestFileSinkRolling_Collect(t *testing.T) {
 	// Remove existing files
diff --git a/internal/io/neuron/source_test.go b/internal/io/neuron/source_test.go
index ce20e47e..8ec2a2e3 100644
--- a/internal/io/neuron/source_test.go
+++ b/internal/io/neuron/source_test.go
@@ -67,7 +67,7 @@ func TestProvision(t *testing.T) {
 	sid := su.SubId(map[string]any{
 		"url": "tcp://127.0.0.1:8000",
 	})
-	assert.Equal(t, "nng:pairtcp://127.0.0.1:8000", sid)
+	assert.Equal(t, "singleton", sid)
 
 	err = s.Connect(ctx)
 	assert.NoError(t, err)
diff --git a/internal/server/rest_test.go b/internal/server/rest_test.go
index 8867636c..8ba54e73 100644
--- a/internal/server/rest_test.go
+++ b/internal/server/rest_test.go
@@ -209,6 +209,45 @@ func (suite *RestTestSuite) Test_sourcesManageHandler() {
 	assert.Equal(suite.T(), http.StatusOK, w.Code)
 }
 
+func (suite *RestTestSuite) TestRecoverRule() {
+	// drop stream
+	req, _ := http.NewRequest(http.MethodDelete, "http://localhost:8080/streams/recoverTest", bytes.NewBufferString("any"))
+	w := httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	req1, _ := http.NewRequest(http.MethodDelete, "http://localhost:8080/rules/recoverTest", bytes.NewBufferString("any"))
+	w1 := httptest.NewRecorder()
+	suite.r.ServeHTTP(w1, req1)
+
+	buf1 := bytes.NewBuffer([]byte(`{"sql":"CREATE stream recoverTest() WITH (DATASOURCE=\"0\", TYPE=\"mqtt\")"}`))
+	req1, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/streams", buf1)
+	w1 = httptest.NewRecorder()
+	suite.r.ServeHTTP(w1, req1)
+	// create rule with trigger false
+	ruleJson := `{"id": "recoverTest","triggered": false,"sql": "select * from recoverTest","actions": [{"log": {}}]}`
+
+	buf2 := bytes.NewBuffer([]byte(ruleJson))
+	req2, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/rules", buf2)
+	w2 := httptest.NewRecorder()
+	suite.r.ServeHTTP(w2, req2)
+
+	req3, _ := http.NewRequest(http.MethodGet, "http://localhost:8080/rules", bytes.NewBufferString("any"))
+	w3 := httptest.NewRecorder()
+	suite.r.ServeHTTP(w3, req3)
+
+	b, err := io.ReadAll(w3.Result().Body)
+	require.NoError(suite.T(), err)
+	got := make([]map[string]string, 0)
+	require.NoError(suite.T(), json.Unmarshal(b, &got))
+	find := false
+	for _, s := range got {
+		if s["id"] == "recoverTest" {
+			find = true
+			require.Equal(suite.T(), "Stopped: canceled manually.", s["status"])
+		}
+	}
+	require.True(suite.T(), find)
+}
+
 func (suite *RestTestSuite) Test_rulesManageHandler() {
 	connection.InitConnectionManager4Test()
 	// Start rules
diff --git a/internal/topo/operator/emitter_operator_test.go b/internal/topo/operator/emitter_operator_test.go
new file mode 100644
index 00000000..4ae56a6d
--- /dev/null
+++ b/internal/topo/operator/emitter_operator_test.go
@@ -0,0 +1,74 @@
+// Copyright 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 operator
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/v2/internal/xsql"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+)
+
+func TestEmitterOpApply(t *testing.T) {
+	tests := []struct {
+		name   string
+		data   any
+		result any
+	}{
+		{
+			name: "tuple",
+			data: &xsql.Tuple{
+				Emitter: "tbl",
+				Message: xsql.Message{
+					"a": int64(6),
+				},
+			},
+			result: &xsql.Tuple{
+				Emitter: "stream",
+				Message: xsql.Message{
+					"a": int64(6),
+				},
+			},
+		},
+		{
+			name: "tuple",
+			data: &xsql.RawTuple{
+				Emitter: "tbl",
+			},
+			result: &xsql.RawTuple{
+				Emitter: "stream",
+			},
+		},
+		{
+			name: "invalid",
+			data: &xsql.Message{
+				"Emitter": "tbl",
+			},
+			result: errors.New("run emitter op error: invalid input *xsql.Message(&map[Emitter:tbl])"),
+		},
+	}
+
+	ctx := mockContext.NewMockContext("testOp", "emitter")
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			pp := &EmitterOp{Emitter: "stream"}
+			result := pp.Apply(ctx, tt.data, nil, nil)
+			assert.Equal(t, tt.result, result)
+		})
+	}
+}
diff --git a/internal/topo/planner/planner_source_test.go b/internal/topo/planner/planner_source_test.go
index 030be6fe..b5b6efc1 100644
--- a/internal/topo/planner/planner_source_test.go
+++ b/internal/topo/planner/planner_source_test.go
@@ -211,15 +211,18 @@ func TestPlanTopo(t *testing.T) {
 				Sources: []string{"source_mqtt.localConnection/topic1"},
 				Edges: map[string][]any{
 					"source_mqtt.localConnection/topic1": {
-						"op_2_ratelimit",
+						"op_2_emitter",
 					},
-					"op_2_ratelimit": {
-						"op_3_payload_decoder",
+					"op_2_emitter": {
+						"op_3_ratelimit",
 					},
-					"op_3_payload_decoder": {
-						"op_4_project",
+					"op_3_ratelimit": {
+						"op_4_payload_decoder",
 					},
-					"op_4_project": {
+					"op_4_payload_decoder": {
+						"op_5_project",
+					},
+					"op_5_project": {
 						"op_logToMemory_0_0_transform",
 					},
 					"op_logToMemory_0_0_transform": {
@@ -238,15 +241,18 @@ func TestPlanTopo(t *testing.T) {
 				Sources: []string{"source_mqtt.localConnection/topic1"},
 				Edges: map[string][]any{
 					"source_mqtt.localConnection/topic1": {
-						"op_src4_2_ratelimit",
+						"op_src4_2_emitter",
 					},
-					"op_src4_2_ratelimit": {
-						"op_src4_3_decoder",
+					"op_src4_2_emitter": {
+						"op_src4_3_ratelimit",
 					},
-					"op_src4_3_decoder": {
-						"op_4_project",
+					"op_src4_3_ratelimit": {
+						"op_src4_4_decoder",
 					},
-					"op_4_project": {
+					"op_src4_4_decoder": {
+						"op_5_project",
+					},
+					"op_5_project": {
 						"op_logToMemory_0_0_transform",
 					},
 					"op_logToMemory_0_0_transform": {
@@ -340,15 +346,18 @@ func TestPlanTopo(t *testing.T) {
 				Sources: []string{"source_mqtt.localConnection/topic1"},
 				Edges: map[string][]any{
 					"source_mqtt.localConnection/topic1": {
-						"op_2_ratelimit",
+						"op_2_emitter",
 					},
-					"op_2_ratelimit": {
-						"op_3_payload_decoder",
+					"op_2_emitter": {
+						"op_3_ratelimit",
 					},
-					"op_3_payload_decoder": {
-						"op_4_project",
+					"op_3_ratelimit": {
+						"op_4_payload_decoder",
 					},
-					"op_4_project": {
+					"op_4_payload_decoder": {
+						"op_5_project",
+					},
+					"op_5_project": {
 						"op_logToMemory_0_0_transform",
 					},
 					"op_logToMemory_0_0_transform": {
@@ -364,15 +373,18 @@ func TestPlanTopo(t *testing.T) {
 			name: "testNngConnSplit",
 			sql:  `SELECT * FROM neuron1`,
 			topo: &def.PrintableTopo{
-				Sources: []string{"source_nng:pairtcp://127.0.0.1:7777"},
+				Sources: []string{"source_nng:pairtcp://127.0.0.1:7777/singleton"},
 				Edges: map[string][]any{
-					"source_nng:pairtcp://127.0.0.1:7777": {
-						"op_2_decoder",
+					"source_nng:pairtcp://127.0.0.1:7777/singleton": {
+						"op_2_emitter",
 					},
-					"op_2_decoder": {
-						"op_3_project",
+					"op_2_emitter": {
+						"op_3_decoder",
 					},
-					"op_3_project": {
+					"op_3_decoder": {
+						"op_4_project",
+					},
+					"op_4_project": {
 						"op_logToMemory_0_0_transform",
 					},
 					"op_logToMemory_0_0_transform": {
diff --git a/test/edgex/pub.go b/test/edgex/pub.go
index 19865243..8ef64b7c 100644
--- a/test/edgex/pub.go
+++ b/test/edgex/pub.go
@@ -1,4 +1,4 @@
-// Copyright 2021-2023 EMQ Technologies Co., Ltd.
+// 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.
@@ -12,9 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//go:build edgex
-// +build edgex
-
 package main
 
 import (
diff --git a/test/edgex/sub/sub.go b/test/edgex/sub/sub.go
index 6bcc143b..a458db00 100644
--- a/test/edgex/sub/sub.go
+++ b/test/edgex/sub/sub.go
@@ -1,4 +1,4 @@
-// Copyright 2021-2023 EMQ Technologies Co., Ltd.
+// 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.
@@ -12,9 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//go:build edgex
-// +build edgex
-
 package main
 
 import (
diff --git a/test/edgex_redis_share_connection_sink_rule.jmx b/test/edgex_redis_share_connection_sink_rule.jmx
index 487d0490..70ca1508 100644
--- a/test/edgex_redis_share_connection_sink_rule.jmx
+++ b/test/edgex_redis_share_connection_sink_rule.jmx
@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <!--
-  ~ Copyright 2023 EMQ Technologies Co., Ltd.
+  ~ Copyright 2023-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.
@@ -107,6 +107,52 @@
           <boolProp name="TransactionController.parent">false</boolProp>
         </TransactionController>
         <hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_CreateConn" enabled="true">
+            <boolProp name="HTTPSampler.postBodyRaw">true</boolProp>
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments">
+              <collectionProp name="Arguments.arguments">
+                <elementProp name="" elementType="HTTPArgument">
+                  <boolProp name="HTTPArgument.always_encode">false</boolProp>
+                  <stringProp name="Argument.value">{&#xd;
+  &quot;id&quot;: &quot;redisMsgBus&quot;,&#xd;
+  &quot;typ&quot;: &quot;edgex&quot;,&#xd;
+  &quot;props&quot;: {&#xd;
+    &quot;protocol&quot;: &quot;redis&quot;,&#xd;
+    &quot;server&quot;: &quot;127.0.0.1&quot;,&#xd;
+    &quot;port&quot;: 6379,&#xd;
+    &quot;type&quot;: &quot;redis&quot;&#xd;
+  }&#xd;
+}</stringProp>
+                  <stringProp name="Argument.metadata">=</stringProp>
+                </elementProp>
+              </collectionProp>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/connections</stringProp>
+            <stringProp name="HTTPSampler.method">POST</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="-1867169789">success</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">true</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
           <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_CreateStream" enabled="true">
             <boolProp name="HTTPSampler.postBodyRaw">true</boolProp>
             <elementProp name="HTTPsampler.Arguments" elementType="Arguments">
@@ -158,7 +204,7 @@
   &quot;actions&quot;: [&#xd;
     {&#xd;
       &quot;edgex&quot;: {&#xd;
-        &quot;connectionSelector&quot;: &quot;edgex.redisMsgBus&quot;,&#xd;
+        &quot;connectionSelector&quot;: &quot;redisMsgBus&quot;,&#xd;
         &quot;topic&quot;: &quot;result&quot;,&#xd;
         &quot;metadata&quot;: &quot;edgex_meta&quot;,&#xd;
         &quot;contentType&quot;: &quot;application/json&quot;,&#xd;
@@ -220,7 +266,7 @@
           </HTTPSamplerProxy>
           <hashTree>
             <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="true">
-              <stringProp name="JSON_PATH">$.source_demo_0_records_in_total</stringProp>
+              <stringProp name="JSON_PATH">$.source_redisMsgBus/events_0_records_in_total</stringProp>
               <stringProp name="EXPECTED_VALUE">0</stringProp>
               <boolProp name="JSONVALIDATION">true</boolProp>
               <boolProp name="EXPECT_NULL">false</boolProp>
@@ -262,7 +308,7 @@
           </HTTPSamplerProxy>
           <hashTree>
             <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="true">
-              <stringProp name="JSON_PATH">$.source_demo_0_records_in_total</stringProp>
+              <stringProp name="JSON_PATH">$.source_redisMsgBus/events_0_records_in_total</stringProp>
               <stringProp name="EXPECTED_VALUE">10</stringProp>
               <boolProp name="JSONVALIDATION">true</boolProp>
               <boolProp name="EXPECT_NULL">false</boolProp>
@@ -272,7 +318,7 @@
             <hashTree/>
             <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="false">
               <stringProp name="JSON_PATH">$.sink_edgex_0_0_records_in_total</stringProp>
-              <stringProp name="EXPECTED_VALUE">6</stringProp>
+              <stringProp name="EXPECTED_VALUE">1</stringProp>
               <boolProp name="JSONVALIDATION">true</boolProp>
               <boolProp name="EXPECT_NULL">false</boolProp>
               <boolProp name="INVERT">false</boolProp>
diff --git a/test/edgexshared.http b/test/edgexshared.http
new file mode 100644
index 00000000..95d43b59
--- /dev/null
+++ b/test/edgexshared.http
@@ -0,0 +1,14 @@
+###
+POST http://{{host}}/connections
+Content-Type: application/json
+
+{
+  "id": "redisMsgBus",
+  "typ": "edgex",
+  "props": {
+    "protocol": "redis",
+    "server": "127.0.0.1",
+    "port": 6379,
+    "type": "redis"
+  }
+}
\ No newline at end of file
diff --git a/test/run_jmeter.sh b/test/run_jmeter.sh
index 78f65c82..e2d6a444 100755
--- a/test/run_jmeter.sh
+++ b/test/run_jmeter.sh
@@ -86,26 +86,26 @@ echo -e "---------------------------------------------\n"
 # /opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/rule_pipeline.jmx -l jmeter_logs/rule_pipeline.jtl -j jmeter_logs/rule_pipeline.log
 # echo -e "---------------------------------------------\n"
 
-#if test $with_edgex = true; then
-#  /opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/select_edgex_condition_rule.jmx -Dbase="$base_dir" -Dfvt="$fvt_dir" -l jmeter_logs/select_edgex_condition_rule.jtl -j jmeter_logs/select_edgex_condition_rule.log
-#  echo -e "---------------------------------------------\n"
-#
-#  /opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/edgex_sink_rule.jmx -Dfvt="$fvt_dir" -l jmeter_logs/edgex_sink_rule.jtl -j jmeter_logs/edgex_sink_rule.log
-#  echo -e "---------------------------------------------\n"
-#
-#  /opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/select_edgex_meta_rule.jmx -Dfvt="$fvt_dir" -l jmeter_logs/select_edgex_meta_rule.jtl -j jmeter_logs/select_edgex_meta_rule.log
-#  echo -e "---------------------------------------------\n"
-#
-#  /opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/edgex_mqtt_sink_rule.jmx -Dfvt="$fvt_dir" -l jmeter_logs/edgex_mqtt_sink_rule.jtl -j jmeter_logs/edgex_mqtt_sink_rule.log
-#  echo -e "---------------------------------------------\n"
-#
-#   /opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/edgex_redis_share_connection_sink_rule.jmx -Dfvt="$fvt_dir" -l jmeter_logs/edgex_redis_share_connection_sink_rule.jtl -j jmeter_logs/edgex_redis_share_connection_sink_rule.log
-#  echo -e "---------------------------------------------\n"
-#
-#
-#  /opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/edgex_array_rule.jmx  -Dfvt="$fvt_dir" -l jmeter_logs/edgex_array_rule.jtl -j jmeter_logs/edgex_array_rule.log
-#  echo -e "---------------------------------------------\n"
-#fi
+if test $with_edgex = true; then
+  /opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/select_edgex_condition_rule.jmx -Dbase="$base_dir" -Dfvt="$fvt_dir" -l jmeter_logs/select_edgex_condition_rule.jtl -j jmeter_logs/select_edgex_condition_rule.log
+  echo -e "---------------------------------------------\n"
+
+  /opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/edgex_sink_rule.jmx -Dfvt="$fvt_dir" -l jmeter_logs/edgex_sink_rule.jtl -j jmeter_logs/edgex_sink_rule.log
+  echo -e "---------------------------------------------\n"
+
+  /opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/select_edgex_meta_rule.jmx -Dfvt="$fvt_dir" -l jmeter_logs/select_edgex_meta_rule.jtl -j jmeter_logs/select_edgex_meta_rule.log
+  echo -e "---------------------------------------------\n"
+
+  /opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/edgex_mqtt_sink_rule.jmx -Dfvt="$fvt_dir" -l jmeter_logs/edgex_mqtt_sink_rule.jtl -j jmeter_logs/edgex_mqtt_sink_rule.log
+  echo -e "---------------------------------------------\n"
+
+   /opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/edgex_redis_share_connection_sink_rule.jmx -Dfvt="$fvt_dir" -l jmeter_logs/edgex_redis_share_connection_sink_rule.jtl -j jmeter_logs/edgex_redis_share_connection_sink_rule.log
+  echo -e "---------------------------------------------\n"
+
+
+  /opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/edgex_array_rule.jmx  -Dfvt="$fvt_dir" -l jmeter_logs/edgex_array_rule.jtl -j jmeter_logs/edgex_array_rule.log
+  echo -e "---------------------------------------------\n"
+fi
 #
 #/opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/plugin_end_2_end.jmx -Dfvt="$fvt_dir" -l jmeter_logs/plugin_end_2_end.jtl -j jmeter_logs/plugin_end_2_end.log
 #echo -e "---------------------------------------------\n"