Repository: lf-edge/ekuiper
Pull request #2998 (2024-07-09 09:41:57)

feat(io): restore edgex source/sink

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2998?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 `38.51262%` with `463 lines` in your changes missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@f2fdc02`). [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/2998?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/edgex/sink.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2998?src=pr&el=tree&filepath=internal%2Fio%2Fedgex%2Fsink.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vZWRnZXgvc2luay5nbw==) | 42.08% | [256 Missing and 11 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2998?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/edgex/source.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2998?src=pr&el=tree&filepath=internal%2Fio%2Fedgex%2Fsource.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vZWRnZXgvc291cmNlLmdv) | 23.62% | [144 Missing and 8 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2998?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/edgex/client/edgex.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2998?src=pr&el=tree&filepath=internal%2Fio%2Fedgex%2Fclient%2Fedgex.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vZWRnZXgvY2xpZW50L2VkZ2V4Lmdv) | 50.56% | [42 Missing and 2 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2998?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    #2998   +/-   ##
============================================
  Coverage             ?   67.60%           
============================================
  Files                ?      348           
  Lines                ?    33832           
  Branches             ?        0           
============================================
  Hits                 ?    22871           
  Misses               ?     8921           
  Partials             ?     2040           
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2998?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/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/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 0184bba2..2d6ea646 100644
--- a/internal/pkg/def/rule.go
+++ b/internal/pkg/def/rule.go
@@ -21,27 +21,27 @@ import (
 )
 
 type RuleOption struct {
-	Debug              bool              `json:"debug" yaml:"debug"`
-	LogFilename        string            `json:"logFilename" yaml:"logFilename"`
-	IsEventTime        bool              `json:"isEventTime" yaml:"isEventTime"`
-	LateTol            cast.DurationConf `json:"lateTolerance" yaml:"lateTolerance"`
+	Debug              bool              `json:"debug,omitempty" yaml:"debug,omitempty"`
+	LogFilename        string            `json:"logFilename,omitempty" yaml:"logFilename,omitempty"`
+	IsEventTime        bool              `json:"isEventTime,omitempty" yaml:"isEventTime,omitempty"`
+	LateTol            cast.DurationConf `json:"lateTolerance,omitempty" yaml:"lateTolerance,omitempty"`
 	Concurrency        int               `json:"concurrency" yaml:"concurrency"`
 	BufferLength       int               `json:"bufferLength" yaml:"bufferLength"`
-	SendMetaToSink     bool              `json:"sendMetaToSink" yaml:"sendMetaToSink"`
-	SendError          bool              `json:"sendError" yaml:"sendError"`
-	Qos                Qos               `json:"qos" yaml:"qos"`
-	CheckpointInterval cast.DurationConf `json:"checkpointInterval" yaml:"checkpointInterval"`
-	Restart            *RestartStrategy  `json:"restartStrategy" yaml:"restartStrategy"`
-	Cron               string            `json:"cron" yaml:"cron"`
-	Duration           string            `json:"duration" yaml:"duration"`
-	CronDatetimeRange  []DatetimeRange   `json:"cronDatetimeRange" yaml:"cronDatetimeRange"`
+	SendMetaToSink     bool              `json:"sendMetaToSink,omitempty" yaml:"sendMetaToSink,omitempty"`
+	SendError          bool              `json:"sendError,omitempty" yaml:"sendError,omitempty"`
+	Qos                Qos               `json:"qos,omitempty" yaml:"qos,omitempty"`
+	CheckpointInterval cast.DurationConf `json:"checkpointInterval,omitempty" yaml:"checkpointInterval,omitempty"`
+	Restart            *RestartStrategy  `json:"restartStrategy,omitempty" yaml:"restartStrategy,omitempty"`
+	Cron               string            `json:"cron,omitempty" yaml:"cron,omitempty"`
+	Duration           string            `json:"duration,omitempty" yaml:"duration,omitempty"`
+	CronDatetimeRange  []DatetimeRange   `json:"cronDatetimeRange,omitempty" yaml:"cronDatetimeRange,omitempty"`
 }
 
 type DatetimeRange struct {
 	Begin          string `json:"begin" yaml:"begin"`
 	End            string `json:"end" yaml:"end"`
-	BeginTimestamp int64  `json:"beginTimestamp"`
-	EndTimestamp   int64  `json:"endTimestamp"`
+	BeginTimestamp int64  `json:"beginTimestamp" yaml:"beginTimestamp"`
+	EndTimestamp   int64  `json:"endTimestamp" yaml:"endTimestamp"`
 }
 
 type RestartStrategy struct {
@@ -53,16 +53,16 @@ type RestartStrategy struct {
 }
 
 type PrintableTopo struct {
-	Sources []string                 `json:"sources"`
-	Edges   map[string][]interface{} `json:"edges"`
+	Sources []string                 `json:"sources" yaml:"sources"`
+	Edges   map[string][]interface{} `json:"edges" yaml:"edges"`
 }
 
 type GraphNode struct {
-	Type     string                 `json:"type"`
-	NodeType string                 `json:"nodeType"`
-	Props    map[string]interface{} `json:"props"`
+	Type     string                 `json:"type" yaml:"type"`
+	NodeType string                 `json:"nodeType" yaml:"nodeType"`
+	Props    map[string]interface{} `json:"props" yaml:"props"`
 	// UI is a placeholder for ui properties
-	UI map[string]interface{} `json:"ui"`
+	UI map[string]interface{} `json:"ui" yaml:"ui"`
 }
 
 // SourceMeta is the metadata of a source node. It describes what existed stream/table to refer to.
@@ -73,20 +73,20 @@ type SourceMeta struct {
 }
 
 type RuleGraph struct {
-	Nodes map[string]*GraphNode `json:"nodes"`
-	Topo  *PrintableTopo        `json:"topo"`
+	Nodes map[string]*GraphNode `json:"nodes" yaml:"nodes"`
+	Topo  *PrintableTopo        `json:"topo" yaml:"topo"`
 }
 
 // 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"`
-	Id        string                   `json:"id,omitempty"`
-	Name      string                   `json:"name,omitempty"` // The display name of a rule
-	Sql       string                   `json:"sql,omitempty"`
-	Graph     *RuleGraph               `json:"graph,omitempty"`
-	Actions   []map[string]interface{} `json:"actions,omitempty"`
-	Options   *RuleOption              `json:"options,omitempty"`
+	Triggered bool                     `json:"triggered,omitempty" yaml:"triggered,omitempty"`
+	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"`
+	Graph     *RuleGraph               `json:"graph,omitempty" yaml:"graph,omitempty"`
+	Actions   []map[string]interface{} `json:"actions,omitempty" yaml:"actions,omitempty"`
+	Options   *RuleOption              `json:"options,omitempty" yaml:"options,omitempty"`
 }
 
 func (r *Rule) IsLongRunningScheduleRule() bool {
diff --git a/internal/plugin/plugin.go b/internal/plugin/plugin.go
index f7a49a65..aa3cd6d7 100644
--- a/internal/plugin/plugin.go
+++ b/internal/plugin/plugin.go
@@ -45,9 +45,9 @@ type Plugin interface {
 
 // IOPlugin Unify model. Flat all properties for each kind.
 type IOPlugin struct {
-	Name       string   `json:"name"`
-	File       string   `json:"file"`
-	ShellParas []string `json:"shellParas"`
+	Name       string   `json:"name" yaml:"name"`
+	File       string   `json:"file" yaml:"file"`
+	ShellParas []string `json:"shellParas,omitempty" yaml:"shellParas,omitempty"`
 }
 
 func (p *IOPlugin) GetName() string {
diff --git a/internal/schema/schema.go b/internal/schema/schema.go
index 07e303ee..bdab489d 100644
--- a/internal/schema/schema.go
+++ b/internal/schema/schema.go
@@ -22,11 +22,11 @@ import (
 )
 
 type Info struct {
-	Type     def.SchemaType `json:"type"`
-	Name     string         `json:"name"`
-	Content  string         `json:"content"`
-	FilePath string         `json:"file"`
-	SoPath   string         `json:"soFile"`
+	Type     def.SchemaType `json:"type" yaml:"type"`
+	Name     string         `json:"name" yaml:"name"`
+	Content  string         `json:"content,omitempty" yaml:"content,omitempty"`
+	FilePath string         `json:"file,omitempty" yaml:"filePath,omitempty"`
+	SoPath   string         `json:"soFile,omitempty" yaml:"soPath,omitempty"`
 }
 
 func (i *Info) InstallScript() string {
diff --git a/internal/server/rest.go b/internal/server/rest.go
index 350f58ab..435dcae4 100644
--- a/internal/server/rest.go
+++ b/internal/server/rest.go
@@ -38,6 +38,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/v2/internal/server/middleware"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/planner"
+	"github.com/lf-edge/ekuiper/v2/internal/trial"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
@@ -179,6 +180,10 @@ func createRestServer(ip string, port int, needToken bool) *http.Server {
 	r.HandleFunc("/connections", connectionsHandler).Methods(http.MethodGet, http.MethodPost)
 	r.HandleFunc("/connections/status", connectionsStatusHandler).Methods(http.MethodGet)
 	r.HandleFunc("/connection/{id}", connectionHandler).Methods(http.MethodGet, http.MethodDelete)
+	r.HandleFunc("/ruletest", testRuleHandler).Methods(http.MethodPost)
+	r.HandleFunc("/ruletest/{name}/start", testRuleStartHandler).Methods(http.MethodPost)
+	r.HandleFunc("/ruletest/{name}", testRuleStopHandler).Methods(http.MethodDelete)
+	r.HandleFunc("/v2/data/export", yamlConfigurationExportHandler).Methods(http.MethodGet)
 	// r.HandleFunc("/connection/websocket", connectionHandler).Methods(http.MethodGet, http.MethodPost, http.MethodDelete)
 	// Register extended routes
 	for k, v := range components {
@@ -203,9 +208,9 @@ func createRestServer(ip string, port int, needToken bool) *http.Server {
 }
 
 type fileContent struct {
-	Name     string `json:"name"`
-	Content  string `json:"content"`
-	FilePath string `json:"file"`
+	Name     string `json:"name" yaml:"name"`
+	Content  string `json:"content,omitempty" yaml:"content,omitempty"`
+	FilePath string `json:"file,omitempty" yaml:"filePath,omitempty"`
 }
 
 func (f *fileContent) InstallScript() string {
@@ -853,3 +858,45 @@ func exportHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Add("Content-Disposition", "Attachment")
 	http.ServeContent(w, r, name, time.Now(), exported)
 }
+
+func testRuleHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		handleError(w, err, "Invalid body", logger)
+		return
+	}
+	id, err := trial.TrialManager.CreateRule(string(body))
+	if err != nil {
+		handleError(w, err, "", logger)
+		return
+	}
+	result := map[string]any{
+		"id":   id,
+		"port": conf.Config.Source.HttpServerPort,
+	}
+	w.WriteHeader(http.StatusOK)
+	jsonResponse(result, w, logger)
+}
+
+func testRuleStartHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	vars := mux.Vars(r)
+	id := vars["name"]
+	err := trial.TrialManager.StartRule(id)
+	if err != nil {
+		handleError(w, err, "start rule error", logger)
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+	fmt.Fprintf(w, "Test rule %s was started", id)
+}
+
+func testRuleStopHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	vars := mux.Vars(r)
+	id := vars["name"]
+	trial.TrialManager.StopRule(id)
+	w.WriteHeader(http.StatusOK)
+	fmt.Fprintf(w, "Test rule %s was stopped", id)
+}
diff --git a/internal/server/yaml_import_export.go b/internal/server/yaml_import_export.go
new file mode 100644
index 00000000..601105e6
--- /dev/null
+++ b/internal/server/yaml_import_export.go
@@ -0,0 +1,252 @@
+// 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 server
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"time"
+
+	"github.com/pingcap/failpoint"
+	"gopkg.in/yaml.v3"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
+	"github.com/lf-edge/ekuiper/v2/internal/plugin"
+	"github.com/lf-edge/ekuiper/v2/internal/schema"
+	"github.com/lf-edge/ekuiper/v2/internal/service"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+)
+
+type MetaConfiguration struct {
+	SourceConfig     map[string]map[string]any `json:"sourceConfig,omitempty" yaml:"sourceConfig,omitempty"`
+	SinkConfig       map[string]map[string]any `json:"sinkConfig,omitempty" yaml:"sinkConfig,omitempty"`
+	ConnectionConfig map[string]map[string]any `json:"connectionConfig,omitempty" yaml:"connectionConfig,omitempty"`
+	// plugins
+	PortablePlugins map[string]*plugin.IOPlugin `json:"portablePlugins,omitempty" yaml:"portablePlugins,omitempty"`
+	// others
+	Service map[string]*service.ServiceCreationRequest `json:"service,omitempty" yaml:"service,omitempty"`
+	Schema  map[string]*schema.Info                    `json:"schema,omitempty" yaml:"schema,omitempty"`
+	Uploads map[string]*fileContent                    `json:"uploads,omitempty" yaml:"uploads,omitempty"`
+	// rules related
+	Streams map[string]*DatasourceExport `json:"streams" yaml:"streams"`
+	Tables  map[string]*DatasourceExport `json:"tables" yaml:"tables"`
+	Rules   map[string]*def.Rule         `json:"rules" yaml:"rules"`
+}
+
+type DatasourceExport struct {
+	SQL string `json:"sql" yaml:"sql"`
+}
+
+func GenMetaConfiguration() (*MetaConfiguration, error) {
+	m := &MetaConfiguration{
+		Streams: make(map[string]*DatasourceExport),
+		Tables:  make(map[string]*DatasourceExport),
+		Rules:   map[string]*def.Rule{},
+	}
+	var err error
+	if err = addConfiguration(m); err != nil {
+		return nil, err
+	}
+	if err = addPlugins(m); err != nil {
+		return nil, err
+	}
+	if err = addService(m); err != nil {
+		return nil, err
+	}
+	if err = addSchema(m); err != nil {
+		return nil, err
+	}
+	if err = addUploads(m); err != nil {
+		return nil, err
+	}
+
+	rset := rulesetProcessor.ExportRuleSet()
+	for key, sql := range rset.Streams {
+		m.Streams[key] = &DatasourceExport{
+			SQL: sql,
+		}
+	}
+	for key, sql := range rset.Tables {
+		m.Tables[key] = &DatasourceExport{
+			SQL: sql,
+		}
+	}
+	for key, v := range rset.Rules {
+		jm := make(map[string]any)
+		if err := json.Unmarshal([]byte(v), &jm); err != nil {
+			return nil, err
+		}
+		d := &def.Rule{}
+		if err := cast.MapToStruct(jm, d); err != nil {
+			return nil, err
+		}
+		m.Rules[key] = d
+	}
+	return m, nil
+}
+
+func addConfiguration(m *MetaConfiguration) error {
+	var err error
+	m.SourceConfig, err = conf.GetCfgFromKVStorage("sources", "", "")
+	if err != nil {
+		return err
+	}
+	m.SinkConfig, err = conf.GetCfgFromKVStorage("sinks", "", "")
+	if err != nil {
+		return err
+	}
+	m.ConnectionConfig, err = conf.GetCfgFromKVStorage("connections", "", "")
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func addPlugins(m *MetaConfiguration) error {
+	if managers["portable"] != nil {
+		want := make(map[string]*plugin.IOPlugin)
+		pm := managers["portable"].Export()
+		failpoint.Inject("mockYamlExport", func() {
+			e := map[string]*plugin.IOPlugin{
+				"p1": {
+					Name: "p1",
+					File: "path",
+				},
+			}
+			b, _ := json.Marshal(e)
+			pm = map[string]string{
+				"p1": string(b),
+			}
+		})
+		for k, v := range pm {
+			p := &plugin.IOPlugin{}
+			if err := json.Unmarshal([]byte(v), p); err != nil {
+				return err
+			}
+			want[k] = p
+		}
+		m.PortablePlugins = want
+	}
+	return nil
+}
+
+func addSchema(m *MetaConfiguration) error {
+	if managers["schema"] != nil {
+		want := make(map[string]*schema.Info)
+		pm := managers["schema"].Export()
+		failpoint.Inject("mockYamlExport", func() {
+			e := map[string]*schema.Info{
+				"p1": {
+					Name:     "p1",
+					FilePath: "path",
+				},
+			}
+			b, _ := json.Marshal(e)
+			pm = map[string]string{
+				"p1": string(b),
+			}
+		})
+		for k, v := range pm {
+			s := &schema.Info{}
+			if err := json.Unmarshal([]byte(v), s); err != nil {
+				return err
+			}
+			want[k] = s
+		}
+		m.Schema = want
+	}
+	return nil
+}
+
+func addUploads(m *MetaConfiguration) error {
+	ue := uploadsExport()
+	failpoint.Inject("mockYamlExport", func() {
+		e := map[string]*fileContent{
+			"p1": {
+				Name:     "p1",
+				FilePath: "path",
+			},
+		}
+		b, _ := json.Marshal(e)
+		ue = map[string]string{
+			"p1": string(b),
+		}
+	})
+	if len(ue) > 0 {
+		want := make(map[string]*fileContent)
+		for k, v := range ue {
+			f := &fileContent{}
+			if err := json.Unmarshal([]byte(v), f); err != nil {
+				return err
+			}
+			want[k] = f
+		}
+		m.Uploads = want
+	}
+	return nil
+}
+
+func addService(m *MetaConfiguration) error {
+	if managers["service"] != nil {
+		want := make(map[string]*service.ServiceCreationRequest)
+		pm := managers["service"].Export()
+		failpoint.Inject("mockYamlExport", func() {
+			e := map[string]*service.ServiceCreationRequest{
+				"p1": {
+					Name: "p1",
+					File: "path",
+				},
+			}
+			b, _ := json.Marshal(e)
+			pm = map[string]string{
+				"p1": string(b),
+			}
+		})
+		for k, v := range pm {
+			s := &service.ServiceCreationRequest{}
+			if err := json.Unmarshal([]byte(v), s); err != nil {
+				return err
+			}
+			want[k] = s
+		}
+		m.Service = want
+	}
+	return nil
+}
+
+func yamlConfigurationExportHandler(w http.ResponseWriter, r *http.Request) {
+	var yamlBytes []byte
+	var err error
+	var m *MetaConfiguration
+	const name = "ekuiper_export.yaml"
+	switch r.Method {
+	case http.MethodGet:
+		m, err = GenMetaConfiguration()
+		if err != nil {
+			handleError(w, err, "", logger)
+			return
+		}
+		yamlBytes, err = yaml.Marshal(m)
+		if err != nil {
+			handleError(w, err, "", logger)
+			return
+		}
+	}
+	w.Header().Set("Content-Type", "application/octet-stream")
+	w.Header().Add("Content-Disposition", "Attachment")
+	http.ServeContent(w, r, name, time.Now(), bytes.NewReader(yamlBytes))
+}
diff --git a/internal/service/manager.go b/internal/service/manager.go
index 296040a4..0a79cc5d 100644
--- a/internal/service/manager.go
+++ b/internal/service/manager.go
@@ -329,8 +329,8 @@ func (m *Manager) deleteFunc(service, name string) error {
 // ** CRUD of the service files **
 
 type ServiceCreationRequest struct {
-	Name string `json:"name"`
-	File string `json:"file"`
+	Name string `json:"name" yaml:"name"`
+	File string `json:"file" yaml:"file"`
 }
 
 func (s *ServiceCreationRequest) InstallScript() string {
diff --git a/internal/topo/node/sink_node.go b/internal/topo/node/sink_node.go
index c41373e7..5945804e 100644
--- a/internal/topo/node/sink_node.go
+++ b/internal/topo/node/sink_node.go
@@ -66,9 +66,9 @@ func newSinkNode(ctx api.StreamContext, name string, rOpt def.RuleOption, eoflim
 
 func (s *SinkNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 	s.prepareExec(ctx, errCh, "sink")
+	err := s.sink.Connect(ctx)
 	go func() {
 		err := infra.SafeRun(func() error {
-			err := s.sink.Connect(ctx)
 			if err != nil {
 				infra.DrainError(ctx, err, errCh)
 			}
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/logicalPlan.go b/internal/topo/planner/logicalPlan.go
index 5396b6a4..45f6b593 100644
--- a/internal/topo/planner/logicalPlan.go
+++ b/internal/topo/planner/logicalPlan.go
@@ -58,7 +58,7 @@ type PlanExplainInfo struct {
 	T        PlanType `json:"type"`
 	Info     string   `json:"info"`
 	ID       int64    `json:"id"`
-	Children []int64  `json:"children"`
+	Children []int64  `json:"children,omitempty"`
 }
 
 func (p *baseLogicalPlan) Explain() string {
diff --git a/internal/topo/planner/optimizer.go b/internal/topo/planner/optimizer.go
index ba773aef..8e9ad39f 100644
--- a/internal/topo/planner/optimizer.go
+++ b/internal/topo/planner/optimizer.go
@@ -17,6 +17,7 @@ package planner
 var optRuleList = []logicalOptRule{
 	&columnPruner{},
 	&predicatePushDown{},
+	&pushProjectionPlan{},
 }
 
 func optimize(p LogicalPlan) (LogicalPlan, error) {
diff --git a/internal/topo/planner/planner.go b/internal/topo/planner/planner.go
index 6863c8e2..2db84571 100644
--- a/internal/topo/planner/planner.go
+++ b/internal/topo/planner/planner.go
@@ -17,6 +17,7 @@ package planner
 import (
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
@@ -147,6 +148,10 @@ func GetExplainInfoFromLogicalPlan(rule *def.Rule) (string, error) {
 	if err != nil {
 		return "", err
 	}
+	return ExplainFromLogicalPlan(lp, rule.Id)
+}
+
+func ExplainFromLogicalPlan(lp LogicalPlan, ruleID string) (string, error) {
 	var setId func(p LogicalPlan, id int64)
 	setId = func(p LogicalPlan, id int64) {
 		p.SetID(id)
@@ -162,15 +167,16 @@ func GetExplainInfoFromLogicalPlan(rule *def.Rule) (string, error) {
 		tmp := ""
 		res := ""
 		for i := 0; i < level; i++ {
-			tmp += "   "
+			tmp += "\t"
 		}
 		p.BuildExplainInfo()
 		if info, ok := p.(RuleRuntimeInfo); ok {
-			info.BuildSchemaInfo(rule.Id)
+			info.BuildSchemaInfo(ruleID)
 		}
 		// Build the explainInfo of the current layer
-		res += tmp + p.Explain() + "\n"
+		res += tmp + strings.TrimSuffix(p.Explain(), "\n")
 		if len(p.Children()) != 0 {
+			res += "\n"
 			for _, v := range p.Children() {
 				res += tmp + getExplainInfo(v, level+1)
 			}
diff --git a/internal/topo/planner/planner_source.go b/internal/topo/planner/planner_source.go
index 92402488..d31388b4 100644
--- a/internal/topo/planner/planner_source.go
+++ b/internal/topo/planner/planner_source.go
@@ -32,7 +32,7 @@ import (
 
 func transformSourceNode(ctx api.StreamContext, t *DataSourcePlan, mockSourcesProp map[string]map[string]any, ruleId string, options *def.RuleOption, index int) (node.DataSourceNode, []node.OperatorNode, int, error) {
 	isSchemaless := t.isSchemaless
-	_, isMock := mockSourcesProp[string(t.name)]
+	mockProps, isMock := mockSourcesProp[string(t.name)]
 	if isMock {
 		t.streamStmt.Options.TYPE = "simulator"
 	}
@@ -60,13 +60,18 @@ func transformSourceNode(ctx api.StreamContext, t *DataSourcePlan, mockSourcesPr
 			return nil, nil, 0, err
 		}
 	}
-	return splitSource(ctx, t, si, options, index, ruleId, pp)
+	return splitSource(ctx, t, si, options, mockProps, index, ruleId, pp)
 }
 
-func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, options *def.RuleOption, index int, ruleId string, pp node.UnOperation) (node.DataSourceNode, []node.OperatorNode, int, error) {
+func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, options *def.RuleOption, mockProps map[string]any, index int, ruleId string, pp node.UnOperation) (node.DataSourceNode, []node.OperatorNode, int, error) {
 	// Get all props
 	props := nodeConf.GetSourceConf(t.streamStmt.Options.TYPE, t.streamStmt.Options)
 	sp := &SourcePropsForSplit{}
+	if len(mockProps) > 0 {
+		for k, v := range mockProps {
+			props[k] = v
+		}
+	}
 	_ = cast.MapToStruct(props, sp)
 	// Create the connector node as source node
 	var (
@@ -75,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
@@ -94,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/planner/push_proj.go b/internal/topo/planner/push_proj.go
new file mode 100644
index 00000000..8f06617b
--- /dev/null
+++ b/internal/topo/planner/push_proj.go
@@ -0,0 +1,110 @@
+// 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 planner
+
+import (
+	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+)
+
+type pushProjectionPlan struct{}
+
+// pushProjectionPlan inject Projection Plan between the shared Datasource and its father only if the Plan have windowPlan
+// We use Projection to remove the unused column before windowPlan in order to reduce memory consuming
+func (pp *pushProjectionPlan) optimize(plan LogicalPlan) (LogicalPlan, error) {
+	if pp.searchWindowPlan(plan) {
+		ctx := &searchCtx{
+			find: make([]*sharedSource, 0),
+		}
+		pp.searchSharedDataSource(ctx, plan, nil)
+		if len(ctx.find) > 0 {
+			pp.pushProjection(ctx)
+		}
+	}
+	return plan, nil
+}
+
+func (pp *pushProjectionPlan) searchWindowPlan(plan LogicalPlan) bool {
+	switch plan.(type) {
+	case *WindowPlan:
+		return true
+	default:
+	}
+	for _, child := range plan.Children() {
+		search := pp.searchWindowPlan(child)
+		if search {
+			return true
+		}
+	}
+	return false
+}
+
+type searchCtx struct {
+	find []*sharedSource
+}
+
+type sharedSource struct {
+	ds     *DataSourcePlan
+	father LogicalPlan
+}
+
+func (pp *pushProjectionPlan) searchSharedDataSource(ctx *searchCtx, plan, father LogicalPlan) {
+	switch ds := plan.(type) {
+	case *DataSourcePlan:
+		if ds.streamStmt.Options.SHARED {
+			ctx.find = append(ctx.find, &sharedSource{
+				ds:     ds,
+				father: father,
+			})
+		}
+	default:
+	}
+	for _, child := range plan.Children() {
+		pp.searchSharedDataSource(ctx, child, plan)
+	}
+}
+
+func (pp *pushProjectionPlan) pushProjection(ctx *searchCtx) {
+	for _, search := range ctx.find {
+		p := ProjectPlan{
+			fields:      buildFields(search.ds),
+			isAggregate: false,
+			sendMeta:    false,
+			enableLimit: false,
+		}.Init()
+		p.children = []LogicalPlan{search.ds}
+		for index, child := range search.father.Children() {
+			if child.ID() == search.ds.ID() {
+				search.father.Children()[index] = p
+				break
+			}
+		}
+	}
+}
+
+func buildFields(ds *DataSourcePlan) []ast.Field {
+	want := make([]ast.Field, 0)
+	if ds.isWildCard {
+		want = append(want, ast.Field{Expr: &ast.Wildcard{}})
+		return want
+	}
+	for k := range ds.streamFields {
+		want = append(want, ast.Field{Name: k, Expr: &ast.FieldRef{Name: k, StreamName: ds.streamStmt.Name}})
+	}
+	return want
+}
+
+func (pp *pushProjectionPlan) name() string {
+	return "push_projection"
+}
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/internal/trial/manager.go b/internal/trial/manager.go
new file mode 100644
index 00000000..33d363a3
--- /dev/null
+++ b/internal/trial/manager.go
@@ -0,0 +1,89 @@
+// Copyright 2023 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 trial
+
+import (
+	"encoding/json"
+	"fmt"
+	"sync"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/http/httpserver"
+	"github.com/lf-edge/ekuiper/v2/internal/topo"
+)
+
+// TrialManager Manager Initialized in the binder
+var TrialManager = &Manager{
+	runs: make(map[string]Run),
+}
+
+// Manager In memory manager for all trial rules
+type Manager struct {
+	sync.RWMutex
+	// ruleId -> *Topo
+	runs map[string]Run
+}
+
+type Run struct {
+	def  *RunDef
+	topo *topo.Topo
+}
+
+func (m *Manager) CreateRule(ruleDef string) (string, error) {
+	def := &RunDef{}
+	err := json.Unmarshal([]byte(ruleDef), def)
+	if err != nil {
+		return "", fmt.Errorf("fail to parse rule definition %s: %s", ruleDef, err)
+	}
+	m.Lock()
+	defer m.Unlock()
+	// If the rule exists, stop it first
+	if r, ok := m.runs[def.Id]; ok {
+		r.topo.Cancel()
+		conf.Log.Warnf("stop last run of test rule %s", def.Id)
+	}
+	t, err := create(def)
+	if err != nil {
+		return "", err
+	}
+	m.runs[def.Id] = Run{
+		def:  def,
+		topo: t,
+	}
+	return def.Id, nil
+}
+
+func (m *Manager) StopRule(ruleId string) {
+	m.Lock()
+	defer m.Unlock()
+	if r, ok := m.runs[ruleId]; ok {
+		r.topo.Cancel()
+		delete(m.runs, ruleId)
+		httpserver.UnRegisterWebSocketEndpoint(r.def.endpoint)
+	} else {
+		conf.Log.Warnf("try to stop test rule %s but it is not found", ruleId)
+	}
+}
+
+func (m *Manager) StartRule(ruleId string) error {
+	m.RLock()
+	defer m.RUnlock()
+	if r, ok := m.runs[ruleId]; ok {
+		trialRun(r.topo)
+	} else {
+		return fmt.Errorf("try to start test rule %s but it is not found", ruleId)
+	}
+	return nil
+}
diff --git a/internal/trial/run.go b/internal/trial/run.go
new file mode 100644
index 00000000..066e9864
--- /dev/null
+++ b/internal/trial/run.go
@@ -0,0 +1,133 @@
+// 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.
+// 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 trial
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/http/httpserver"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
+	"github.com/lf-edge/ekuiper/v2/internal/topo"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/planner"
+	"github.com/lf-edge/ekuiper/v2/pkg/infra"
+)
+
+type RunDef struct {
+	Id        string                    `json:"id"`
+	Sql       string                    `json:"sql"`
+	Mock      map[string]map[string]any `json:"mockSource"`
+	SinkProps map[string]any            `json:"sinkProps"`
+
+	endpoint string
+}
+
+func genTrialRuleID(def *RunDef) string {
+	return "$$_" + uuid.New().String() + def.Id
+}
+
+func genTrialRule(rd *RunDef, sinkProps map[string]interface{}) *def.Rule {
+	id := genTrialRuleID(rd)
+	rt := def.GetDefaultRule(id, rd.Sql)
+	rt.Actions = []map[string]interface{}{
+		{
+			"websocket": sinkProps,
+		},
+	}
+	return rt
+}
+
+func create(def *RunDef) (*topo.Topo, error) {
+	endpoint := "/test/" + def.Id
+	def.endpoint = endpoint
+	_, _, err := httpserver.RegisterWebSocketEndpoint(context.Background(), endpoint)
+	if err != nil {
+		return nil, err
+	}
+	sinkProps := map[string]any{
+		"path":      endpoint,
+		"sendError": true,
+	}
+	for k, v := range def.SinkProps {
+		sinkProps[k] = v
+	}
+	trialRule := genTrialRule(def, sinkProps)
+	// Add trial run prefix for rule id to avoid duplicate rule id with real rules in runtime or other trial rule
+	tp, err := planner.PlanSQLWithSourcesAndSinks(trialRule, prepareSourceProps(def.Mock))
+	if err != nil {
+		return nil, fmt.Errorf("fail to run rule %s: %s", def.Id, err)
+	}
+	// Try run
+	// TODO currently some static validations are done in runtime, so start to run to detect them. This adds time penalty for this API.
+	// 	In the future, we should do it in planning.
+	err = infra.SafeRun(func() error {
+		select {
+		case e := <-tp.Open():
+			if e != nil {
+				return e
+			}
+		case <-time.After(10 * time.Millisecond):
+			return nil
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("fail to run rule %s: %s", def.Id, err)
+	}
+	tp.Cancel()
+	return tp, nil
+}
+
+func trialRun(tp *topo.Topo) {
+	go func() {
+		timeout := time.After(5 * time.Minute)
+		err := infra.SafeRun(func() error {
+			select {
+			case err := <-tp.Open():
+				if err != nil {
+					conf.Log.Errorf("closing test run for error: %v", err)
+					tp.Cancel()
+					return err
+				}
+			case <-timeout:
+				tp.GetContext().GetLogger().Debugf("trial run stops after timeout")
+				tp.Cancel()
+			}
+			return nil
+		})
+		if err != nil {
+			conf.Log.Debugf("trial run error: %v", err)
+		}
+	}()
+}
+
+func prepareSourceProps(mockSource map[string]map[string]any) map[string]map[string]any {
+	for source, props := range mockSource {
+		v, ok := props["interval"]
+		if ok {
+			vi, ok := v.(int)
+			if ok {
+				props["interval"] = fmt.Sprintf("%dms", vi)
+				mockSource[source] = props
+				continue
+			}
+		}
+	}
+	return mockSource
+}
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/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/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 298c881b..8ba54e73 100644
--- a/internal/server/rest_test.go
+++ b/internal/server/rest_test.go
@@ -96,6 +96,9 @@ func (suite *RestTestSuite) SetupTest() {
 	r.HandleFunc("/data/import/status", configurationStatusHandler).Methods(http.MethodGet)
 	r.HandleFunc("/connections", connectionsHandler).Methods(http.MethodGet, http.MethodPost)
 	r.HandleFunc("/connection/{id}", connectionHandler).Methods(http.MethodGet, http.MethodDelete)
+	r.HandleFunc("/ruletest", testRuleHandler).Methods(http.MethodPost)
+	r.HandleFunc("/ruletest/{name}/start", testRuleStartHandler).Methods(http.MethodPost)
+	r.HandleFunc("/ruletest/{name}", testRuleStopHandler).Methods(http.MethodDelete)
 	// r.HandleFunc("/connection/websocket", connectionHandler).Methods(http.MethodGet, http.MethodPost, http.MethodDelete)
 	r.HandleFunc("/metadata/sinks/{name}/confKeys/{confKey}", sinkConfKeyHandler).Methods(http.MethodDelete, http.MethodPut)
 	suite.r = r
@@ -206,7 +209,47 @@ 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
 	if rules, err := ruleProcessor.GetAllRules(); err != nil {
 		logger.Infof("Start rules error: %s", err)
@@ -339,8 +382,10 @@ func (suite *RestTestSuite) Test_rulesManageHandler() {
 	suite.r.ServeHTTP(w1, req1)
 	returnVal, _ = io.ReadAll(w1.Result().Body) //nolint
 	returnStr := string(returnVal)
-	expect = "{\"type\":\"ProjectPlan\",\"info\":\"Fields:[ * ]\",\"id\":0,\"children\":[1]}\n\n   {\"type\":\"DataSourcePlan\",\"info\":\"StreamName: alert\",\"id\":1,\"children\":null}\n\n"
-	assert.Equal(suite.T(), expect, returnStr)
+	expect = `
+{"type":"ProjectPlan","info":"Fields:[ * ]","id":0,"children":[1]}
+	{"type":"DataSourcePlan","info":"StreamName: alert","id":1}`
+	assert.Equal(suite.T(), strings.Trim(expect, "\n"), returnStr)
 
 	req1, _ = http.NewRequest(http.MethodGet, "http://localhost:8080/rules/rule32211/explain", bytes.NewBufferString("any"))
 	w1 = httptest.NewRecorder()
@@ -419,7 +464,7 @@ func (suite *RestTestSuite) Test_rulesManageHandler() {
 	suite.r.ServeHTTP(w1, req1)
 
 	returnVal, _ = io.ReadAll(w1.Result().Body)
-	expect = `{"triggered":true,"id":"rule321","sql":"select * from alert","actions":[{"nop":{}}],"options":{"debug":false,"logFilename":"","isEventTime":false,"lateTolerance":"1s","concurrency":1,"bufferLength":1024,"sendMetaToSink":false,"sendError":true,"qos":0,"checkpointInterval":"5m0s","restartStrategy":{"attempts":0,"delay":"1s","multiplier":2,"maxDelay":"30s","jitter":0.1},"cron":"","duration":"","cronDatetimeRange":null}}`
+	expect = `{"triggered":true,"id":"rule321","sql":"select * from alert","actions":[{"nop":{}}],"options":{"lateTolerance":"1s","concurrency":1,"bufferLength":1024,"sendError":true,"checkpointInterval":"5m0s","restartStrategy":{"attempts":0,"delay":"1s","multiplier":2,"maxDelay":"30s","jitter":0.1}}}`
 	assert.Equal(suite.T(), expect, string(returnVal))
 
 	// delete rule
diff --git a/internal/server/trial_test.go b/internal/server/trial_test.go
new file mode 100644
index 00000000..9451fbc1
--- /dev/null
+++ b/internal/server/trial_test.go
@@ -0,0 +1,119 @@
+// 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 server
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"time"
+
+	"github.com/gorilla/websocket"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/http/httpserver"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
+	"github.com/lf-edge/ekuiper/v2/internal/processor"
+	"github.com/lf-edge/ekuiper/v2/internal/trial"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+func (suite *RestTestSuite) TestRuleTest() {
+	ip := "127.0.0.1"
+	port := 10087
+	httpserver.InitGlobalServerManager(ip, port, nil)
+	connection.InitConnectionManager4Test()
+	conf.IsTesting = true
+	conf.InitConf()
+	dataDir, err := conf.GetDataLoc()
+	require.NoError(suite.T(), err)
+	require.NoError(suite.T(), store.SetupDefault(dataDir))
+
+	rd := &trial.RunDef{
+		Id:  "mock1",
+		Sql: "select * from trialDemo",
+		Mock: map[string]map[string]any{
+			"trialDemo": {
+				"data": []map[string]any{
+					{
+						"a": 1,
+					},
+				},
+				"loop":     true,
+				"interval": 100,
+			},
+		},
+		SinkProps: map[string]any{
+			"sendSingle": true,
+		},
+	}
+	b, err := json.Marshal(rd)
+	require.NoError(suite.T(), err)
+	req2, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/ruletest", bytes.NewBuffer(b))
+	w2 := httptest.NewRecorder()
+	suite.r.ServeHTTP(w2, req2)
+	require.NotEqual(suite.T(), http.StatusOK, w2.Code)
+
+	p := processor.NewStreamProcessor()
+	p.ExecStmt("DROP STREAM trialDemo")
+	_, err = p.ExecStmt("CREATE STREAM trialDemo () WITH (DATASOURCE=\"trialDemo\", TYPE=\"simulator\", FORMAT=\"json\", KEY=\"ts\")")
+	require.NoError(suite.T(), err)
+	defer p.ExecStmt("DROP STREAM trialDemo")
+
+	req2, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/ruletest", bytes.NewBuffer(b))
+	w2 = httptest.NewRecorder()
+	suite.r.ServeHTTP(w2, req2)
+	require.Equal(suite.T(), http.StatusOK, w2.Code)
+	u := url.URL{Scheme: "ws", Host: "localhost:10087", Path: "/test/mock1"}
+	c1, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
+	assert.NoError(suite.T(), err)
+	defer c1.Close()
+
+	recvCh := make(chan []byte, 10)
+	closeCh := make(chan struct{}, 10)
+	go func() {
+		_, data, err := c1.ReadMessage()
+		require.NoError(suite.T(), err)
+		recvCh <- data
+	}()
+	go func() {
+		for {
+			select {
+			case <-closeCh:
+				return
+			default:
+				timex.Add(100 * time.Millisecond)
+				time.Sleep(100 * time.Millisecond)
+			}
+		}
+	}()
+
+	time.Sleep(100 * time.Millisecond)
+	req2, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/ruletest/mock1/start", bytes.NewBuffer([]byte{}))
+	w2 = httptest.NewRecorder()
+	suite.r.ServeHTTP(w2, req2)
+	require.Equal(suite.T(), http.StatusOK, w2.Code)
+	require.Equal(suite.T(), `{"a":1}`, string(<-recvCh))
+	req2, _ = http.NewRequest(http.MethodDelete, "http://localhost:8080/ruletest/mock1", bytes.NewBuffer([]byte{}))
+	w2 = httptest.NewRecorder()
+	suite.r.ServeHTTP(w2, req2)
+	require.Equal(suite.T(), http.StatusOK, w2.Code)
+	closeCh <- struct{}{}
+}
diff --git a/internal/server/yaml_import_export_test.go b/internal/server/yaml_import_export_test.go
new file mode 100644
index 00000000..dfe4125f
--- /dev/null
+++ b/internal/server/yaml_import_export_test.go
@@ -0,0 +1,69 @@
+// 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 server
+
+import (
+	"testing"
+
+	"github.com/pingcap/failpoint"
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
+)
+
+func TestMetaConfiguration(t *testing.T) {
+	conf.InitConf()
+	conf.IsTesting = true
+	connection.InitConnectionManager4Test()
+	for _, v := range components {
+		v.register()
+	}
+	InitConfManagers()
+	prepare(t)
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/server/mockYamlExport", "return(true)")
+	defer failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/server/mockYamlExport")
+	m, err := GenMetaConfiguration()
+	require.NoError(t, err)
+	require.True(t, len(m.SourceConfig) > 0)
+	require.True(t, len(m.SinkConfig) > 0)
+	require.True(t, len(m.ConnectionConfig) > 0)
+	require.True(t, len(m.PortablePlugins) > 0)
+	require.True(t, len(m.Service) > 0)
+	require.True(t, len(m.Schema) > 0)
+	require.True(t, len(m.Uploads) > 0)
+	require.True(t, len(m.Streams) > 0)
+	require.True(t, len(m.Rules) > 0)
+}
+
+func prepare(t *testing.T) {
+	deleteRule("metaConf")
+	ruleProcessor.ExecDrop("metaConf")
+	streamProcessor.ExecStreamSql(`drop stream metaConfTest`)
+	require.NoError(t, conf.WriteCfgIntoKVStorage("sources", "mqtt", "demo1", map[string]any{
+		"a": 1,
+	}))
+	require.NoError(t, conf.WriteCfgIntoKVStorage("sinks", "mqtt", "demo1", map[string]any{
+		"a": 1,
+	}))
+	require.NoError(t, conf.WriteCfgIntoKVStorage("connections", "mqtt", "demo1", map[string]any{
+		"a": 1,
+	}))
+	_, err := streamProcessor.ExecStreamSql(`create stream metaConfTest() WITH (DATASOURCE="/API/DATA",CONF_KEY="demo1")`)
+	require.NoError(t, err)
+	rulejson := `{"trigger":false,"id":"metaConf","sql":"select * from metaConfTest","actions":[{"log":{}}]}`
+	_, err = createRule("metaConf", rulejson)
+	require.NoError(t, err)
+}
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/explainInfo_test.go b/internal/topo/planner/explainInfo_test.go
index 8c29ac5d..2ff61e45 100644
--- a/internal/topo/planner/explainInfo_test.go
+++ b/internal/topo/planner/explainInfo_test.go
@@ -24,7 +24,7 @@ func TestDataSourcePlanExplainInfo(t *testing.T) {
 					"field3": {},
 				},
 			},
-			res: "{\"type\":\"DataSourcePlan\",\"info\":\"StreamName: test1, Fields:[ field1, field2, field3 ]\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"DataSourcePlan\",\"info\":\"StreamName: test1, Fields:[ field1, field2, field3 ]\",\"id\":0}\n",
 			t:   "DataSourcePlan",
 		},
 		{
@@ -53,19 +53,19 @@ func TestDataSourcePlanExplainInfo(t *testing.T) {
 					"s3": {},
 				},
 			},
-			res: "{\"type\":\"DataSourcePlan\",\"info\":\"StreamName: test3, Fields:[ column1, column2, id ], StreamFields:[ s1, s2, s3 ]\",\"id\":2,\"children\":null}\n",
+			res: "{\"type\":\"DataSourcePlan\",\"info\":\"StreamName: test3, Fields:[ column1, column2, id ], StreamFields:[ s1, s2, s3 ]\",\"id\":2}\n",
 			t:   "DataSourcePlan",
 		},
 		{
 			p: &DataSourcePlan{
 				name: "test4",
 			},
-			res: "{\"type\":\"DataSourcePlan\",\"info\":\"StreamName: test4\",\"id\":3,\"children\":null}\n",
+			res: "{\"type\":\"DataSourcePlan\",\"info\":\"StreamName: test4\",\"id\":3}\n",
 			t:   "DataSourcePlan",
 		},
 		{
 			p:   &DataSourcePlan{},
-			res: "{\"type\":\"DataSourcePlan\",\"info\":\"\",\"id\":4,\"children\":null}\n",
+			res: "{\"type\":\"DataSourcePlan\",\"info\":\"\",\"id\":4}\n",
 			t:   "DataSourcePlan",
 		},
 	}
@@ -108,7 +108,7 @@ func TestAggregatePlanExplainInfo(t *testing.T) {
 					},
 				},
 			},
-			res: "{\"type\":\"AggregatePlan\",\"info\":\"Dimension:{ Call:{ name:lpad, args:[$$default.name, 1] } }\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"AggregatePlan\",\"info\":\"Dimension:{ Call:{ name:lpad, args:[$$default.name, 1] } }\",\"id\":0}\n",
 			t:   "AggregatePlan",
 		},
 		{
@@ -139,7 +139,7 @@ func TestAggregatePlanExplainInfo(t *testing.T) {
 					ast.Dimension{Expr: &ast.FieldRef{Name: "department", StreamName: ast.DefaultStream}},
 				},
 			},
-			res: "{\"type\":\"AggregatePlan\",\"info\":\"Dimension:{ $$default.department }\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"AggregatePlan\",\"info\":\"Dimension:{ $$default.department }\",\"id\":0}\n",
 			t:   "AggregatePlan",
 		},
 	}
@@ -190,7 +190,7 @@ func TestAnalyticFuncsPlanExplainInfo(t *testing.T) {
 					},
 				},
 			},
-			res: "{\"type\":\"AnalyticFuncsPlan\",\"info\":\"Funcs:[ Call:{ name:lag, args:[src1.temp] } ], FieldFuncs:[ Call:{ name:lag, args:[src1.name] }, Call:{ name:latest, args:[Call:{ name:lag, args:[src1.name] }] } ]\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"AnalyticFuncsPlan\",\"info\":\"Funcs:[ Call:{ name:lag, args:[src1.temp] } ], FieldFuncs:[ Call:{ name:lag, args:[src1.name] }, Call:{ name:latest, args:[Call:{ name:lag, args:[src1.name] }] } ]\",\"id\":0}\n",
 			t:   "AnalyticFuncsPlan",
 		},
 		{
@@ -278,7 +278,7 @@ func TestFilterPlanExplainInfo(t *testing.T) {
 					RHS: &ast.StringLiteral{Val: "v1"},
 				},
 			},
-			res: "{\"type\":\"FilterPlan\",\"info\":\"Condition:{ binaryExpr:{ src1.name = v1 } }, \",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"FilterPlan\",\"info\":\"Condition:{ binaryExpr:{ src1.name = v1 } }, \",\"id\":0}\n",
 			t:   "FilterPlan",
 		},
 		{
@@ -336,7 +336,7 @@ func TestHavingPlanExplainInfo(t *testing.T) {
 					RHS: &ast.IntegerLiteral{Val: 2},
 				},
 			},
-			res: "{\"type\":\"HavingPlan\",\"info\":\"Condition:{ binaryExpr:{ Call:{ name:count, args:[*] } > 2 } }, \",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"HavingPlan\",\"info\":\"Condition:{ binaryExpr:{ Call:{ name:count, args:[*] } > 2 } }, \",\"id\":0}\n",
 			t:   "HavingPlan",
 		},
 		{
@@ -413,7 +413,7 @@ func TestJoinAlignPlanExplainInfo(t *testing.T) {
 			p: &JoinAlignPlan{
 				Emitters: []string{"tableInPlanner"},
 			},
-			res: "{\"type\":\"JoinAlignPlan\",\"info\":\"Emitters:[ tableInPlanner ]\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"JoinAlignPlan\",\"info\":\"Emitters:[ tableInPlanner ]\",\"id\":0}\n",
 			t:   "JoinAlignPlan",
 		},
 		{
@@ -478,7 +478,7 @@ func TestJoinPlanExplainInfo(t *testing.T) {
 					},
 				}},
 			},
-			res: "{\"type\":\"JoinPlan\",\"info\":\"Joins:[ { joinType:INNER_JOIN, binaryExpr:{ binaryExpr:{ binaryExpr:{ src1.temp > 20 } OR binaryExpr:{ src2.hum > 60 } } AND binaryExpr:{ src1.id1 = src2.id2 } } } ]\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"JoinPlan\",\"info\":\"Joins:[ { joinType:INNER_JOIN, binaryExpr:{ binaryExpr:{ binaryExpr:{ src1.temp > 20 } OR binaryExpr:{ src2.hum > 60 } } AND binaryExpr:{ src1.id1 = src2.id2 } } } ]\",\"id\":0}\n",
 			t:   "JoinPlan",
 		},
 		{
@@ -557,7 +557,7 @@ func TestLookupPlanExplainInfo(t *testing.T) {
 					},
 				},
 			},
-			res: "{\"type\":\"LookupPlan\",\"info\":\"Join:{ joinType:LEFT_JOIN, expr:binaryExpr:{ left.device_id = good.id } }\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"LookupPlan\",\"info\":\"Join:{ joinType:LEFT_JOIN, expr:binaryExpr:{ left.device_id = good.id } }\",\"id\":0}\n",
 			t:   "LookupPlan",
 		},
 		{
@@ -907,7 +907,7 @@ func TestWatermarkPlanExplainInfo(t *testing.T) {
 				Emitters:      []string{"id", "grade"},
 				SendWatermark: false,
 			},
-			res: "{\"type\":\"WatermarkPlan\",\"info\":\"Emitters:[ id, grade ], SendWatermark:false\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"WatermarkPlan\",\"info\":\"Emitters:[ id, grade ], SendWatermark:false\",\"id\":0}\n",
 			t:   "WatermarkPlan",
 		},
 		{
@@ -969,7 +969,7 @@ func TestWindowFuncPlanExplainInfo(t *testing.T) {
 					},
 				},
 			},
-			res: "{\"type\":\"WindowFuncPlan\",\"info\":\"windowFuncFields:[ {name:AName, expr:}, {name:AName, expr:}, {name:Name}, {name:AName} ]\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"WindowFuncPlan\",\"info\":\"windowFuncFields:[ {name:AName, expr:}, {name:AName, expr:}, {name:Name}, {name:AName} ]\",\"id\":0}\n",
 			t:   "WindowFuncPlan",
 		},
 		{
@@ -1015,7 +1015,7 @@ func TestOrderPlanExplainInfo(t *testing.T) {
 			p: &OrderPlan{
 				SortFields: []ast.SortField{{Uname: "name", Name: "name", Ascending: false, FieldExpr: &ast.FieldRef{Name: "name", StreamName: ast.DefaultStream}}},
 			},
-			res: "{\"type\":\"OrderPlan\",\"info\":\"SortFields:[ sortField:{ name:name, ascending:false, fieldExpr:{ $$default.name } } ]\",\"id\":0,\"children\":null}\n",
+			res: "{\"type\":\"OrderPlan\",\"info\":\"SortFields:[ sortField:{ name:name, ascending:false, fieldExpr:{ $$default.name } } ]\",\"id\":0}\n",
 			t:   "OrderPlan",
 		},
 		{
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/internal/topo/planner/planner_test.go b/internal/topo/planner/planner_test.go
index 4b60d32d..0a897a6e 100644
--- a/internal/topo/planner/planner_test.go
+++ b/internal/topo/planner/planner_test.go
@@ -4863,112 +4863,3 @@ func TestTransformSourceNode(t *testing.T) {
 		})
 	}
 }
-
-func TestGetLogicalPlanForExplain(t *testing.T) {
-	kv, err := store.GetKV("stream")
-	if err != nil {
-		t.Error(err)
-		return
-	}
-	streamSqls := map[string]string{
-		"src1": `CREATE STREAM src1 (
-					id1 BIGINT,
-					temp BIGINT,
-					name string,
-					myarray array(string)
-				) WITH (DATASOURCE="src1", FORMAT="json", KEY="ts");`,
-		"src2": `CREATE STREAM src2 (
-					id2 BIGINT,
-					hum BIGINT
-				) WITH (DATASOURCE="src2", FORMAT="json", KEY="ts", TIMESTAMP_FORMAT="YYYY-MM-dd HH:mm:ss");`,
-		"tableInPlanner": `CREATE TABLE tableInPlanner (
-					id BIGINT,
-					name STRING,
-					value STRING,
-					hum BIGINT
-				) WITH (TYPE="file");`,
-	}
-	types := map[string]ast.StreamType{
-		"src1":           ast.TypeStream,
-		"src2":           ast.TypeStream,
-		"tableInPlanner": ast.TypeTable,
-	}
-	for name, sql := range streamSqls {
-		s, err := json.Marshal(&xsql.StreamInfo{
-			StreamType: types[name],
-			Statement:  sql,
-		})
-		if err != nil {
-			t.Error(err)
-			t.Fail()
-		}
-		err = kv.Set(name, string(s))
-		if err != nil {
-			t.Error(err)
-			t.Fail()
-		}
-	}
-	streams := make(map[string]*ast.StreamStmt)
-	for n := range streamSqls {
-		streamStmt, err := xsql.GetDataSource(kv, n)
-		if err != nil {
-			t.Errorf("fail to get stream %s, please check if stream is created", n)
-			return
-		}
-		streams[n] = streamStmt
-	}
-
-	ref := &ast.AliasRef{
-		Expression: &ast.Call{
-			Name:     "row_number",
-			FuncType: ast.FuncTypeWindow,
-		},
-	}
-	ref.SetRefSource([]string{})
-
-	tests := []struct {
-		rule *def.Rule
-		res  string
-		err  string
-	}{
-		{
-			rule: &def.Rule{
-				Triggered: false,
-				Id:        "test",
-				Sql:       "select name, row_number() as index from src1",
-				Actions: []map[string]interface{}{
-					{
-						"log": map[string]interface{}{},
-					},
-				},
-				Options: defaultOption,
-			},
-			res: "{\"type\":\"ProjectPlan\",\"info\":\"Fields:[ $$alias.index, src1.name ]\",\"id\":0,\"children\":[1]}\n\n   {\"type\":\"WindowFuncPlan\",\"info\":\"windowFuncFields:[ {name:index, expr:$$alias.index} ]\",\"id\":1,\"children\":[2]}\n\n         {\"type\":\"DataSourcePlan\",\"info\":\"StreamName: src1, StreamFields:[ name ]\",\"id\":2,\"children\":null}\n\n",
-		},
-		{
-			rule: &def.Rule{
-				Triggered: false,
-				Id:        "test",
-				Sql:       "select name, row_number() from src1",
-				Actions: []map[string]interface{}{
-					{
-						"log": map[string]interface{}{},
-					},
-				},
-				Options: defaultOption,
-			},
-			res: "{\"type\":\"ProjectPlan\",\"info\":\"Fields:[ src1.name, Call:{ name:row_number } ]\",\"id\":0,\"children\":[1]}\n\n   {\"type\":\"WindowFuncPlan\",\"info\":\"windowFuncFields:[ {name:row_number, expr:Call:{ name:row_number }} ]\",\"id\":1,\"children\":[2]}\n\n         {\"type\":\"DataSourcePlan\",\"info\":\"StreamName: src1, StreamFields:[ name ]\",\"id\":2,\"children\":null}\n\n",
-		},
-	}
-	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
-
-	for i, tt := range tests {
-		explain, err := GetExplainInfoFromLogicalPlan(tt.rule)
-		if err != nil {
-			t.Errorf(err.Error())
-		}
-		if !reflect.DeepEqual(explain, tt.res) {
-			t.Errorf("case %d: expect validate %v but got %v", i, tt.res, explain)
-		}
-	}
-}
diff --git a/internal/topo/planner/push_proj_test.go b/internal/topo/planner/push_proj_test.go
new file mode 100644
index 00000000..d187c1d1
--- /dev/null
+++ b/internal/topo/planner/push_proj_test.go
@@ -0,0 +1,80 @@
+// 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 planner
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
+	"github.com/lf-edge/ekuiper/v2/internal/xsql"
+	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+)
+
+func TestPushProjection(t *testing.T) {
+	kv, err := store.GetKV("stream")
+	require.NoError(t, err)
+	require.NoError(t, prepareStream())
+	sql := `select a as c from sharedStream group by countwindow(2)`
+	stmt, err := xsql.NewParser(strings.NewReader(sql)).Parse()
+	assert.NoError(t, err)
+	p, err := createLogicalPlan(stmt, &def.RuleOption{
+		Qos: 0,
+	}, kv)
+	require.NoError(t, err)
+	explain, err := ExplainFromLogicalPlan(p, "")
+	require.NoError(t, err)
+	expect := `
+{"type":"ProjectPlan","info":"Fields:[ $$alias.c ]","id":0,"children":[1]}
+	{"type":"WindowPlan","info":"{ length:2, windowType:COUNT_WINDOW, limit: 0 }","id":1,"children":[2]}
+			{"type":"ProjectPlan","info":"Fields:[ sharedStream.a ]","id":2,"children":[3]}
+					{"type":"DataSourcePlan","info":"StreamName: sharedStream, StreamFields:[ a ]","id":3}`
+	require.Equal(t, strings.TrimPrefix(expect, "\n"), explain)
+}
+
+func prepareStream() error {
+	kv, err := store.GetKV("stream")
+	if err != nil {
+		return err
+	}
+	streamSqls := map[string]string{
+		"sharedStream": `CREATE STREAM sharedStream (
+					a BIGINT,
+					b BIGINT,
+				) WITH (DATASOURCE="src1", SHARED="true");`,
+	}
+	types := map[string]ast.StreamType{
+		"sharedStream": ast.TypeStream,
+	}
+	for name, sql := range streamSqls {
+		s, err := json.Marshal(&xsql.StreamInfo{
+			StreamType: types[name],
+			Statement:  sql,
+		})
+		if err != nil {
+			return err
+		}
+		err = kv.Set(name, string(s))
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
diff --git a/internal/trial/manager_test.go b/internal/trial/manager_test.go
new file mode 100644
index 00000000..71a1df12
--- /dev/null
+++ b/internal/trial/manager_test.go
@@ -0,0 +1,172 @@
+// Copyright 2023 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 trial
+
+import (
+	"net/url"
+	"testing"
+	"time"
+
+	"github.com/gorilla/websocket"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/http/httpserver"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
+	"github.com/lf-edge/ekuiper/v2/internal/processor"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+// Run two test rules in parallel. Rerun one of the rules
+func TestTrialRule(t *testing.T) {
+	ip := "127.0.0.1"
+	port := 10087
+	httpserver.InitGlobalServerManager(ip, port, nil)
+	connection.InitConnectionManager4Test()
+	conf.IsTesting = true
+	conf.InitConf()
+	dataDir, err := conf.GetDataLoc()
+	require.NoError(t, err)
+	require.NoError(t, store.SetupDefault(dataDir))
+	p := processor.NewStreamProcessor()
+	p.ExecStmt("DROP STREAM demo")
+	// Test 1 wrong rule
+	mockDef1 := `{"id":"rule1","sql":"select * from demo","mockSource":{"demo":{"data":[{"name":"demo","value":1}],"interval":100,"loop":true}},"sinkProps":{"sendSingle":true}}`
+	_, err = TrialManager.CreateRule(mockDef1)
+	require.Error(t, err)
+	require.Equal(t, "fail to run rule rule1: fail to get stream demo, please check if stream is created", err.Error())
+
+	_, err = p.ExecStmt("CREATE STREAM demo () WITH (DATASOURCE=\"demo\", TYPE=\"simulator\", FORMAT=\"json\", KEY=\"ts\")")
+	require.NoError(t, err)
+	defer p.ExecStmt("DROP STREAM demo")
+
+	// Test 2 valid rule with mock
+	testValidTrial(t, mockDef1)
+
+	// Test 3 Runtime error rule
+	testRuntimeErrorTrial(t)
+
+	// Test 4 Rule without mock
+	testRealSourceTrial(t)
+	require.Equal(t, 0, len(TrialManager.runs))
+}
+
+func testValidTrial(t *testing.T, mockDef1 string) {
+	// Test 2 valid rule with mock
+	id, err := TrialManager.CreateRule(mockDef1)
+	require.NoError(t, err)
+	require.Equal(t, "rule1", id)
+	// Read from ws
+	u := url.URL{Scheme: "ws", Host: "localhost:10087", Path: "/test/rule1"}
+	c1, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
+	assert.NoError(t, err)
+	recvCh := make(chan []byte, 10)
+	closeCh := make(chan struct{}, 10)
+	go func() {
+		_, data, err := c1.ReadMessage()
+		require.NoError(t, err)
+		recvCh <- data
+	}()
+	go func() {
+		for {
+			select {
+			case <-closeCh:
+				return
+			default:
+				timex.Add(100 * time.Millisecond)
+				time.Sleep(100 * time.Millisecond)
+			}
+		}
+	}()
+	time.Sleep(100 * time.Millisecond)
+	require.NoError(t, TrialManager.StartRule("rule1"))
+	require.Equal(t, []byte(`{"name":"demo","value":1}`), <-recvCh)
+	c1.Close()
+	TrialManager.StopRule("rule1")
+	closeCh <- struct{}{}
+}
+
+func testRuntimeErrorTrial(t *testing.T) {
+	// Test 3 Runtime error rule
+	mockDefErr := `{"id":"ruleErr","sql":"select name + value from demo","mockSource":{"demo":{"data":[{"name":"demo","value":1}],"interval":100,"loop":true}},"sinkProps":{"sendSingle":true}}`
+	id, err := TrialManager.CreateRule(mockDefErr)
+	require.NoError(t, err)
+	require.Equal(t, "ruleErr", id)
+	// Read from ws
+	u := url.URL{Scheme: "ws", Host: "localhost:10087", Path: "/test/ruleErr"}
+	c2, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
+	require.NoError(t, err)
+	recvCh := make(chan []byte, 10)
+	closeCh := make(chan struct{}, 10)
+	go func() {
+		_, data, err := c2.ReadMessage()
+		require.NoError(t, err)
+		recvCh <- data
+	}()
+	go func() {
+		for {
+			select {
+			case <-closeCh:
+				return
+			default:
+				timex.Add(100 * time.Millisecond)
+				time.Sleep(100 * time.Millisecond)
+			}
+		}
+	}()
+	time.Sleep(10 * time.Millisecond)
+	require.NoError(t, TrialManager.StartRule(id))
+	require.Equal(t, `run Select error: expr: binaryExpr:{ demo.name + demo.value } meet error, err:invalid operation string(demo) + float64(1)`, string(<-recvCh))
+	TrialManager.StopRule(id)
+	closeCh <- struct{}{}
+	c2.Close()
+}
+
+func testRealSourceTrial(t *testing.T) {
+	noMockDef := `{"id":"rule2","sql":"select * from demo","sinkProps":{"sendSingle":true}}`
+	id, err := TrialManager.CreateRule(noMockDef)
+	assert.Equal(t, "rule2", id)
+	assert.NoError(t, err)
+
+	u := url.URL{Scheme: "ws", Host: "localhost:10087", Path: "/test/rule2"}
+	c3, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
+	require.NoError(t, err)
+	recvCh := make(chan []byte, 10)
+	closeCh := make(chan struct{}, 10)
+	go func() {
+		_, data, err := c3.ReadMessage()
+		require.NoError(t, err)
+		recvCh <- data
+	}()
+	go func() {
+		for {
+			select {
+			case <-closeCh:
+				return
+			default:
+				timex.Add(time.Second)
+				time.Sleep(100 * time.Millisecond)
+			}
+		}
+	}()
+	time.Sleep(10 * time.Millisecond)
+	require.NoError(t, TrialManager.StartRule(id))
+	require.Equal(t, "{\"humidity\":50,\"temperature\":22.5}", string(<-recvCh))
+	TrialManager.StopRule(id)
+	closeCh <- struct{}{}
+	c3.Close()
+}
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"