Repository: lf-edge/ekuiper
Pull request #1183 (2022-02-22 09:19:40)

refactor(connection): refactor mqtt/edgex connection share
Signed-off-by: Jianxiang Ran <rxan_embedded@163.com>


Patch:
diff --git a/docs/en_US/rules/sources/mqtt.md b/docs/en_US/rules/sources/mqtt.md
index b804ae13..49474d10 100644
--- a/docs/en_US/rules/sources/mqtt.md
+++ b/docs/en_US/rules/sources/mqtt.md
@@ -148,14 +148,48 @@ Expected field type.
 
 ## Override the default settings
 
-If you have a specific connection that need to overwrite the default settings, you can create a customized section. In the previous sample, we create a specific setting named with `demo`.  Then you can specify the configuration with option `CONF_KEY` when creating the stream definition (see [stream specs](../../sqls/streams.md) for more info).
+If you have a specific connection that need to overwrite the default settings, you can create a customized section. In the previous sample, we create a specific setting named with `demo_conf`.  Then you can specify the configuration with option `CONF_KEY` when creating the stream definition (see [stream specs](../../sqls/streams.md) for more info).
 
 **Sample**
 
 ```
 demo (
 		...
-	) WITH (DATASOURCE="test/", FORMAT="JSON", KEY="USERID", CONF_KEY="demo");
+	) WITH (DATASOURCE="test/", FORMAT="JSON", KEY="USERID", CONF_KEY="demo_conf");
 ```
 
 The configuration keys used for these specific settings are the same as in `default` settings, any values specified in specific settings will overwrite the values in `default` section.
+
+## Use same connection selector in multiple customized config section
+
+If user creates two config sections, for example
+
+```yaml
+#Override the global configurations
+demo_conf: #Conf_key
+  qos: 0
+  connectionSelector: mqtt.localConnection 
+  servers: [tcp://10.211.55.6:1883, tcp://127.0.0.1]
+
+#Override the global configurations
+demo2_conf: #Conf_key
+  qos: 0
+  connentionSelector: mqtt.localConnection
+  servers: [tcp://10.211.55.6:1883, tcp://127.0.0.1]
+```
+
+create two streams using the config defined above
+```
+demo (
+		...
+	) WITH (DATASOURCE="test/", FORMAT="JSON", CONF_KEY="demo_conf");
+
+demo2 (
+		...
+	) WITH (DATASOURCE="test2/", FORMAT="JSON", CONF_KEY="demo2_conf");
+
+```
+When create rules using the defined streams, the rules will share the same connection in source part. 
+The `DATASOURCE` here will be used as mqtt subscription topics, and subscription  `Qos` defined in config section.
+So stream `demo` will subscribe to topic `test/` with Qos 0 and stream `demo2` will subscribe to topic `test2/` with Qos 0 in this example.
+But if  `DATASOURCE` is same and `qos` not, will only subscribe one time when the first rule starts.       
diff --git a/docs/zh_CN/rules/sources/mqtt.md b/docs/zh_CN/rules/sources/mqtt.md
index 20e31f38..54f5d64c 100755
--- a/docs/zh_CN/rules/sources/mqtt.md
+++ b/docs/zh_CN/rules/sources/mqtt.md
@@ -155,3 +155,36 @@ demo (
 
 这些特定设置使用的配置键与 `default` 设置中的配置键相同，在特定设置中指定的任何值都将覆盖 `default` 部分中的值。
 
+## 在多个配置块中引用同一个 connectionSelector
+
+如下所示，用户创建了两个配置项
+```yaml
+#Override the global configurations
+demo_conf: #Conf_key
+  qos: 0
+  connectionSelector: mqtt.localConnection 
+  servers: [tcp://10.211.55.6:1883, tcp://127.0.0.1]
+
+#Override the global configurations
+demo2_conf: #Conf_key
+  qos: 0
+  connentionSelector: mqtt.localConnection
+  servers: [tcp://10.211.55.6:1883, tcp://127.0.0.1]
+```
+基于以上配置，创建了两个数据流
+
+```
+demo (
+		...
+	) WITH (DATASOURCE="test/", FORMAT="JSON", CONF_KEY="demo_conf");
+
+demo2 (
+		...
+	) WITH (DATASOURCE="test2/", FORMAT="JSON", CONF_KEY="demo2_conf");
+
+```
+
+当相应的规则分别引用以上数据流时，规则之间的源部分将共享连接。
+在这里 `DATASOURCE` 对应 mqtt 订阅的 topic, 配置项中的 `qos` 将用作订阅时的 `Qos`.
+在这个例子中，`demo` 以 Qos 0 订阅 topic `test/`,`demo2` 以 Qos 0 订阅 topic `test2/`
+值得注意的是如果两个规则订阅的 `topic` 完全一样而 `Qos` 不同，那么只会订阅一次并以首先启动的规则订阅为准。
\ No newline at end of file
diff --git a/etc/connections/connection.yaml b/etc/connections/connection.yaml
index 58e9d4b8..effad8bd 100644
--- a/etc/connections/connection.yaml
+++ b/etc/connections/connection.yaml
@@ -8,7 +8,6 @@ mqtt:
     #rootCaPath: /var/kuiper/xyz-rootca.pem
     #insecureSkipVerify: false
     #protocolVersion: 3
-    clientid: ekuiper
   cloudConnection: #connection key
     servers: ["tcp://broker.emqx.io:1883"]
     username: user1
diff --git a/etc/sinks/edgex.json b/etc/sinks/edgex.json
index b04abd22..6ce755dc 100644
--- a/etc/sinks/edgex.json
+++ b/etc/sinks/edgex.json
@@ -54,7 +54,7 @@
       }
     },
     {
-      "name": "host",
+      "name": "server",
       "optional": true,
       "control": "text",
       "default": "localhost",
@@ -111,8 +111,8 @@
         "zh_CN": " 发布的主题的前缀。发送的主题将采用动态拼接，格式为`$topicPrefix/$profileName/$deviceName/$sourceName` 。"
       },
       "label": {
-        "en_US": "Topic",
-        "zh_CN": "主题"
+        "en_US": "Topic Prefix",
+        "zh_CN": "主题前缀"
       }
     },
     {
diff --git a/internal/conf/load.go b/internal/conf/load.go
index aa83e54c..498c0d8c 100644
--- a/internal/conf/load.go
+++ b/internal/conf/load.go
@@ -291,3 +291,19 @@ func extractNamesFromElement(jsonMap map[string]interface{}) []string {
 
 	return result
 }
+
+func Printable(m map[string]interface{}) map[string]interface{} {
+	printableMap := make(map[string]interface{})
+	for k, v := range m {
+		if strings.ToLower(k) == "password" {
+			printableMap[k] = "***"
+		} else {
+			if vm, ok := v.(map[string]interface{}); ok {
+				printableMap[k] = Printable(vm)
+			} else {
+				printableMap[k] = v
+			}
+		}
+	}
+	return printableMap
+}
diff --git a/internal/server/server.go b/internal/server/server.go
index c8c0360a..9724f3e1 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -26,6 +26,7 @@ import (
 	"github.com/lf-edge/ekuiper/internal/plugin/portable/runtime"
 	"github.com/lf-edge/ekuiper/internal/processor"
 	"github.com/lf-edge/ekuiper/internal/service"
+	"github.com/lf-edge/ekuiper/internal/topo/connection/factory"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 
 	"context"
@@ -51,6 +52,7 @@ func StartUp(Version, LoadFileType string) {
 	conf.LoadFileType = LoadFileType
 	startTimeStamp = time.Now().Unix()
 	conf.InitConf()
+	factory.InitClientsFactory()
 
 	err := store.SetupWithKuiperConfig(conf.Config)
 	if err != nil {
diff --git a/internal/topo/connection/client_edgex.go b/internal/topo/connection/client_edgex.go
deleted file mode 100644
index 6d8201b5..00000000
--- a/internal/topo/connection/client_edgex.go
+++ /dev/null
@@ -1,108 +0,0 @@
-// Copyright 2021 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
-// +build edgex
-
-package connection
-
-import (
-	"fmt"
-	"github.com/edgexfoundry/go-mod-messaging/v2/messaging"
-	"github.com/edgexfoundry/go-mod-messaging/v2/pkg/types"
-	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/pkg/cast"
-)
-
-func init() {
-	registerClientFactory("edgex", func(s *conf.ConSelector) Client {
-		return &EdgexClient{selector: s}
-	})
-}
-
-type EdgexClient struct {
-	selector *conf.ConSelector
-	mbconf   types.MessageBusConfig
-	client   messaging.MessageClient
-}
-
-type EdgexConf struct {
-	Protocol string            `json:"protocol"`
-	Server   string            `json:"server"`
-	Port     int               `json:"port"`
-	Type     string            `json:"type"`
-	Optional map[string]string `json:"optional"`
-}
-
-func (es *EdgexClient) CfgValidate(props map[string]interface{}) error {
-	c := &EdgexConf{}
-	err := cast.MapToStructStrict(props, c)
-	if err != nil {
-		return fmt.Errorf("map config map to struct %v fail for connection selector %s with error: %v", props, es.selector.ConnSelectorStr, err)
-	}
-
-	if c.Server == "" {
-		return fmt.Errorf("missing server property for connection selector %s", es.selector.ConnSelectorStr)
-	}
-
-	if c.Port == 0 {
-		return fmt.Errorf("missing port property for connection selector %s", es.selector.ConnSelectorStr)
-	}
-
-	if c.Type != messaging.ZeroMQ && c.Type != messaging.MQTT && c.Type != messaging.Redis {
-		return fmt.Errorf("specified wrong type value %s for connection selector %s", c.Type, es.selector.ConnSelectorStr)
-	}
-
-	mbconf := types.MessageBusConfig{
-		SubscribeHost: types.HostInfo{
-			Protocol: c.Protocol,
-			Host:     c.Server,
-			Port:     c.Port,
-		},
-		PublishHost: types.HostInfo{
-			Host:     c.Server,
-			Port:     c.Port,
-			Protocol: c.Protocol,
-		},
-		Type: c.Type}
-	mbconf.Optional = c.Optional
-	es.mbconf = mbconf
-
-	return nil
-}
-
-func (es *EdgexClient) GetClient() (interface{}, error) {
-
-	client, err := messaging.NewMessageClient(es.mbconf)
-	if err != nil {
-		return nil, err
-	}
-
-	if err := client.Connect(); err != nil {
-		conf.Log.Errorf("The connection to edgex messagebus failed for connection selector : %s.", es.selector.ConnSelectorStr)
-		return nil, fmt.Errorf("Failed to connect to edgex message bus: " + err.Error())
-	}
-	conf.Log.Infof("The connection to edgex messagebus is established successfully for connection selector : %s.", es.selector.ConnSelectorStr)
-
-	es.client = client
-	return client, nil
-}
-
-func (es *EdgexClient) CloseClient() error {
-	conf.Log.Infof("Closing the connection to edgex messagebus for connection selector : %s.", es.selector.ConnSelectorStr)
-	if e := es.client.Disconnect(); e != nil {
-		return e
-	}
-	return nil
-}
diff --git a/internal/topo/connection/client_mqtt.go b/internal/topo/connection/client_mqtt.go
deleted file mode 100644
index 831fd9a9..00000000
--- a/internal/topo/connection/client_mqtt.go
+++ /dev/null
@@ -1,126 +0,0 @@
-package connection
-
-import (
-	"crypto/tls"
-	"fmt"
-	MQTT "github.com/eclipse/paho.mqtt.golang"
-	"github.com/google/uuid"
-	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/pkg/cert"
-	"github.com/lf-edge/ekuiper/pkg/cast"
-	"strings"
-)
-
-func init() {
-	registerClientFactory("mqtt", func(s *conf.ConSelector) Client {
-		return &MQTTClient{selector: s}
-	})
-}
-
-type MQTTConnectionConfig struct {
-	Servers            []string `json:"servers"`
-	PVersion           string   `json:"protocolVersion"`
-	ClientId           string   `json:"clientid"`
-	Uname              string   `json:"username"`
-	Password           string   `json:"password"`
-	Certification      string   `json:"certificationPath"`
-	PrivateKPath       string   `json:"privateKeyPath"`
-	RootCaPath         string   `json:"rootCaPath"`
-	InsecureSkipVerify bool     `json:"insecureSkipVerify"`
-}
-
-type MQTTClient struct {
-	srv      string
-	clientid string
-	pVersion uint
-	uName    string
-	password string
-	tls      *tls.Config
-
-	selector *conf.ConSelector
-	conn     MQTT.Client
-}
-
-func (ms *MQTTClient) CfgValidate(props map[string]interface{}) error {
-
-	cfg := MQTTConnectionConfig{}
-
-	err := cast.MapToStructStrict(props, &cfg)
-	if err != nil {
-		return fmt.Errorf("failed to get config for %s, the error is %s", ms.selector.ConnSelectorStr, err)
-	}
-
-	if srvs := cfg.Servers; srvs != nil && len(srvs) > 0 {
-		ms.srv = srvs[0]
-	} else {
-		return fmt.Errorf("missing server property for %s", ms.selector.ConnSelectorStr)
-	}
-
-	if cfg.ClientId == "" {
-		if newUUID, err := uuid.NewUUID(); err != nil {
-			return fmt.Errorf("failed to get uuid for %s, the error is %s", ms.selector.ConnSelectorStr, err)
-		} else {
-			ms.clientid = newUUID.String()
-		}
-	} else {
-		ms.clientid = cfg.ClientId
-	}
-
-	ms.pVersion = 3
-	if cfg.PVersion == "3.1.1" {
-		ms.pVersion = 4
-	}
-
-	tlsOpts := cert.TlsConfigurationOptions{
-		SkipCertVerify: cfg.InsecureSkipVerify,
-		CertFile:       cfg.Certification,
-		KeyFile:        cfg.PrivateKPath,
-		CaFile:         cfg.RootCaPath,
-	}
-	conf.Log.Infof("Connect MQTT broker with TLS configs: %v for connection selector: %s.", tlsOpts, ms.selector.ConnSelectorStr)
-	tlscfg, err := cert.GenerateTLSForClient(tlsOpts)
-	if err != nil {
-		return err
-	}
-
-	ms.tls = tlscfg
-
-	ms.uName = cfg.Uname
-	ms.password = strings.Trim(cfg.Password, " ")
-
-	return nil
-}
-
-func (ms *MQTTClient) GetClient() (interface{}, error) {
-
-	opts := MQTT.NewClientOptions().AddBroker(ms.srv).SetProtocolVersion(ms.pVersion).SetCleanSession(false)
-
-	opts = opts.SetTLSConfig(ms.tls)
-
-	if ms.uName != "" {
-		opts = opts.SetUsername(ms.uName)
-	}
-	if ms.password != "" {
-		opts = opts.SetPassword(ms.password)
-	}
-	opts = opts.SetClientID(ms.clientid)
-	opts = opts.SetAutoReconnect(true)
-
-	c := MQTT.NewClient(opts)
-	if token := c.Connect(); token.Wait() && token.Error() != nil {
-		conf.Log.Errorf("The connection to mqtt broker failed for connection selector: %s ", ms.selector.ConnSelectorStr)
-		return nil, fmt.Errorf("found error when connecting for connection selector %s: %s", ms.selector.ConnSelectorStr, token.Error())
-	}
-	conf.Log.Infof("The connection to mqtt broker is established successfully for connection selector: %s.", ms.selector.ConnSelectorStr)
-
-	ms.conn = c
-	return c, nil
-}
-
-func (ms *MQTTClient) CloseClient() error {
-	conf.Log.Infof("Closing the connection to mqtt broker for connection selector: %s", ms.selector.ConnSelectorStr)
-	if ms.conn != nil && ms.conn.IsConnected() {
-		ms.conn.Disconnect(5000)
-	}
-	return nil
-}
diff --git a/internal/topo/connection/client_wrapper.go b/internal/topo/connection/client_wrapper.go
deleted file mode 100644
index d161b657..00000000
--- a/internal/topo/connection/client_wrapper.go
+++ /dev/null
@@ -1,64 +0,0 @@
-package connection
-
-import (
-	"github.com/lf-edge/ekuiper/internal/conf"
-)
-
-type ClientFactoryFunc func(super *conf.ConSelector) Client
-
-type Client interface {
-	CfgValidate(map[string]interface{}) error
-	GetClient() (interface{}, error)
-	CloseClient() error
-}
-
-type clientWrapper struct {
-	cli    Client
-	conn   interface{}
-	refCnt uint32
-}
-
-func NewClientWrapper(client Client, selector *conf.ConSelector) (*clientWrapper, error) {
-	props, err := selector.ReadCfgFromYaml()
-	if err != nil {
-		return nil, err
-	}
-	err = client.CfgValidate(props)
-	if err != nil {
-		return nil, err
-	}
-	var con interface{}
-
-	con, err = client.GetClient()
-	if err != nil {
-		return nil, err
-	}
-
-	cliWpr := &clientWrapper{
-		cli:    client,
-		conn:   con,
-		refCnt: 1,
-	}
-
-	return cliWpr, nil
-}
-
-func (c *clientWrapper) addRef() {
-	c.refCnt = c.refCnt + 1
-}
-
-func (c *clientWrapper) subRef() {
-	c.refCnt = c.refCnt - 1
-}
-
-func (c *clientWrapper) IsRefEmpty() bool {
-	return c.refCnt == 0
-}
-
-func (c *clientWrapper) clean() {
-	_ = c.cli.CloseClient()
-}
-
-func (c *clientWrapper) getInstance() interface{} {
-	return c.conn
-}
diff --git a/internal/topo/connection/clients/client_registy.go b/internal/topo/connection/clients/client_registy.go
new file mode 100644
index 00000000..738c02c0
--- /dev/null
+++ b/internal/topo/connection/clients/client_registy.go
@@ -0,0 +1,109 @@
+// Copyright 2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package clients
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"strings"
+	"sync"
+)
+
+type clientRegistry struct {
+	Lock                sync.Mutex
+	supportedClientType []string
+	clientFactory       map[string]ClientFactoryFunc
+	ShareClientStore    map[string]ClientWrapper
+}
+
+var ClientRegistry = clientRegistry{
+	clientFactory:       make(map[string]ClientFactoryFunc),
+	Lock:                sync.Mutex{},
+	supportedClientType: make([]string, 0),
+	ShareClientStore:    make(map[string]ClientWrapper),
+}
+
+func RegisterClientFactory(clientType string, creatorFunc ClientFactoryFunc) {
+	ClientRegistry.Lock.Lock()
+	ClientRegistry.clientFactory[clientType] = creatorFunc
+	ClientRegistry.supportedClientType = append(ClientRegistry.supportedClientType, clientType)
+	ClientRegistry.Lock.Unlock()
+}
+
+func getConnectionSelector(props map[string]interface{}) (ConnectionSelector string, err error) {
+	for key, v := range props {
+		if strings.EqualFold(key, "connectionSelector") {
+			if conVal, ok := v.(string); ok {
+				return strings.ToLower(conVal), nil
+			} else {
+				return "", fmt.Errorf("connectionSelector value: %v is not string", v)
+			}
+		}
+	}
+	return "", nil
+}
+
+func GetClient(connectionType string, props map[string]interface{}) (api.MessageClient, error) {
+	ClientRegistry.Lock.Lock()
+	defer ClientRegistry.Lock.Unlock()
+
+	connectSelector, err := getConnectionSelector(props)
+	if err != nil {
+		return nil, err
+	}
+	if connectSelector != "" {
+		if cliWpr, found := ClientRegistry.ShareClientStore[connectSelector]; found {
+			cliWpr.AddRef()
+			return cliWpr, nil
+		}
+	}
+
+	clientCreator, ok := ClientRegistry.clientFactory[connectionType]
+	if !ok {
+		conf.Log.Errorf("can not find clientCreator for connection type : %s. only support %s", connectionType, ClientRegistry.supportedClientType)
+		return nil, fmt.Errorf("can not find clientCreator for connection type : %s. only support %s", connectionType, ClientRegistry.supportedClientType)
+	}
+
+	if connectSelector != "" {
+		selectCfg := &conf.ConSelector{
+			ConnSelectorStr: connectSelector,
+		}
+		if err := selectCfg.Init(); err != nil {
+			return nil, err
+		}
+		cf, err := selectCfg.ReadCfgFromYaml()
+		if err != nil {
+			return nil, err
+		}
+		cliWpr, err := clientCreator(cf)
+		if err != nil {
+			conf.Log.Errorf("can not create client for connection selector : %s have error %s", connectSelector, err)
+			return nil, err
+		}
+		cliWpr.SetConnectionSelector(connectSelector)
+		conf.Log.Infof("Init client wrapper for client type %s and connection selector %s", connectionType, connectSelector)
+		ClientRegistry.ShareClientStore[connectSelector] = cliWpr
+		return cliWpr, nil
+	} else {
+		cliWpr, err := clientCreator(props)
+		if err != nil {
+			conf.Log.Errorf("can not create client for cfg : %v have error %s", props, err)
+			return nil, err
+		}
+		conf.Log.Infof("Init client wrapper for client type %s", connectionType)
+		return cliWpr, nil
+	}
+}
diff --git a/internal/topo/connection/clients/edgex/edgex.go b/internal/topo/connection/clients/edgex/edgex.go
new file mode 100644
index 00000000..0024a92f
--- /dev/null
+++ b/internal/topo/connection/clients/edgex/edgex.go
@@ -0,0 +1,159 @@
+// Copyright 2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build edgex
+// +build edgex
+
+package edgex
+
+import (
+	"fmt"
+	"github.com/edgexfoundry/go-mod-messaging/v2/messaging"
+	"github.com/edgexfoundry/go-mod-messaging/v2/pkg/types"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/pkg/cast"
+	"strings"
+)
+
+type EdgexClient struct {
+	mbconf types.MessageBusConfig
+	client messaging.MessageClient
+}
+
+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) {
+	var 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 *EdgexClient) CfgValidate(props map[string]interface{}) error {
+	edgeAddr := "localhost"
+	c := &EdgexConf{
+		Protocol: "redis",
+		Port:     6379,
+		Type:     messaging.Redis,
+		Optional: nil,
+	}
+
+	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.ZeroMQ && c.Type != messaging.MQTT && c.Type != messaging.Redis {
+		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{
+		SubscribeHost: types.HostInfo{
+			Host:     edgeAddr,
+			Port:     c.Port,
+			Protocol: c.Protocol,
+		},
+		PublishHost: 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 *EdgexClient) Connect() error {
+	client, err := messaging.NewMessageClient(es.mbconf)
+	if err != nil {
+		return err
+	}
+
+	if err := client.Connect(); err != nil {
+		conf.Log.Errorf("The connection to edgex messagebus failed.")
+		return fmt.Errorf("Failed to connect to edgex message bus: " + err.Error())
+	}
+	es.client = client
+	return nil
+}
+
+func (es *EdgexClient) 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 *EdgexClient) 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 *EdgexClient) GetClient() (interface{}, error) {
+
+	client, err := messaging.NewMessageClient(es.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.")
+
+	es.client = client
+	return client, nil
+}
+
+func (es *EdgexClient) 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/topo/connection/clients/edgex/edgex_wrapper.go b/internal/topo/connection/clients/edgex/edgex_wrapper.go
new file mode 100644
index 00000000..21603b77
--- /dev/null
+++ b/internal/topo/connection/clients/edgex/edgex_wrapper.go
@@ -0,0 +1,275 @@
+// Copyright 2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build edgex
+// +build edgex
+
+package edgex
+
+import (
+	"fmt"
+	"github.com/edgexfoundry/go-mod-messaging/v2/pkg/types"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/topo/connection/clients"
+	defaultCtx "github.com/lf-edge/ekuiper/internal/topo/context"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"strings"
+	"sync"
+)
+
+type edgexCtxKey int
+
+const (
+	_ edgexCtxKey = iota
+	ctxKeyEdgeXRequest
+)
+
+type RequestInfo struct {
+	ContentType string
+}
+
+// WithRequestInfo creates a new context that has MqttRequestInfo injected.
+func WithRequestInfo(parent *defaultCtx.DefaultContext, reqInfo *RequestInfo) *defaultCtx.DefaultContext {
+	return defaultCtx.WithValue(parent, ctxKeyEdgeXRequest, reqInfo)
+}
+
+// GetRequestInfo tries to retrieve MqttRequestInfo from the given context.
+// If it doesn't exist, an nil is returned.
+func GetRequestInfo(parent *defaultCtx.DefaultContext) *RequestInfo {
+	if reqInfo := parent.Value(ctxKeyEdgeXRequest); reqInfo != nil {
+		return reqInfo.(*RequestInfo)
+	}
+	return nil
+}
+
+type messageHandler func(stopChan chan struct{}, msgChan chan types.MessageEnvelope)
+
+type edgexSubscriptionInfo struct {
+	topic          string
+	handler        messageHandler
+	stop           chan struct{}
+	topicConsumers []*clients.ConsumerInfo
+}
+
+type edgexClientWrapper struct {
+	cli *EdgexClient
+
+	subLock sync.RWMutex
+	//topic: subscriber
+	//multiple go routine can sub same topic
+	topicSubscriptions map[string]*edgexSubscriptionInfo
+
+	//consumerId: SubscribedTopics
+	subscribers map[string]clients.SubscribedTopics
+
+	conSelector string
+
+	refLock sync.RWMutex
+	refCnt  uint64
+}
+
+func NewEdgeClientWrapper(props map[string]interface{}) (clients.ClientWrapper, error) {
+	if props == nil {
+		conf.Log.Warnf("props is nill for mqtt client wrapper")
+	}
+	client := &EdgexClient{}
+	err := client.CfgValidate(props)
+	if err != nil {
+		return nil, err
+	}
+
+	cliWpr := &edgexClientWrapper{
+		cli:                client,
+		subLock:            sync.RWMutex{},
+		topicSubscriptions: make(map[string]*edgexSubscriptionInfo),
+		subscribers:        make(map[string]clients.SubscribedTopics),
+		refCnt:             1,
+	}
+
+	err = client.Connect()
+	if err != nil {
+		return nil, err
+	}
+	return cliWpr, nil
+}
+
+func (mc *edgexClientWrapper) Publish(c api.StreamContext, topic string, message []byte) error {
+	env := types.NewMessageEnvelope(message, c)
+	reqInfo := GetRequestInfo(c.(*defaultCtx.DefaultContext))
+	if reqInfo == nil {
+		return fmt.Errorf("not find reqInfo for mqtt subscription %s_%s_%d", c.GetRuleId(), c.GetOpId(), c.GetInstanceId())
+	}
+	env.ContentType = reqInfo.ContentType
+
+	err := mc.cli.Publish(env, topic)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (mc *edgexClientWrapper) messageHandler(topic string, sub *edgexSubscriptionInfo) func(stopChan chan struct{}, msgChan chan types.MessageEnvelope) {
+	return func(stopChan chan struct{}, msgChan chan types.MessageEnvelope) {
+		for {
+			select {
+			case <-stopChan:
+				conf.Log.Infof("message handler for topic %s stopped", topic)
+				return
+			case msg, ok := <-msgChan:
+				if !ok {
+					for _, consumer := range sub.topicConsumers {
+						close(consumer.ConsumerChan)
+					}
+					return
+				}
+				//broadcast to all topic subscribers
+				if sub != nil {
+					for _, consumer := range sub.topicConsumers {
+						select {
+						case consumer.ConsumerChan <- &msg:
+							break
+						default:
+							conf.Log.Warnf("consumer chan full for request id %s", consumer.ConsumerId)
+						}
+					}
+				}
+			}
+		}
+	}
+}
+
+func (mc *edgexClientWrapper) Subscribe(c api.StreamContext, subChan []api.TopicChannel, messageErrors chan error) error {
+	log := c.GetLogger()
+
+	mc.subLock.Lock()
+	defer mc.subLock.Unlock()
+
+	subId := fmt.Sprintf("%s_%s_%d", c.GetRuleId(), c.GetOpId(), c.GetInstanceId())
+	if _, ok := mc.subscribers[subId]; ok {
+		return fmt.Errorf("already have subscription %s", subId)
+	}
+
+	subTopics := clients.SubscribedTopics{
+		Topics: make([]string, 0),
+	}
+	for _, tpChan := range subChan {
+		tpc := tpChan.Topic
+		subTopics.Topics = append(subTopics.Topics, tpc)
+		sub, found := mc.topicSubscriptions[tpc]
+		if found {
+			sub.topicConsumers = append(sub.topicConsumers, &clients.ConsumerInfo{
+				ConsumerId:   subId,
+				ConsumerChan: tpChan.Messages,
+				SubErrors:    messageErrors,
+			})
+			log.Infof("subscription for topic %s already exists, reqId is %s, total subs %d", tpc, subId, len(sub.topicConsumers))
+		} else {
+			sub := &edgexSubscriptionInfo{
+				topic: tpc,
+				stop:  make(chan struct{}, 1),
+				topicConsumers: []*clients.ConsumerInfo{
+					{
+						ConsumerId:   subId,
+						ConsumerChan: tpChan.Messages,
+						SubErrors:    messageErrors,
+					},
+				},
+			}
+			log.Infof("new subscription for topic %s, reqId is %s", tpc, subId)
+			message := make(chan types.MessageEnvelope)
+			errChan := make(chan error)
+
+			if err := mc.cli.Subscribe(message, tpc, errChan); err != nil {
+				return err
+			}
+			sub.handler = mc.messageHandler(tpc, sub)
+			go sub.handler(sub.stop, message)
+
+			mc.topicSubscriptions[tpc] = sub
+		}
+	}
+	mc.subscribers[subId] = subTopics
+
+	return nil
+}
+
+func (mc *edgexClientWrapper) unsubscribe(c api.StreamContext) {
+	log := c.GetLogger()
+	mc.subLock.Lock()
+	defer mc.subLock.Unlock()
+
+	subId := fmt.Sprintf("%s_%s_%d", c.GetRuleId(), c.GetOpId(), c.GetInstanceId())
+	subTopics, found := mc.subscribers[subId]
+	if !found {
+		log.Errorf("not found subscription id %s", subId)
+		return
+	}
+
+	for _, tpc := range subTopics.Topics {
+		if sub, found := mc.topicSubscriptions[tpc]; found {
+			for index, consumer := range sub.topicConsumers {
+				if strings.EqualFold(subId, consumer.ConsumerId) {
+					close(consumer.ConsumerChan)
+					sub.topicConsumers = append(sub.topicConsumers[:index], sub.topicConsumers[index+1:]...)
+					log.Infof("unsubscription topic %s for reqId %s, total subs %d", tpc, subId, len(sub.topicConsumers))
+				}
+			}
+			if 0 == len(sub.topicConsumers) {
+				delete(mc.topicSubscriptions, tpc)
+				sub.stop <- struct{}{}
+			}
+		}
+	}
+	delete(mc.subscribers, subId)
+}
+
+func (mc *edgexClientWrapper) SetConnectionSelector(conSelector string) {
+	mc.conSelector = conSelector
+}
+
+func (mc *edgexClientWrapper) Release(c api.StreamContext) {
+	mc.unsubscribe(c)
+
+	clients.ClientRegistry.Lock.Lock()
+	mc.DeRef(c)
+	clients.ClientRegistry.Lock.Unlock()
+}
+
+func (mc *edgexClientWrapper) AddRef() {
+	mc.refLock.Lock()
+	defer mc.refLock.Unlock()
+
+	mc.refCnt = mc.refCnt + 1
+	conf.Log.Infof("edgex client wrapper add refence for connection selector %s total refcount %d", mc.conSelector, mc.refCnt)
+}
+
+func (mc *edgexClientWrapper) DeRef(c api.StreamContext) {
+	log := c.GetLogger()
+	mc.refLock.Lock()
+	defer mc.refLock.Unlock()
+
+	mc.refCnt = mc.refCnt - 1
+	if mc.refCnt != 0 {
+		conf.Log.Infof("edgex client wrapper derefence for connection selector %s total refcount %d", mc.conSelector, mc.refCnt)
+	}
+	if mc.refCnt == 0 {
+		log.Infof("mqtt client wrapper reference count 0")
+		if mc.conSelector != "" {
+			conf.Log.Infof("remove mqtt client wrapper for connection selector %s", mc.conSelector)
+			delete(clients.ClientRegistry.ShareClientStore, mc.conSelector)
+		}
+		_ = mc.cli.Disconnect()
+	}
+
+}
diff --git a/internal/topo/connection/clients/mqtt/mqtt.go b/internal/topo/connection/clients/mqtt/mqtt.go
new file mode 100644
index 00000000..6da26bcd
--- /dev/null
+++ b/internal/topo/connection/clients/mqtt/mqtt.go
@@ -0,0 +1,155 @@
+// Copyright 2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mqtt
+
+import (
+	"crypto/tls"
+	"fmt"
+	MQTT "github.com/eclipse/paho.mqtt.golang"
+	"github.com/google/uuid"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/pkg/cert"
+	"github.com/lf-edge/ekuiper/pkg/cast"
+	"github.com/lf-edge/ekuiper/pkg/errorx"
+	"strings"
+)
+
+type MQTTConnectionConfig struct {
+	Servers            []string `json:"servers"`
+	Server             string   `json:"server"`
+	PVersion           string   `json:"protocolVersion"`
+	ClientId           string   `json:"clientid"`
+	Uname              string   `json:"username"`
+	Password           string   `json:"password"`
+	Certification      string   `json:"certificationPath"`
+	PrivateKPath       string   `json:"privateKeyPath"`
+	RootCaPath         string   `json:"rootCaPath"`
+	InsecureSkipVerify bool     `json:"insecureSkipVerify"`
+}
+
+type MQTTClient struct {
+	srv      string
+	clientid string
+	pVersion uint
+	uName    string
+	password string
+	tls      *tls.Config
+
+	conn MQTT.Client
+}
+
+func (ms *MQTTClient) CfgValidate(props map[string]interface{}) error {
+
+	cfg := MQTTConnectionConfig{}
+
+	err := cast.MapToStruct(props, &cfg)
+	if err != nil {
+		return fmt.Errorf("failed to get config for %v, the error is %s", cfg, err)
+	}
+
+	if srv := cfg.Servers; len(srv) != 0 {
+		ms.srv = srv[0]
+	} else if cfg.Server != "" {
+		ms.srv = cfg.Server
+	} else {
+		return fmt.Errorf("missing server property for %v", cfg)
+	}
+
+	if cfg.ClientId == "" {
+		if newUUID, err := uuid.NewUUID(); err != nil {
+			return fmt.Errorf("failed to get uuid for %v, the error is %s", cfg, err)
+		} else {
+			ms.clientid = newUUID.String()
+		}
+	} else {
+		ms.clientid = cfg.ClientId
+	}
+
+	ms.pVersion = 3
+	if cfg.PVersion == "3.1.1" {
+		ms.pVersion = 4
+	}
+
+	tlsOpts := cert.TlsConfigurationOptions{
+		SkipCertVerify: cfg.InsecureSkipVerify,
+		CertFile:       cfg.Certification,
+		KeyFile:        cfg.PrivateKPath,
+		CaFile:         cfg.RootCaPath,
+	}
+	conf.Log.Infof("Connect MQTT broker %s with TLS configs: %v.", ms.srv, tlsOpts)
+	tlscfg, err := cert.GenerateTLSForClient(tlsOpts)
+	if err != nil {
+		return err
+	}
+	ms.tls = tlscfg
+	ms.uName = cfg.Uname
+	ms.password = strings.Trim(cfg.Password, " ")
+
+	return nil
+}
+
+func (ms *MQTTClient) onConnectLost(_ MQTT.Client, err error) {
+	conf.Log.Warnf("The connection to mqtt broker %s client id %s disconnected with error: %s ", ms.srv, ms.clientid, err.Error())
+}
+
+func (ms *MQTTClient) Connect(handler MQTT.OnConnectHandler) error {
+
+	opts := MQTT.NewClientOptions().AddBroker(ms.srv).SetProtocolVersion(ms.pVersion).SetCleanSession(false)
+
+	opts = opts.SetTLSConfig(ms.tls)
+
+	if ms.uName != "" {
+		opts = opts.SetUsername(ms.uName)
+	}
+	if ms.password != "" {
+		opts = opts.SetPassword(ms.password)
+	}
+	opts = opts.SetClientID(ms.clientid)
+	opts = opts.SetAutoReconnect(true)
+	opts.OnConnect = handler
+	opts.OnConnectionLost = ms.onConnectLost
+
+	c := MQTT.NewClient(opts)
+	if token := c.Connect(); token.Wait() && token.Error() != nil {
+		conf.Log.Errorf("The connection to mqtt broker %s failed : %s ", ms.srv, token.Error())
+		return fmt.Errorf("found error when connecting for %s: %s", ms.srv, token.Error())
+	}
+	conf.Log.Infof("The connection to mqtt broker is established successfully for %s.", ms.srv)
+
+	ms.conn = c
+	return nil
+}
+
+func (ms *MQTTClient) Subscribe(topic string, qos byte, handler MQTT.MessageHandler) error {
+	if token := ms.conn.Subscribe(topic, qos, handler); token.Wait() && token.Error() != nil {
+		return fmt.Errorf("%s: %s", errorx.IOErr, token.Error())
+	}
+	return nil
+}
+
+func (ms *MQTTClient) Publish(topic string, qos byte, retained bool, message []byte) error {
+	if token := ms.conn.Publish(topic, qos, retained, message); token.Wait() && token.Error() != nil {
+		return fmt.Errorf("%s: %s", errorx.IOErr, token.Error())
+	}
+	return nil
+}
+
+func (ms *MQTTClient) Disconnect() error {
+	conf.Log.Infof("Closing the connection to mqtt broker for %s", ms.srv)
+	if ms.conn != nil && ms.conn.IsConnected() {
+		ms.conn.Disconnect(5000)
+	}
+	return nil
+}
diff --git a/internal/topo/connection/clients/mqtt/mqtt_wrapper.go b/internal/topo/connection/clients/mqtt/mqtt_wrapper.go
new file mode 100644
index 00000000..c65f5635
--- /dev/null
+++ b/internal/topo/connection/clients/mqtt/mqtt_wrapper.go
@@ -0,0 +1,266 @@
+// Copyright 2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mqtt
+
+import (
+	"fmt"
+	pahoMqtt "github.com/eclipse/paho.mqtt.golang"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/topo/connection/clients"
+	defaultCtx "github.com/lf-edge/ekuiper/internal/topo/context"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"strings"
+	"sync"
+)
+
+type mqttCtxKey int
+
+const (
+	_ mqttCtxKey = iota
+	ctxKeyMqttRequest
+)
+
+type RequestInfo struct {
+	Qos      byte
+	Retained bool
+}
+
+// WithRequestInfo creates a new context that has MqttRequestInfo injected.
+func WithRequestInfo(parent *defaultCtx.DefaultContext, reqInfo *RequestInfo) *defaultCtx.DefaultContext {
+	return defaultCtx.WithValue(parent, ctxKeyMqttRequest, reqInfo)
+}
+
+// GetRequestInfo tries to retrieve MqttRequestInfo from the given context.
+// If it doesn't exist, an nil is returned.
+func GetRequestInfo(parent *defaultCtx.DefaultContext) *RequestInfo {
+	if reqInfo := parent.Value(ctxKeyMqttRequest); reqInfo != nil {
+		return reqInfo.(*RequestInfo)
+	}
+	return nil
+}
+
+type mqttSubscriptionInfo struct {
+	topic        string
+	qos          byte
+	topicHandler pahoMqtt.MessageHandler
+
+	topicConsumers []*clients.ConsumerInfo
+}
+
+type mqttClientWrapper struct {
+	cli *MQTTClient
+
+	subLock sync.RWMutex
+	//topic: subscriber
+	//multiple go routine can sub same topic
+	topicSubscriptions map[string]*mqttSubscriptionInfo
+
+	//consumerId: SubscribedTopics
+	subscribers map[string]clients.SubscribedTopics
+
+	conSelector string
+
+	refLock sync.RWMutex
+	refCnt  uint64
+}
+
+func NewMqttClientWrapper(props map[string]interface{}) (clients.ClientWrapper, error) {
+	if props == nil {
+		conf.Log.Warnf("props is nill for mqtt client wrapper")
+	}
+	client := &MQTTClient{}
+	err := client.CfgValidate(props)
+	if err != nil {
+		return nil, err
+	}
+	cliWpr := &mqttClientWrapper{
+		cli:                client,
+		subLock:            sync.RWMutex{},
+		topicSubscriptions: make(map[string]*mqttSubscriptionInfo),
+		subscribers:        make(map[string]clients.SubscribedTopics),
+		refCnt:             1,
+	}
+
+	err = client.Connect(cliWpr.onConnectHandler)
+	if err != nil {
+		return nil, err
+	}
+
+	return cliWpr, nil
+}
+
+func (mc *mqttClientWrapper) onConnectHandler(_ pahoMqtt.Client) {
+	// activeSubscriptions will be empty on the first connection.
+	// On a re-connect is when the subscriptions must be re-created.
+	conf.Log.Infof("The connection to mqtt broker %s client id %s established", mc.cli.srv, mc.cli.clientid)
+	mc.subLock.RLock()
+	defer mc.subLock.RUnlock()
+	for topic, subscription := range mc.topicSubscriptions {
+		token := mc.cli.conn.Subscribe(topic, subscription.qos, subscription.topicHandler)
+		if token.Error() != nil {
+			for _, con := range subscription.topicConsumers {
+				con.SubErrors <- token.Error()
+			}
+		}
+	}
+}
+
+func (mc *mqttClientWrapper) newMessageHandler(sub *mqttSubscriptionInfo) pahoMqtt.MessageHandler {
+	return func(client pahoMqtt.Client, message pahoMqtt.Message) {
+		if sub != nil {
+			for _, consumer := range sub.topicConsumers {
+				select {
+				case consumer.ConsumerChan <- message:
+					break
+				default:
+					conf.Log.Warnf("consumer chan full for request id %s", consumer.ConsumerId)
+				}
+			}
+		}
+	}
+}
+
+func (mc *mqttClientWrapper) Publish(c api.StreamContext, topic string, message []byte) error {
+	reqInfo := GetRequestInfo(c.(*defaultCtx.DefaultContext))
+	if reqInfo == nil {
+		return fmt.Errorf("not find reqInfo for mqtt subscription %s_%s_%d", c.GetRuleId(), c.GetOpId(), c.GetInstanceId())
+	}
+	err := mc.cli.Publish(topic, reqInfo.Qos, reqInfo.Retained, message)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (mc *mqttClientWrapper) Subscribe(c api.StreamContext, subChan []api.TopicChannel, messageErrors chan error) error {
+	log := c.GetLogger()
+
+	mc.subLock.Lock()
+	defer mc.subLock.Unlock()
+
+	subId := fmt.Sprintf("%s_%s_%d", c.GetRuleId(), c.GetOpId(), c.GetInstanceId())
+	if _, ok := mc.subscribers[subId]; ok {
+		return fmt.Errorf("already have subscription %s", subId)
+	}
+	reqInfo := GetRequestInfo(c.(*defaultCtx.DefaultContext))
+	if reqInfo == nil {
+		return fmt.Errorf("not find reqInfo for mqtt subscription %s", subId)
+	}
+
+	subTopics := clients.SubscribedTopics{
+		Topics: make([]string, 0),
+	}
+
+	for _, tpChan := range subChan {
+		tpc := tpChan.Topic
+		subTopics.Topics = append(subTopics.Topics, tpc)
+		sub, found := mc.topicSubscriptions[tpc]
+		if found {
+			sub.topicConsumers = append(sub.topicConsumers, &clients.ConsumerInfo{
+				ConsumerId:   subId,
+				ConsumerChan: tpChan.Messages,
+				SubErrors:    messageErrors,
+			})
+			log.Infof("subscription for topic %s already exists, reqId is %s, total subs %d", tpc, subId, len(sub.topicConsumers))
+		} else {
+			sub := &mqttSubscriptionInfo{
+				topic: tpc,
+				qos:   reqInfo.Qos,
+				topicConsumers: []*clients.ConsumerInfo{
+					{
+						ConsumerId:   subId,
+						ConsumerChan: tpChan.Messages,
+						SubErrors:    messageErrors,
+					},
+				},
+			}
+			sub.topicHandler = mc.newMessageHandler(sub)
+			log.Infof("new subscription for topic %s, reqId is %s", tpc, subId)
+			token := mc.cli.conn.Subscribe(tpc, reqInfo.Qos, sub.topicHandler)
+			if token.Error() != nil {
+				messageErrors <- token.Error()
+				return token.Error()
+			}
+			mc.topicSubscriptions[tpc] = sub
+		}
+	}
+	mc.subscribers[subId] = subTopics
+
+	return nil
+}
+
+func (mc *mqttClientWrapper) unsubscribe(c api.StreamContext) {
+	log := c.GetLogger()
+	mc.subLock.Lock()
+	defer mc.subLock.Unlock()
+
+	subId := fmt.Sprintf("%s_%s_%d", c.GetRuleId(), c.GetOpId(), c.GetInstanceId())
+	subTopics, found := mc.subscribers[subId]
+	if !found {
+		log.Errorf("not found subscription id %s", subId)
+		return
+	}
+
+	for _, tpc := range subTopics.Topics {
+		if sub, found := mc.topicSubscriptions[tpc]; found {
+			for index, consumer := range sub.topicConsumers {
+				if strings.EqualFold(subId, consumer.ConsumerId) {
+					close(consumer.ConsumerChan)
+					sub.topicConsumers = append(sub.topicConsumers[:index], sub.topicConsumers[index+1:]...)
+					log.Infof("unsubscription topic %s for reqId %s, total subs %d", tpc, subId, len(sub.topicConsumers))
+				}
+			}
+			if 0 == len(sub.topicConsumers) {
+				delete(mc.topicSubscriptions, tpc)
+			}
+		}
+	}
+	delete(mc.subscribers, subId)
+}
+
+func (mc *mqttClientWrapper) Release(c api.StreamContext) {
+	mc.unsubscribe(c)
+
+	clients.ClientRegistry.Lock.Lock()
+	mc.DeRef(c)
+	clients.ClientRegistry.Lock.Unlock()
+}
+
+func (mc *mqttClientWrapper) SetConnectionSelector(conSelector string) {
+	mc.conSelector = conSelector
+}
+
+func (mc *mqttClientWrapper) AddRef() {
+	mc.refLock.Lock()
+	defer mc.refLock.Unlock()
+	mc.refCnt = mc.refCnt + 1
+}
+
+func (mc *mqttClientWrapper) DeRef(c api.StreamContext) {
+	log := c.GetLogger()
+	mc.refLock.Lock()
+	defer mc.refLock.Unlock()
+
+	mc.refCnt = mc.refCnt - 1
+	if mc.refCnt == 0 {
+		log.Infof("mqtt client wrapper reference count 0")
+		if mc.conSelector != "" {
+			conf.Log.Infof("remove mqtt client wrapper for connection selector %s", mc.conSelector)
+			delete(clients.ClientRegistry.ShareClientStore, mc.conSelector)
+		}
+		_ = mc.cli.Disconnect()
+	}
+}
diff --git a/internal/topo/connection/clients/type.go b/internal/topo/connection/clients/type.go
new file mode 100644
index 00000000..2dcb5d81
--- /dev/null
+++ b/internal/topo/connection/clients/type.go
@@ -0,0 +1,39 @@
+// Copyright 2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package clients
+
+import (
+	"github.com/lf-edge/ekuiper/pkg/api"
+)
+
+type ConsumerInfo struct {
+	ConsumerId   string
+	ConsumerChan chan<- interface{}
+	SubErrors    chan error
+}
+
+type SubscribedTopics struct {
+	Topics []string
+}
+
+type ClientFactoryFunc func(props map[string]interface{}) (ClientWrapper, error)
+
+type ClientWrapper interface {
+	Subscribe(c api.StreamContext, subChan []api.TopicChannel, messageErrors chan error) error
+	Release(c api.StreamContext)
+	Publish(c api.StreamContext, topic string, message []byte) error
+	SetConnectionSelector(conSelector string)
+	AddRef()
+}
diff --git a/internal/topo/connection/factory/apply.go b/internal/topo/connection/factory/apply.go
new file mode 100644
index 00000000..0551e14b
--- /dev/null
+++ b/internal/topo/connection/factory/apply.go
@@ -0,0 +1,27 @@
+// Copyright 2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package factory
+
+import (
+	"github.com/lf-edge/ekuiper/internal/topo/connection/clients"
+)
+
+var clientsFactory = make(map[string]clients.ClientFactoryFunc)
+
+func InitClientsFactory() {
+	for k, v := range clientsFactory {
+		clients.RegisterClientFactory(k, v)
+	}
+}
diff --git a/internal/topo/connection/factory/builtin.go b/internal/topo/connection/factory/builtin.go
new file mode 100644
index 00000000..e9fd07ba
--- /dev/null
+++ b/internal/topo/connection/factory/builtin.go
@@ -0,0 +1,26 @@
+// Copyright 2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package factory
+
+import (
+	"github.com/lf-edge/ekuiper/internal/topo/connection/clients"
+	"github.com/lf-edge/ekuiper/internal/topo/connection/clients/mqtt"
+)
+
+func init() {
+	clientsFactory["mqtt"] = func(props map[string]interface{}) (clients.ClientWrapper, error) {
+		return mqtt.NewMqttClientWrapper(props)
+	}
+}
diff --git a/internal/topo/connection/factory/ext_edgex.go b/internal/topo/connection/factory/ext_edgex.go
new file mode 100644
index 00000000..c592275a
--- /dev/null
+++ b/internal/topo/connection/factory/ext_edgex.go
@@ -0,0 +1,29 @@
+// Copyright 2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build edgex
+// +build edgex
+
+package factory
+
+import (
+	"github.com/lf-edge/ekuiper/internal/topo/connection/clients"
+	"github.com/lf-edge/ekuiper/internal/topo/connection/clients/edgex"
+)
+
+func init() {
+	clientsFactory["edgex"] = func(props map[string]interface{}) (clients.ClientWrapper, error) {
+		return edgex.NewEdgeClientWrapper(props)
+	}
+}
diff --git a/internal/topo/connection/manager.go b/internal/topo/connection/manager.go
deleted file mode 100644
index 49fd4a5e..00000000
--- a/internal/topo/connection/manager.go
+++ /dev/null
@@ -1,84 +0,0 @@
-package connection
-
-import (
-	"fmt"
-	"github.com/lf-edge/ekuiper/internal/conf"
-	"sync"
-)
-
-var m = clientManager{
-	clientFactory:       make(map[string]ClientFactoryFunc),
-	lock:                sync.Mutex{},
-	supportedClientType: make([]string, 0),
-	clientMap:           make(map[string]*clientWrapper),
-}
-
-type clientManager struct {
-	lock                sync.Mutex
-	supportedClientType []string
-	clientFactory       map[string]ClientFactoryFunc
-	clientMap           map[string]*clientWrapper
-}
-
-func registerClientFactory(clientType string, creatorFunc ClientFactoryFunc) {
-	m.lock.Lock()
-	m.clientFactory[clientType] = creatorFunc
-	m.supportedClientType = append(m.supportedClientType, clientType)
-	m.lock.Unlock()
-}
-
-func GetConnection(connectSelector string) (interface{}, error) {
-
-	m.lock.Lock()
-	defer m.lock.Unlock()
-
-	var cliWpr *clientWrapper
-	var found bool
-
-	cliWpr, found = m.clientMap[connectSelector]
-	if found {
-		cliWpr.addRef()
-	} else {
-		selectCfg := &conf.ConSelector{
-			ConnSelectorStr: connectSelector,
-		}
-		err := selectCfg.Init()
-		if err != nil {
-			conf.Log.Errorf("connection selector: %s have error %s.", connectSelector, err)
-			return nil, err
-		}
-
-		clientCreator, ok := m.clientFactory[selectCfg.Type]
-		if !ok {
-			conf.Log.Errorf("can not find clientCreator for connection selector : %s only support %s", connectSelector, m.supportedClientType)
-			return nil, fmt.Errorf("can not find clientCreator for connection selector : %s. only support %s", connectSelector, m.supportedClientType)
-		}
-
-		client := clientCreator(selectCfg)
-
-		cliWpr, err = NewClientWrapper(client, selectCfg)
-		if err != nil {
-			conf.Log.Errorf("can not create client for connection selector : %s have error %s", connectSelector, err)
-			return nil, err
-		}
-
-		m.clientMap[connectSelector] = cliWpr
-	}
-
-	conf.Log.Infof("connection selector: %s GetConnection count %d.", connectSelector, cliWpr.refCnt)
-
-	return cliWpr.getInstance(), nil
-}
-
-func ReleaseConnection(connectSelector string) {
-	m.lock.Lock()
-	defer m.lock.Unlock()
-	if v, ok := m.clientMap[connectSelector]; ok {
-		v.subRef()
-		conf.Log.Infof("connection selector: %s ReleaseConnection count %d.", connectSelector, v.refCnt)
-		if v.IsRefEmpty() {
-			v.clean()
-			delete(m.clientMap, connectSelector)
-		}
-	}
-}
diff --git a/internal/topo/context/default.go b/internal/topo/context/default.go
index b55cd160..8ef11231 100644
--- a/internal/topo/context/default.go
+++ b/internal/topo/context/default.go
@@ -19,7 +19,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/topo/connection"
+	"github.com/lf-edge/ekuiper/internal/topo/connection/clients"
 	"github.com/lf-edge/ekuiper/internal/topo/transform"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
@@ -252,10 +252,6 @@ func (c *DefaultContext) SaveState(checkpointId int64) error {
 	return nil
 }
 
-func (c *DefaultContext) GetConnection(connectSelector string) (interface{}, error) {
-	return connection.GetConnection(connectSelector)
-}
-
-func (c *DefaultContext) ReleaseConnection(connectSelector string) {
-	connection.ReleaseConnection(connectSelector)
+func (c *DefaultContext) GetClient(clientType string, config map[string]interface{}) (api.MessageClient, error) {
+	return clients.GetClient(clientType, config)
 }
diff --git a/internal/topo/context/func_context.go b/internal/topo/context/func_context.go
index caef4d6d..ff1b050c 100644
--- a/internal/topo/context/func_context.go
+++ b/internal/topo/context/func_context.go
@@ -16,7 +16,7 @@ package context
 
 import (
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/topo/connection"
+	"github.com/lf-edge/ekuiper/internal/topo/connection/clients"
 	"github.com/lf-edge/ekuiper/pkg/api"
 )
 
@@ -56,12 +56,8 @@ func (c *DefaultFuncContext) GetFuncId() int {
 	return c.funcId
 }
 
-func (c *DefaultFuncContext) GetConnection(connectSelector string) (interface{}, error) {
-	return connection.GetConnection(connectSelector)
-}
-
-func (c *DefaultFuncContext) ReleaseConnection(connectSelector string) {
-	connection.ReleaseConnection(connectSelector)
+func (c *DefaultFuncContext) GetClient(clientType string, config map[string]interface{}) (api.MessageClient, error) {
+	return clients.GetClient(clientType, config)
 }
 
 func (c *DefaultFuncContext) convertKey(key string) string {
diff --git a/internal/topo/node/source_node.go b/internal/topo/node/source_node.go
index 74dd12c8..6fa45a72 100644
--- a/internal/topo/node/source_node.go
+++ b/internal/topo/node/source_node.go
@@ -88,7 +88,7 @@ func (m *SourceNode) Open(ctx api.StreamContext, errCh chan<- error) {
 			props["$retainSize"] = m.options.RETAIN_SIZE
 		}
 		m.reset()
-		logger.Infof("open source node with props %v, concurrency: %d, bufferLength: %d", m.props, m.concurrency, m.bufferLength)
+		logger.Infof("open source node with props %v, concurrency: %d, bufferLength: %d", conf.Printable(m.props), m.concurrency, m.bufferLength)
 		for i := 0; i < m.concurrency; i++ { // workers
 			go func(instance int) {
 				//Do open source instances
@@ -110,7 +110,7 @@ func (m *SourceNode) Open(ctx api.StreamContext, errCh chan<- error) {
 
 				defer func() {
 					logger.Infof("source %s done", m.name)
-					m.close(ctx, logger)
+					m.close()
 					buffer.Close()
 				}()
 
@@ -180,14 +180,8 @@ func (m *SourceNode) drainError(errCh chan<- error, err error, ctx api.StreamCon
 	return
 }
 
-func (m *SourceNode) close(ctx api.StreamContext, logger api.Logger) {
-	if !m.options.SHARED {
-		for _, s := range m.sources {
-			if err := s.Close(ctx); err != nil {
-				logger.Warnf("close source fails: %v", err)
-			}
-		}
-	} else {
+func (m *SourceNode) close() {
+	if m.options.SHARED {
 		removeSourceInstance(m)
 	}
 }
diff --git a/internal/topo/node/source_pool.go b/internal/topo/node/source_pool.go
index 9d012137..cd3659c6 100644
--- a/internal/topo/node/source_pool.go
+++ b/internal/topo/node/source_pool.go
@@ -154,7 +154,7 @@ func (p *sourcePool) deleteInstance(k string, node *SourceNode, index int) {
 		end := s.detach(instanceKey)
 		if end {
 			s.cancel()
-			_ = s.source.Close(s.ctx)
+
 			s.dataCh.Close()
 			delete(p.registry, k)
 		}
@@ -300,7 +300,11 @@ func start(poolCtx api.StreamContext, node *SourceNode, s api.Source, instanceIn
 	}
 	chs := newSourceInstanceChannels(node.bufferLength)
 
-	go s.Open(ctx.WithInstance(instanceIndex), chs.dataCh.In, chs.errorCh)
+	go func() {
+		nctx := ctx.WithInstance(instanceIndex)
+		defer s.Close(nctx)
+		s.Open(nctx, chs.dataCh.In, chs.errorCh)
+	}()
 	return &sourceInstance{
 		source:                 s,
 		sourceInstanceChannels: chs,
diff --git a/internal/topo/sink/edgex_sink.go b/internal/topo/sink/edgex_sink.go
index 88f958b2..6ea8dd89 100644
--- a/internal/topo/sink/edgex_sink.go
+++ b/internal/topo/sink/edgex_sink.go
@@ -24,9 +24,9 @@ import (
 	v2 "github.com/edgexfoundry/go-mod-core-contracts/v2/common"
 	"github.com/edgexfoundry/go-mod-core-contracts/v2/dtos"
 	"github.com/edgexfoundry/go-mod-core-contracts/v2/dtos/requests"
-	"github.com/edgexfoundry/go-mod-messaging/v2/messaging"
-	"github.com/edgexfoundry/go-mod-messaging/v2/pkg/types"
 	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/topo/connection/clients/edgex"
+	defaultCtx "github.com/lf-edge/ekuiper/internal/topo/context"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
@@ -41,40 +41,28 @@ const (
 )
 
 type EdgexConf struct {
-	Protocol           string            `json:"protocol"`
-	Host               string            `json:"host"`
-	Port               int               `json:"port"`
-	Topic              string            `json:"topic"`
-	TopicPrefix        string            `json:"topicPrefix"`
-	Type               string            `json:"type"`
-	MessageType        messageType       `json:"messageType"`
-	ContentType        string            `json:"contentType"`
-	DeviceName         string            `json:"deviceName"`
-	ProfileName        string            `json:"profileName"`
-	SourceName         string            `json:"sourceName"`
-	Metadata           string            `json:"metadata"`
-	Optional           map[string]string `json:"optional"`
-	ConnectionSelector string            `json:"connectionSelector"`
-	DataTemplate       string            `json:"dataTemplate"`
+	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"`
 }
 
 type EdgexMsgBusSink struct {
 	c *EdgexConf
 
+	config map[string]interface{}
 	topic  string
-	conf   *types.MessageBusConfig
-	client messaging.MessageClient
+
+	cli api.MessageClient
 }
 
 func (ems *EdgexMsgBusSink) Configure(ps map[string]interface{}) error {
 
-	var (
-		defaultProtocol = "redis"
-		defaultServer   = "localhost"
-		defaultType     = messaging.Redis
-		defaultPort     = 6379
-	)
-
 	c := &EdgexConf{
 		MessageType: MessageTypeEvent,
 		ContentType: "application/json",
@@ -87,47 +75,12 @@ func (ems *EdgexMsgBusSink) Configure(ps map[string]interface{}) error {
 		return fmt.Errorf("read properties %v fail with error: %v", ps, err)
 	}
 
-	if c.ConnectionSelector != "" {
-		conf.Log.Infof("use connection selector %s for edgex sink", c.ConnectionSelector)
-		if c.Protocol != "" || c.Host != "" || c.Type != "" || c.Port != 0 {
-			return fmt.Errorf("connectionSelector can not coexist with other connection configs, properties %v", ps)
-		}
-
-		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)
-		}
-	} else {
-		if c.Host == "" {
-			c.Host = defaultServer
-		}
-		if c.Protocol == "" {
-			c.Protocol = defaultProtocol
-		}
-		if c.Type == "" {
-			c.Type = defaultType
-		}
-		if c.Port == 0 {
-			c.Port = defaultPort
-		}
-		if c.Port < 0 {
-			return fmt.Errorf("specified wrong port value, expect positive integer but got %d", c.Port)
-		}
-
-		if c.Type != messaging.ZeroMQ && c.Type != messaging.MQTT && c.Type != messaging.Redis {
-			return fmt.Errorf("specified wrong type value %s", c.Type)
-		}
-
-		if c.MessageType != MessageTypeEvent && c.MessageType != MessageTypeRequest {
-			return fmt.Errorf("specified wrong messageType value %s", c.MessageType)
-		}
+	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.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 != "" {
@@ -136,52 +89,23 @@ func (ems *EdgexMsgBusSink) Configure(ps map[string]interface{}) error {
 	if c.DataTemplate != "" {
 		conf.Log.Warn("edgex sink does not support dataTemplate, just ignore")
 	}
-
 	ems.c = c
-	ems.conf = &types.MessageBusConfig{
-		PublishHost: types.HostInfo{
-			Host:     c.Host,
-			Port:     c.Port,
-			Protocol: c.Protocol,
-		},
-		Type:     c.Type,
-		Optional: c.Optional,
-	}
-
-	return nil
-}
+	ems.config = ps
 
-func (ems *EdgexMsgBusSink) getClient(ctx api.StreamContext) error {
-	log := ctx.GetLogger()
-	if ems.c.ConnectionSelector != "" {
-		con, err := ctx.GetConnection(ems.c.ConnectionSelector)
-		if err != nil {
-			log.Errorf("The edgex client for connection selector %s get fail with error: %s", ems.c.ConnectionSelector, err)
-			return err
-		}
-		ems.client = con.(messaging.MessageClient)
-		log.Infof("The edge client for connection selector %s get successfully", ems.c.ConnectionSelector)
-	} else {
-		c, err := messaging.NewMessageClient(*ems.conf)
-		if err != nil {
-			return err
-		}
-		ems.client = c
-		if err := ems.client.Connect(); err != nil {
-			return fmt.Errorf("Failed to connect to edgex message bus: " + err.Error())
-		}
-		log.Infof("The connection to edgex messagebus is established successfully.")
-	}
 	return nil
 }
 
 func (ems *EdgexMsgBusSink) Open(ctx api.StreamContext) error {
 	log := ctx.GetLogger()
-	log.Infof("Using configuration for EdgeX message bus sink: %+v", ems.c)
-	if err := ems.getClient(ctx); err != nil {
-		return fmt.Errorf("Failed to get edgex message client: " + err.Error())
+
+	cli, err := ctx.GetClient("edgex", ems.config)
+	if err != nil {
+		log.Errorf("found error when get edgex client, error %s", err.Error())
+		return err
 	}
 
+	ems.cli = cli
+
 	if ems.c.SourceName == "" {
 		ems.c.SourceName = ctx.GetRuleId()
 	}
@@ -570,8 +494,6 @@ func (ems *EdgexMsgBusSink) Collect(ctx api.StreamContext, item interface{}) err
 			return fmt.Errorf("unexpected error MarshalEvent %v", err)
 		}
 	}
-	env := types.NewMessageEnvelope(data, ctx)
-	env.ContentType = ems.c.ContentType
 
 	if ems.topic == "" { // dynamic topic
 		topic = fmt.Sprintf("%s/%s/%s/%s", ems.c.TopicPrefix, evt.ProfileName, evt.DeviceName, evt.SourceName)
@@ -579,7 +501,12 @@ func (ems *EdgexMsgBusSink) Collect(ctx api.StreamContext, item interface{}) err
 		topic = ems.topic
 	}
 
-	if e := ems.client.Publish(env, topic); e != nil {
+	req := &edgex.RequestInfo{
+		ContentType: ems.c.ContentType,
+	}
+	c := edgex.WithRequestInfo(ctx.(*defaultCtx.DefaultContext), req)
+
+	if e := ems.cli.Publish(c, topic, data); e != nil {
 		logger.Errorf("%s: found error %s when publish to EdgeX message bus.\n", e)
 		return fmt.Errorf("%s:%s", errorx.IOErr, e.Error())
 	}
@@ -591,13 +518,8 @@ func (ems *EdgexMsgBusSink) Collect(ctx api.StreamContext, item interface{}) err
 func (ems *EdgexMsgBusSink) Close(ctx api.StreamContext) error {
 	logger := ctx.GetLogger()
 	logger.Infof("Closing edgex sink")
-	if ems.client != nil && ems.c.ConnectionSelector == "" {
-		if e := ems.client.Disconnect(); e != nil {
-			return e
-		}
-	}
-	if ems.c.ConnectionSelector != "" {
-		ctx.ReleaseConnection(ems.c.ConnectionSelector)
+	if ems.cli != nil {
+		ems.cli.Release(ctx)
 	}
 	return nil
 }
diff --git a/internal/topo/sink/mqtt_sink.go b/internal/topo/sink/mqtt_sink.go
index 4112b7fa..51c3fee3 100644
--- a/internal/topo/sink/mqtt_sink.go
+++ b/internal/topo/sink/mqtt_sink.go
@@ -15,33 +15,21 @@
 package sink
 
 import (
-	"crypto/tls"
 	"fmt"
-	MQTT "github.com/eclipse/paho.mqtt.golang"
-	"github.com/google/uuid"
-	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/topo/connection/clients/mqtt"
+	defaultCtx "github.com/lf-edge/ekuiper/internal/topo/context"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
-	"strings"
 )
 
 type MQTTSink struct {
-	srv      string
 	tpc      string
-	clientid string
-	pVersion uint
 	qos      byte
-	uName    string
-	password string
-	certPath string
-	pkeyPath string
-	conSel   string
+	retained bool
 
-	insecureSkipVerify bool
-	retained           bool
-
-	conn MQTT.Client
+	config map[string]interface{}
+	cli    api.MessageClient
 }
 
 func (ms *MQTTSink) hasKeys(str []string, ps map[string]interface{}) bool {
@@ -54,99 +42,6 @@ func (ms *MQTTSink) hasKeys(str []string, ps map[string]interface{}) bool {
 }
 
 func (ms *MQTTSink) Configure(ps map[string]interface{}) error {
-	conSelector := ""
-	if pk, ok := ps["connectionSelector"]; ok {
-		if v, ok := pk.(string); ok {
-			conSelector = v
-		}
-		keys := []string{"server", "clientId", "protocolVersion", "username", "password", "certificationPath", "privateKeyPath", "insecureSkipVerify"}
-		if ms.hasKeys(keys, ps) {
-			return fmt.Errorf("already have connection selector: %s, remove connection related config", conSelector)
-		}
-		ms.conSel = conSelector
-	} else {
-
-		srv := ""
-		if pk, ok := ps["server"]; ok {
-			if v, ok := pk.(string); ok {
-				srv = v
-			}
-		} else {
-			return fmt.Errorf("mqtt sink is missing property server")
-		}
-
-		clientid, ok := ps["clientId"]
-		if !ok {
-			if uuid, err := uuid.NewUUID(); err != nil {
-				return fmt.Errorf("mqtt sink fails to get uuid, the error is %s", err)
-			} else {
-
-				clientid = uuid.String()
-			}
-		}
-		var pVersion uint = 3
-		pVersionStr, ok := ps["protocolVersion"]
-		if ok {
-			v, _ := pVersionStr.(string)
-			if v == "3.1" {
-				pVersion = 3
-			} else if v == "3.1.1" {
-				pVersion = 4
-			} else {
-				return fmt.Errorf("unknown protocol version %s, the value could be only 3.1 or 3.1.1 (also refers to MQTT version 4)", pVersionStr)
-			}
-		}
-
-		uName := ""
-		un, ok := ps["username"]
-		if ok {
-			v, _ := un.(string)
-			if strings.Trim(v, " ") != "" {
-				uName = v
-			}
-		}
-
-		password := ""
-		pwd, ok := ps["password"]
-		if ok {
-			v, _ := pwd.(string)
-			if strings.Trim(v, " ") != "" {
-				password = v
-			}
-		}
-
-		certPath := ""
-		if cp, ok := ps["certificationPath"]; ok {
-			if v, ok := cp.(string); ok {
-				certPath = v
-			}
-		}
-
-		pKeyPath := ""
-		if pk, ok := ps["privateKeyPath"]; ok {
-			if v, ok := pk.(string); ok {
-				pKeyPath = v
-			}
-		}
-
-		insecureSkipVerify := false
-		if pk, ok := ps["insecureSkipVerify"]; ok {
-			if v, ok := pk.(bool); ok {
-				insecureSkipVerify = v
-			}
-		}
-
-		ms.srv = srv
-		ms.clientid = clientid.(string)
-		ms.pVersion = pVersion
-
-		ms.uName = uName
-		ms.password = password
-		ms.certPath = certPath
-		ms.pkeyPath = pKeyPath
-		ms.insecureSkipVerify = insecureSkipVerify
-	}
-
 	tpc, ok := ps["topic"]
 	if !ok {
 		return fmt.Errorf("mqtt sink is missing property topic")
@@ -169,6 +64,7 @@ func (ms *MQTTSink) Configure(ps map[string]interface{}) error {
 		}
 	}
 
+	ms.config = ps
 	ms.qos = qos
 	ms.tpc = tpc.(string)
 	ms.retained = retained
@@ -177,68 +73,16 @@ func (ms *MQTTSink) Configure(ps map[string]interface{}) error {
 }
 
 func (ms *MQTTSink) Open(ctx api.StreamContext) error {
-	var client MQTT.Client
 	log := ctx.GetLogger()
-	if ms.conSel != "" {
-		con, err := ctx.GetConnection(ms.conSel)
-		if err != nil {
-			log.Errorf("The mqtt client for connection selector %s get fail with error: %s", ms.conSel, err)
-			return err
-		}
-		client = con.(MQTT.Client)
-		log.Infof("The mqtt client for connection selector %s get successfully", ms.conSel)
-	} else {
-		log.Infof("Opening mqtt sink for rule %s.", ctx.GetRuleId())
-		opts := MQTT.NewClientOptions().AddBroker(ms.srv).SetClientID(ms.clientid)
-
-		if ms.certPath != "" || ms.pkeyPath != "" {
-			log.Infof("Connect MQTT broker with certification and keys.")
-			if cp, err := conf.ProcessPath(ms.certPath); err == nil {
-				if kp, err1 := conf.ProcessPath(ms.pkeyPath); err1 == nil {
-					if cer, err2 := tls.LoadX509KeyPair(cp, kp); err2 != nil {
-						return err2
-					} else {
-						opts.SetTLSConfig(&tls.Config{Certificates: []tls.Certificate{cer}, InsecureSkipVerify: ms.insecureSkipVerify})
-					}
-				} else {
-					return err1
-				}
-			} else {
-				return err
-			}
-		} else {
-			log.Infof("Connect MQTT broker with username and password.")
-			if ms.uName != "" {
-				opts = opts.SetUsername(ms.uName)
-			}
-
-			if ms.password != "" {
-				opts = opts.SetPassword(ms.password)
-			}
-		}
-
-		opts.SetAutoReconnect(true)
-		var reconn = false
-		opts.SetConnectionLostHandler(func(client MQTT.Client, e error) {
-			log.Errorf("The connection %s is disconnected due to error %s, will try to re-connect later.", ms.srv+": "+ms.clientid, e)
-			ms.conn = client
-			reconn = true
-		})
-
-		opts.SetOnConnectHandler(func(client MQTT.Client) {
-			if reconn {
-				log.Infof("The connection is %s re-established successfully.", ms.srv+": "+ms.clientid)
-			}
-		})
 
-		client = MQTT.NewClient(opts)
-		if token := client.Connect(); token.Wait() && token.Error() != nil {
-			return fmt.Errorf("Found error: %s", token.Error())
-		}
-		log.Infof("The connection to server %s:%d was established successfully", ms.srv, ms.clientid)
+	cli, err := ctx.GetClient("mqtt", ms.config)
+	if err != nil {
+		log.Errorf("found error when get mqtt client config %v, error %s", ms.config, err.Error())
+		return err
 	}
 
-	ms.conn = client
+	ms.cli = cli
+
 	return nil
 }
 
@@ -248,7 +92,7 @@ func (ms *MQTTSink) Collect(ctx api.StreamContext, item interface{}) error {
 	if err != nil {
 		return err
 	}
-	c := ms.conn
+
 	logger.Debugf("%s publish %s", ctx.GetOpId(), jsonBytes)
 	tpc, err := ctx.ParseDynamicProp(ms.tpc, item)
 	if err != nil {
@@ -257,8 +101,15 @@ func (ms *MQTTSink) Collect(ctx api.StreamContext, item interface{}) error {
 	if tpc, ok := tpc.(string); !ok {
 		return fmt.Errorf("the value %v of dynamic prop %s for topic is not a string", ms.tpc, tpc)
 	}
-	if token := c.Publish(tpc.(string), ms.qos, ms.retained, jsonBytes); token.Wait() && token.Error() != nil {
-		return fmt.Errorf("%s: %s", errorx.IOErr, token.Error())
+
+	req := &mqtt.RequestInfo{
+		Qos:      ms.qos,
+		Retained: ms.retained,
+	}
+	c := mqtt.WithRequestInfo(ctx.(*defaultCtx.DefaultContext), req)
+
+	if err := ms.cli.Publish(c, tpc.(string), jsonBytes); err != nil {
+		return fmt.Errorf("%s: %s", errorx.IOErr, err.Error())
 	}
 	return nil
 }
@@ -266,11 +117,8 @@ func (ms *MQTTSink) Collect(ctx api.StreamContext, item interface{}) error {
 func (ms *MQTTSink) Close(ctx api.StreamContext) error {
 	logger := ctx.GetLogger()
 	logger.Infof("Closing mqtt sink")
-	if ms.conn != nil && ms.conn.IsConnected() && ms.conSel == "" {
-		ms.conn.Disconnect(5000)
-	}
-	if ms.conSel != "" {
-		ctx.ReleaseConnection(ms.conSel)
+	if ms.cli != nil {
+		ms.cli.Release(ctx)
 	}
 	return nil
 }
diff --git a/internal/topo/source/edgex_source.go b/internal/topo/source/edgex_source.go
index c0e9b8ad..75061b2e 100644
--- a/internal/topo/source/edgex_source.go
+++ b/internal/topo/source/edgex_source.go
@@ -23,36 +23,26 @@ import (
 	v2 "github.com/edgexfoundry/go-mod-core-contracts/v2/common"
 	"github.com/edgexfoundry/go-mod-core-contracts/v2/dtos"
 	"github.com/edgexfoundry/go-mod-core-contracts/v2/dtos/requests"
-	"github.com/edgexfoundry/go-mod-messaging/v2/messaging"
 	"github.com/edgexfoundry/go-mod-messaging/v2/pkg/types"
-	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
-	"github.com/lf-edge/ekuiper/pkg/message"
 	"strconv"
 	"strings"
 )
 
 type EdgexSource struct {
-	mbconf types.MessageBusConfig
-	conSel string
+	cli api.MessageClient
 
-	client      messaging.MessageClient
-	subscribed  bool
+	config      map[string]interface{}
 	topic       string
 	messageType messageType
+	buflen      int
 }
 
 type EdgexConf struct {
-	Format             string            `json:"format"`
-	Protocol           string            `json:"protocol"`
-	Server             string            `json:"server"`
-	Port               int               `json:"port"`
-	Topic              string            `json:"topic"`
-	Type               string            `json:"type"`
-	MessageType        messageType       `json:"messageType"`
-	Optional           map[string]string `json:"optional"`
-	ConnectionSelector string            `json:"connectionSelector"`
+	Topic       string      `json:"topic"`
+	MessageType messageType `json:"messageType"`
+	BufferLen   int         `json:"bufferLength"`
 }
 
 type messageType string
@@ -64,109 +54,62 @@ const (
 
 func (es *EdgexSource) Configure(_ string, props map[string]interface{}) error {
 	c := &EdgexConf{
-		Format:      message.FormatJson,
-		Protocol:    "redis",
-		Server:      "localhost",
-		Port:        6379,
-		Type:        messaging.Redis,
 		MessageType: MessageTypeEvent,
 	}
 	err := cast.MapToStruct(props, c)
 	if err != nil {
 		return fmt.Errorf("read properties %v fail with error: %v", props, err)
 	}
-	if c.Format != message.FormatJson {
-		return fmt.Errorf("edgex source only supports `json` format")
+	if c.BufferLen <= 0 {
+		c.BufferLen = 1024
 	}
-
-	if c.MessageType != MessageTypeEvent && c.MessageType != MessageTypeRequest {
-		return fmt.Errorf("specified wrong messageType value %s", c.MessageType)
-	}
-
-	if c.Type != messaging.ZeroMQ && c.Type != messaging.MQTT && c.Type != messaging.Redis {
-		return fmt.Errorf("specified wrong type value %s", c.Type)
-	}
-
-	mbconf := types.MessageBusConfig{SubscribeHost: types.HostInfo{Protocol: c.Protocol, Host: c.Server, Port: c.Port}, Type: c.Type}
-	mbconf.Optional = c.Optional
-	if c.ConnectionSelector == "" {
-		printConf(mbconf)
-	} else {
-		conf.Log.Infof("use connection selector %s for edgex source", c.ConnectionSelector)
-	}
-	es.conSel = c.ConnectionSelector
-	es.mbconf = mbconf
+	es.buflen = c.BufferLen
 	es.messageType = c.MessageType
 	es.topic = c.Topic
+	es.config = props
 
 	return nil
 }
 
-// Modify the copied conf to print no password.
-func printConf(mbconf types.MessageBusConfig) {
-	var 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 *EdgexSource) getClient(ctx api.StreamContext) error {
-	log := ctx.GetLogger()
-	if es.conSel != "" {
-		con, err := ctx.GetConnection(es.conSel)
-		if err != nil {
-			log.Errorf("The edgex client for connection selector %s get fail with error: %s", es.conSel, err)
-			return err
-		}
-		es.client = con.(messaging.MessageClient)
-		log.Infof("The edge client for connection selector %s get successfully", es.conSel)
-	} else {
-		c, err := messaging.NewMessageClient(es.mbconf)
-		if err != nil {
-			return err
-		}
-		es.client = c
-
-		if err := es.client.Connect(); err != nil {
-			return fmt.Errorf("Failed to connect to edgex message bus: " + err.Error())
-		}
-		log.Infof("The connection to edgex messagebus is established successfully.")
-	}
-	return nil
-}
-
 func (es *EdgexSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error) {
 	log := ctx.GetLogger()
-	if err := es.getClient(ctx); err != nil {
-		info := fmt.Errorf("Failed to get edgex message client: " + err.Error())
-		log.Errorf(info.Error())
-		errCh <- info
+
+	cli, err := ctx.GetClient("edgex", es.config)
+	if err != nil {
+		errCh <- err
+		log.Errorf("found error when get edgex client, error %s", err.Error())
 		return
 	}
-	messages := make(chan types.MessageEnvelope)
-	topics := []types.TopicChannel{{Topic: es.topic, Messages: messages}}
-	err := make(chan error)
-	if e := es.client.Subscribe(topics, err); e != nil {
+	es.cli = cli.(api.MessageClient)
+
+	messages := make(chan interface{}, es.buflen)
+	topics := []api.TopicChannel{{Topic: es.topic, Messages: messages}}
+	subErrs := make(chan error, len(topics))
+	if e := es.cli.Subscribe(ctx, topics, subErrs); e != nil {
 		log.Errorf("Failed to subscribe to edgex messagebus topic %s.\n", e)
 		errCh <- e
 	} else {
-		es.subscribed = true
 		log.Infof("Successfully subscribed to edgex messagebus topic %s.", es.topic)
 		for {
 			select {
-			case e1 := <-err:
+			case <-ctx.Done():
+				log.Infof("Exit subscription to edgex messagebus topic %s.", es.topic)
+				return
+			case e1 := <-subErrs:
 				errCh <- e1
 				return
-			case env, ok := <-messages:
+			case msg, ok := <-messages:
 				if !ok { // the source is closed
+					log.Infof("Exit subscription to edgex messagebus topic %s.", es.topic)
+					return
+				}
+
+				env, ok := msg.(*types.MessageEnvelope)
+				if !ok {
+					log.Errorf("can not convert interface data to mqtt message.")
 					return
 				}
+
 				if strings.EqualFold(env.ContentType, "application/json") {
 					var r interface{}
 					switch es.messageType {
@@ -349,13 +292,10 @@ func convertFloatArray(v string, bitSize int) (interface{}, error) {
 }
 
 func (es *EdgexSource) Close(ctx api.StreamContext) error {
-	if es.subscribed && es.conSel == "" {
-		if e := es.client.Disconnect(); e != nil {
-			return e
-		}
-	}
-	if es.conSel != "" {
-		ctx.ReleaseConnection(es.conSel)
+	log := ctx.GetLogger()
+	log.Infof("EdgeX Source instance %d Done.", ctx.GetInstanceId())
+	if es.cli != nil {
+		es.cli.Release(ctx)
 	}
 	return nil
 }
diff --git a/internal/topo/source/mqtt_source.go b/internal/topo/source/mqtt_source.go
index a9740b9b..06782301 100644
--- a/internal/topo/source/mqtt_source.go
+++ b/internal/topo/source/mqtt_source.go
@@ -16,53 +16,36 @@ package source
 
 import (
 	"fmt"
-	MQTT "github.com/eclipse/paho.mqtt.golang"
-	"github.com/google/uuid"
+	pahoMqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/pkg/cert"
+	"github.com/lf-edge/ekuiper/internal/topo/connection/clients/mqtt"
+	defaultCtx "github.com/lf-edge/ekuiper/internal/topo/context"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"github.com/lf-edge/ekuiper/pkg/message"
 	"path"
 	"strconv"
-	"strings"
 )
 
 type MQTTSource struct {
-	srv        string
-	qos        int
-	format     string
-	tpc        string
-	clientid   string
-	pVersion   uint
-	uName      string
-	password   string
-	certPath   string
-	pkeyPath   string
-	rootCapath string
-	conSel     string
-	InSecure   bool
+	qos    int
+	format string
+	tpc    string
+	buflen int
 
+	config map[string]interface{}
 	model  modelVersion
 	schema map[string]interface{}
-	conn   MQTT.Client
+
+	cli api.MessageClient
 }
 
 type MQTTConfig struct {
-	Format             string   `json:"format"`
-	Qos                int      `json:"qos"`
-	Servers            []string `json:"servers"`
-	Clientid           string   `json:"clientid"`
-	PVersion           string   `json:"protocolVersion"`
-	Uname              string   `json:"username"`
-	Password           string   `json:"password"`
-	Certification      string   `json:"certificationPath"`
-	PrivateKPath       string   `json:"privateKeyPath"`
-	RootCaPath         string   `json:"rootCaPath"`
-	InsecureSkipVerify bool     `json:"insecureSkipVerify"`
-	KubeedgeModelFile  string   `json:"kubeedgeModelFile"`
-	KubeedgeVersion    string   `json:"kubeedgeVersion"`
-	ConnectionSelector string   `json:"connectionSelector"`
+	Format            string `json:"format"`
+	Qos               int    `json:"qos"`
+	BufferLen         int    `json:"bufferLength"`
+	KubeedgeModelFile string `json:"kubeedgeModelFile"`
+	KubeedgeVersion   string `json:"kubeedgeVersion"`
 }
 
 func (ms *MQTTSource) WithSchema(_ string) *MQTTSource {
@@ -70,34 +53,21 @@ func (ms *MQTTSource) WithSchema(_ string) *MQTTSource {
 }
 
 func (ms *MQTTSource) Configure(topic string, props map[string]interface{}) error {
-	cfg := &MQTTConfig{}
+	cfg := &MQTTConfig{
+		BufferLen: 1024,
+	}
 	err := cast.MapToStruct(props, cfg)
 	if err != nil {
 		return fmt.Errorf("read properties %v fail with error: %v", props, err)
 	}
-	ms.tpc = topic
-	if cfg.ConnectionSelector == "" {
-		if srvs := cfg.Servers; srvs != nil && len(srvs) > 0 {
-			ms.srv = srvs[0]
-		} else {
-			return fmt.Errorf("missing server property")
-		}
+	if cfg.BufferLen <= 0 {
+		cfg.BufferLen = 1024
 	}
-	ms.conSel = cfg.ConnectionSelector
+	ms.buflen = cfg.BufferLen
+	ms.tpc = topic
 	ms.format = cfg.Format
-	ms.clientid = cfg.Clientid
 	ms.qos = cfg.Qos
-
-	ms.pVersion = 3
-	if cfg.PVersion == "3.1.1" {
-		ms.pVersion = 4
-	}
-
-	ms.uName = cfg.Uname
-	ms.password = strings.Trim(cfg.Password, " ")
-	ms.certPath = cfg.Certification
-	ms.pkeyPath = cfg.PrivateKPath
-	ms.rootCapath = cfg.RootCaPath
+	ms.config = props
 
 	if 0 != len(cfg.KubeedgeModelFile) {
 		p := path.Join("sources", cfg.KubeedgeModelFile)
@@ -111,131 +81,88 @@ func (ms *MQTTSource) Configure(topic string, props map[string]interface{}) erro
 }
 
 func (ms *MQTTSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error) {
-	var client MQTT.Client
 	log := ctx.GetLogger()
 
-	if ms.conSel != "" {
-		con, err := ctx.GetConnection(ms.conSel)
-		if err != nil {
-			log.Errorf("The mqtt client for connection selector %s get fail with error: %s", ms.conSel, err)
-			errCh <- err
-			return
-		}
-		client = con.(MQTT.Client)
-		log.Infof("The mqtt client for connection selector %s get successfully", ms.conSel)
-	} else {
-		opts := MQTT.NewClientOptions().AddBroker(ms.srv).SetProtocolVersion(ms.pVersion)
-		if ms.clientid == "" {
-			if newUUID, err := uuid.NewUUID(); err != nil {
-				errCh <- fmt.Errorf("failed to get uuid, the error is %s", err)
-				return
-			} else {
-				ms.clientid = newUUID.String()
-				opts = opts.SetClientID(newUUID.String())
-			}
-		} else {
-			opts = opts.SetClientID(ms.clientid)
-		}
-
-		tlsOpts := cert.TlsConfigurationOptions{
-			SkipCertVerify: ms.InSecure,
-			CertFile:       ms.certPath,
-			KeyFile:        ms.pkeyPath,
-			CaFile:         ms.rootCapath,
-		}
-		log.Infof("Connect MQTT broker with TLS configs. %v", tlsOpts)
-		tlscfg, err := cert.GenerateTLSForClient(tlsOpts)
-		if err != nil {
-			errCh <- err
-			return
-		}
-
-		opts = opts.SetTLSConfig(tlscfg)
-
-		log.Infof("Connect MQTT broker with username and password.")
-		if ms.uName != "" {
-			opts = opts.SetUsername(ms.uName)
-		} else {
-			log.Infof("The username is empty.")
-		}
-
-		if ms.password != "" {
-			opts = opts.SetPassword(ms.password)
-		} else {
-			log.Infof("The password is empty.")
-		}
-		opts.SetAutoReconnect(true)
-		var reconn = false
-		opts.SetConnectionLostHandler(func(client MQTT.Client, e error) {
-			log.Errorf("The connection %s is disconnected due to error %s, will try to re-connect later.", ms.srv+": "+ms.clientid, e)
-			reconn = true
-			subscribe(ms, client, ctx, consumer)
-		})
-
-		opts.SetOnConnectHandler(func(client MQTT.Client) {
-			if reconn {
-				log.Infof("The connection is %s re-established successfully.", ms.srv+": "+ms.clientid)
-				subscribe(ms, client, ctx, consumer)
-			}
-		})
-
-		client = MQTT.NewClient(opts)
-
-		if token := client.Connect(); token.Wait() && token.Error() != nil {
-			errCh <- fmt.Errorf("found error when connecting to %s: %s", ms.srv, token.Error())
-			return
-		}
-		log.Infof("The connection to server %s:%s was established successfully", ms.srv, ms.clientid)
+	cli, err := ctx.GetClient("mqtt", ms.config)
+	if err != nil {
+		errCh <- err
+		log.Errorf("found error when get mqtt client config %v, error %s", ms.config, err.Error())
+		return
+	}
+	ms.cli = cli
+	err = subscribe(ms, ctx, consumer)
+	if err != nil {
+		errCh <- err
 	}
-
-	ms.conn = client
-	subscribe(ms, client, ctx, consumer)
 }
 
-func subscribe(ms *MQTTSource, client MQTT.Client, ctx api.StreamContext, consumer chan<- api.SourceTuple) {
+func subscribe(ms *MQTTSource, ctx api.StreamContext, consumer chan<- api.SourceTuple) error {
 	log := ctx.GetLogger()
-	h := func(client MQTT.Client, msg MQTT.Message) {
-		log.Debugf("instance %d received %s", ctx.GetInstanceId(), msg.Payload())
-		result, e := message.Decode(msg.Payload(), ms.format)
-		//The unmarshal type can only be bool, float64, string, []interface{}, map[string]interface{}, nil
-		if e != nil {
-			log.Errorf("Invalid data format, cannot decode %s to %s format with error %s", string(msg.Payload()), ms.format, e)
-			return
-		}
-
-		meta := make(map[string]interface{})
-		meta["topic"] = msg.Topic()
-		meta["messageid"] = strconv.Itoa(int(msg.MessageID()))
-
-		if nil != ms.model {
-			sliErr := ms.model.checkType(result, msg.Topic())
-			for _, v := range sliErr {
-				log.Errorf(v)
-			}
-		}
 
-		select {
-		case consumer <- api.NewDefaultSourceTuple(result, meta):
-			log.Debugf("send data to source node")
-		case <-ctx.Done():
-			return
-		}
+	messages := make(chan interface{}, ms.buflen)
+	topics := []api.TopicChannel{{Topic: ms.tpc, Messages: messages}}
+	err := make(chan error, len(topics))
+	req := &mqtt.RequestInfo{
+		Qos: byte(ms.qos),
 	}
+	c := mqtt.WithRequestInfo(ctx.(*defaultCtx.DefaultContext), req)
 
-	if token := client.Subscribe(ms.tpc, byte(ms.qos), h); token.Wait() && token.Error() != nil {
-		log.Errorf("Found error: %s", token.Error())
+	if e := ms.cli.Subscribe(c, topics, err); e != nil {
+		log.Errorf("Failed to subscribe to mqtt topic %s, error %s\n", ms.tpc, e.Error())
+		return e
 	} else {
-		log.Infof("Successfully subscribe to topic %s", ms.tpc)
+		log.Infof("Successfully subscribed to topic %s.", ms.tpc)
+		for {
+			select {
+			case <-ctx.Done():
+				log.Infof("Exit subscription to edgex messagebus topic %s.", ms.tpc)
+				return nil
+			case e1 := <-err:
+				log.Errorf("the subscription to mqtt topic %s have error %s.\n", ms.tpc, e1.Error())
+				return e1
+			case env, ok := <-messages:
+				if !ok { // the source is closed
+					log.Infof("Exit subscription to edgex messagebus topic %s.", ms.tpc)
+					return nil
+				}
+				msg, ok := env.(pahoMqtt.Message)
+				if !ok {
+					log.Errorf("can not convert interface data to mqtt message %s.", ms.tpc)
+					return nil
+				}
+				result, e := message.Decode(msg.Payload(), ms.format)
+				//The unmarshal type can only be bool, float64, string, []interface{}, map[string]interface{}, nil
+				if e != nil {
+					log.Errorf("Invalid data format, cannot decode %s to %s format with error %s", string(msg.Payload()), ms.format, e)
+					return e
+				}
+
+				meta := make(map[string]interface{})
+				meta["topic"] = msg.Topic()
+				meta["messageid"] = strconv.Itoa(int(msg.MessageID()))
+
+				if nil != ms.model {
+					sliErr := ms.model.checkType(result, msg.Topic())
+					for _, v := range sliErr {
+						log.Errorf(v)
+					}
+				}
+
+				select {
+				case consumer <- api.NewDefaultSourceTuple(result, meta):
+					log.Debugf("send data to source node")
+				case <-ctx.Done():
+					return nil
+				}
+			}
+		}
 	}
 }
 
 func (ms *MQTTSource) Close(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("Mqtt Source instance %d Done", ctx.GetInstanceId())
-	if ms.conn != nil && ms.conn.IsConnected() && ms.conSel == "" {
-		ms.conn.Disconnect(5000)
-	}
-	if ms.conSel != "" {
-		ctx.ReleaseConnection(ms.conSel)
+	if ms.cli != nil {
+		ms.cli.Release(ctx)
 	}
 	return nil
 }
diff --git a/internal/xsql/lexical.go b/internal/xsql/lexical.go
index 087e3bc5..70b04e36 100644
--- a/internal/xsql/lexical.go
+++ b/internal/xsql/lexical.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2022 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -308,7 +308,10 @@ func (s *Scanner) ScanString() (tok ast.Token, lit string) {
 			buf.WriteRune(ch)
 		}
 	}
-	r, _ := strconv.Unquote(buf.String())
+	r, err := strconv.Unquote(buf.String())
+	if err != nil {
+		return ast.ILLEGAL, "invalid string: " + buf.String()
+	}
 	return ast.STRING, r
 }
 
diff --git a/pkg/api/stream.go b/pkg/api/stream.go
index 0cd1ec3d..2165b9eb 100644
--- a/pkg/api/stream.go
+++ b/pkg/api/stream.go
@@ -150,8 +150,7 @@ type StreamContext interface {
 	GetState(key string) (interface{}, error)
 	DeleteState(key string) error
 	// Connection related methods
-	GetConnection(connectSelector string) (interface{}, error)
-	ReleaseConnection(connectSelector string)
+	GetClient(clientType string, config map[string]interface{}) (MessageClient, error)
 	// Properties processing, prop is a json path
 	ParseDynamicProp(prop string, data interface{}) (interface{}, error)
 	// Transform output according to the properties like syntax
@@ -188,3 +187,17 @@ const (
 )
 
 type Qos int
+
+type MessageClient interface {
+	Subscribe(c StreamContext, subChan []TopicChannel, messageErrors chan error) error
+	Release(c StreamContext)
+	Publish(c StreamContext, topic string, message []byte) error
+}
+
+// TopicChannel is the data structure for subscriber
+type TopicChannel struct {
+	// Topic for subscriber to filter on if any
+	Topic string
+	// Messages is the returned message channel for the subscriber
+	Messages chan<- interface{}
+}


Test patch:
diff --git a/internal/conf/load_test.go b/internal/conf/load_test.go
index 2f351554..83fd3ed8 100644
--- a/internal/conf/load_test.go
+++ b/internal/conf/load_test.go
@@ -15,6 +15,7 @@
 package conf
 
 import (
+	"fmt"
 	"os"
 	"reflect"
 	"testing"
@@ -149,3 +150,17 @@ func createExpectedRandomConfigMap() map[string]interface{} {
 	}
 	return input
 }
+
+func TestPrintable(t *testing.T) {
+	bef := map[string]interface{}{
+		"password": "password",
+		"Password": "password",
+		"optional": map[string]interface{}{
+			"password": "password",
+			"Password": "password",
+		},
+	}
+	after := Printable(bef)
+
+	_, _ = fmt.Printf("after %v", after)
+}
diff --git a/internal/topo/connection/client_edgex_test.go b/internal/topo/connection/client_edgex_test.go
deleted file mode 100644
index 113da02b..00000000
--- a/internal/topo/connection/client_edgex_test.go
+++ /dev/null
@@ -1,131 +0,0 @@
-// Copyright 2021 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
-// +build edgex
-
-package connection
-
-import (
-	"github.com/edgexfoundry/go-mod-messaging/v2/pkg/types"
-	"github.com/lf-edge/ekuiper/internal/conf"
-	"testing"
-)
-
-func TestEdgex_CfgValidate(t *testing.T) {
-	type fields struct {
-		mbconf types.MessageBusConfig
-	}
-	type args struct {
-		props map[string]interface{}
-	}
-	tests := []struct {
-		name    string
-		fields  fields
-		args    args
-		wantErr bool
-	}{
-		{
-			name:   "config pass",
-			fields: fields{},
-			args: args{props: map[string]interface{}{
-				"protocol": "tcp",
-				"server":   "127.0.0.1",
-				"port":     int64(1883),
-				"type":     "mqtt",
-				"optional": map[string]string{
-					"ClientId": "client1",
-					"Username": "user1",
-				},
-			}},
-			wantErr: false,
-		},
-		{
-			name:   "config not case sensitive",
-			fields: fields{},
-			args: args{props: map[string]interface{}{
-				"Protocol": "tcp",
-				"server":   "127.0.0.1",
-				"Port":     1883,
-				"type":     "mqtt",
-				"optional": map[string]string{
-					"ClientId": "client1",
-					"Username": "user1",
-				},
-			}},
-
-			wantErr: false,
-		},
-		{
-			name:   "have unwanted config items topic",
-			fields: fields{},
-			args: args{props: map[string]interface{}{
-				"protocol": "tcp",
-				"server":   "127.0.0.1",
-				"port":     1883,
-				"type":     "mqtt",
-				"optional": map[string]string{
-					"ClientId": "client1",
-					"Username": "user1",
-				},
-				"topic": "demo",
-			}},
-
-			wantErr: true,
-		},
-		{
-			name:   "config type not in zero/mqtt/redis ",
-			fields: fields{},
-			args: args{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 ",
-			fields: fields{},
-			args: args{props: map[string]interface{}{
-				"protocol": "tcp",
-				"type":     "mqtt",
-				"optional": map[string]string{
-					"ClientId": "client1",
-					"Username": "user1",
-				},
-			}},
-
-			wantErr: true,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			es := &EdgexClient{
-				selector: &conf.ConSelector{
-					ConnSelectorStr: "testSelector",
-				},
-				mbconf: tt.fields.mbconf,
-			}
-			if err := es.CfgValidate(tt.args.props); (err != nil) != tt.wantErr {
-				t.Errorf("CfgValidate() error = %v, wantErr %v", err, tt.wantErr)
-			}
-		})
-	}
-}
diff --git a/internal/topo/connection/clients/edgex/client_edgex_test.go b/internal/topo/connection/clients/edgex/client_edgex_test.go
new file mode 100644
index 00000000..3ceecb92
--- /dev/null
+++ b/internal/topo/connection/clients/edgex/client_edgex_test.go
@@ -0,0 +1,215 @@
+// Copyright 2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build edgex
+// +build edgex
+
+package edgex
+
+import (
+	"github.com/edgexfoundry/go-mod-messaging/v2/pkg/types"
+	"reflect"
+	"testing"
+)
+
+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]string{
+					"ClientId": "client1",
+					"Username": "user1",
+				},
+			},
+			wantErr: false,
+			expConf: types.MessageBusConfig{
+				PublishHost: types.HostInfo{
+					Host:     "127.0.0.1",
+					Port:     1883,
+					Protocol: "tcp",
+				},
+				SubscribeHost: types.HostInfo{
+					Host:     "127.0.0.1",
+					Port:     1883,
+					Protocol: "tcp",
+				},
+				Type: "mqtt",
+				Optional: map[string]string{
+					"ClientId": "client1",
+					"Username": "user1",
+				},
+			},
+		},
+		{
+			name: "config pass",
+			props: map[string]interface{}{
+				"protocol": "redis",
+				"server":   "edgex-redis",
+				"port":     6379,
+				"type":     "redis",
+			},
+			wantErr: false,
+			expConf: types.MessageBusConfig{
+				PublishHost: types.HostInfo{
+					Host:     "edgex-redis",
+					Port:     6379,
+					Protocol: "redis",
+				},
+				SubscribeHost: 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{
+				PublishHost: types.HostInfo{
+					Host:     "127.0.0.1",
+					Port:     1883,
+					Protocol: "tcp",
+				},
+				SubscribeHost: 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 zero/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{
+				PublishHost: types.HostInfo{
+					Host:     "localhost",
+					Port:     6379,
+					Protocol: "redis",
+				},
+				SubscribeHost: 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":     "zmq",
+				"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 := &EdgexClient{}
+			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/topo/connection/client_mqtt_test.go b/internal/topo/connection/clients/mqtt/client_mqtt_test.go
similarity index 70%
rename from internal/topo/connection/client_mqtt_test.go
rename to internal/topo/connection/clients/mqtt/client_mqtt_test.go
index accc7eee..0f12eb14 100644
--- a/internal/topo/connection/client_mqtt_test.go
+++ b/internal/topo/connection/clients/mqtt/client_mqtt_test.go
@@ -1,7 +1,20 @@
-package connection
+// Copyright 2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mqtt
 
 import (
-	"github.com/lf-edge/ekuiper/internal/conf"
 	"reflect"
 	"testing"
 )
@@ -20,7 +33,7 @@ func TestMQTTClient_CfgValidate(t *testing.T) {
 			name: "config pass",
 			args: args{
 				props: map[string]interface{}{
-					"servers": []string{"tcp:127.0.0.1"},
+					"server": "tcp:127.0.0.1",
 				},
 			},
 			wantErr: false,
@@ -29,7 +42,7 @@ func TestMQTTClient_CfgValidate(t *testing.T) {
 			name: "config are not case sensitive",
 			args: args{
 				props: map[string]interface{}{
-					"SERVERS": []string{"tcp:127.0.0.1"},
+					"SERVER": "tcp:127.0.0.1",
 				},
 			},
 			wantErr: false,
@@ -38,20 +51,20 @@ func TestMQTTClient_CfgValidate(t *testing.T) {
 			name: "config server addr key error",
 			args: args{
 				props: map[string]interface{}{
-					"server": []string{"tcp:127.0.0.1"},
+					"servers": []string{"tcp:127.0.0.1"},
 				},
 			},
-			wantErr: true,
+			wantErr: false,
 		},
 		{
 			name: "config have unwanted topic fields",
 			args: args{
 				props: map[string]interface{}{
-					"servers": []string{"tcp:127.0.0.1"},
-					"topic":   "demo",
+					"server": "tcp:127.0.0.1",
+					"topic":  "demo",
 				},
 			},
-			wantErr: true,
+			wantErr: false,
 		},
 		{
 			name: "config no server addr",
@@ -66,7 +79,7 @@ func TestMQTTClient_CfgValidate(t *testing.T) {
 			name: "config no server addr",
 			args: args{
 				props: map[string]interface{}{
-					"servers": []string{},
+					"server": "",
 				},
 			},
 			wantErr: true,
@@ -75,7 +88,7 @@ func TestMQTTClient_CfgValidate(t *testing.T) {
 			name: "config miss cert key file",
 			args: args{
 				props: map[string]interface{}{
-					"servers":           []string{"tcp:127.0.0.1"},
+					"server":            "tcp:127.0.0.1",
 					"certificationPath": "./not_exist.crt",
 					"privateKeyPath":    "./not_exist.key",
 				},
@@ -85,11 +98,7 @@ func TestMQTTClient_CfgValidate(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			ms := &MQTTClient{
-				selector: &conf.ConSelector{
-					ConnSelectorStr: "testSelector",
-				},
-			}
+			ms := &MQTTClient{}
 			err := ms.CfgValidate(tt.args.props)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("CfgValidate() error = %v, wantErr %v", err, tt.wantErr)
@@ -100,17 +109,13 @@ func TestMQTTClient_CfgValidate(t *testing.T) {
 
 func TestMQTTClient_CfgResult(t *testing.T) {
 	props := map[string]interface{}{
-		"servers":  []string{"tcp:127.0.0.1:1883"},
+		"server":   "tcp:127.0.0.1:1883",
 		"USERNAME": "demo",
 		"Password": "password",
 		"clientID": "clientid",
 	}
 
-	ms := &MQTTClient{
-		selector: &conf.ConSelector{
-			ConnSelectorStr: "testSelector",
-		},
-	}
+	ms := &MQTTClient{}
 
 	_ = ms.CfgValidate(props)
 
diff --git a/internal/topo/connection/connect_selector_test.go b/internal/topo/connection/connect_selector_test.go
deleted file mode 100644
index 0db9c696..00000000
--- a/internal/topo/connection/connect_selector_test.go
+++ /dev/null
@@ -1,154 +0,0 @@
-package connection
-
-import (
-	"github.com/lf-edge/ekuiper/internal/conf"
-	"os"
-	"reflect"
-	"testing"
-)
-
-func Test_getConnectionConf(t *testing.T) {
-	type args struct {
-		connectionSelector string
-	}
-	tests := []struct {
-		name    string
-		args    args
-		want    map[string]interface{}
-		wantErr bool
-	}{
-		{
-			name: "mqtt:localConnection",
-			args: args{
-				connectionSelector: "mqtt.localConnection",
-			},
-			want: map[string]interface{}{
-				"servers":  []interface{}{"tcp://127.0.0.1:1883"},
-				"username": "ekuiper",
-				"password": "password",
-				"clientid": "ekuiper",
-			},
-			wantErr: false,
-		},
-		{
-			name: "mqtt:cloudConnection",
-			args: args{
-				connectionSelector: "mqtt.cloudConnection",
-			},
-			want: map[string]interface{}{
-				"servers":  []interface{}{"tcp://broker.emqx.io:1883"},
-				"username": "user1",
-				"password": "password",
-			},
-			wantErr: false,
-		},
-		{
-			name: "mqtt:mqtt_conf3 not exist",
-			args: args{
-				connectionSelector: "mqtt.mqtt_conf3",
-			},
-			wantErr: true,
-		},
-		{
-			name: "edgex:redisMsgBus",
-			args: args{
-				connectionSelector: "edgex.redisMsgBus",
-			},
-			want: map[string]interface{}{
-				"protocol": "redis",
-				"server":   "127.0.0.1",
-				"port":     6379,
-				"type":     "redis",
-			},
-			wantErr: false,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			c := conf.ConSelector{
-				ConnSelectorStr: tt.args.connectionSelector,
-			}
-			_ = c.Init()
-
-			got, err := c.ReadCfgFromYaml()
-			if (err != nil) != tt.wantErr {
-				t.Errorf("getConnectionConf() error = %v, wantErr %v", err, tt.wantErr)
-				return
-			}
-			if !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("getConnectionConf() got = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func Test_getConnectionConfWithEnv(t *testing.T) {
-	mqttServerKey := "CONNECTION__MQTT__LOCALCONNECTION__SERVERS"
-	mqttServerValue := "[tcp://broker.emqx.io:1883]"
-
-	edgexPortKey := "CONNECTION__EDGEX__REDISMSGBUS__PORT"
-	edgexPortValue := "6666"
-
-	err := os.Setenv(mqttServerKey, mqttServerValue)
-	if err != nil {
-		t.Error(err)
-	}
-	err = os.Setenv(edgexPortKey, edgexPortValue)
-	if err != nil {
-		t.Error(err)
-	}
-
-	type args struct {
-		connectionSelector string
-	}
-	tests := []struct {
-		name    string
-		args    args
-		want    map[string]interface{}
-		wantErr bool
-	}{
-		{
-			name: "mqtt:localConnection",
-			args: args{
-				connectionSelector: "mqtt.localConnection",
-			},
-			want: map[string]interface{}{
-				"servers":  []interface{}{"tcp://broker.emqx.io:1883"},
-				"username": "ekuiper",
-				"password": "password",
-				"clientid": "ekuiper",
-			},
-			wantErr: false,
-		},
-		{
-			name: "edgex:redisMsgBus",
-			args: args{
-				connectionSelector: "edgex.redisMsgBus",
-			},
-			want: map[string]interface{}{
-				"protocol": "redis",
-				"server":   "127.0.0.1",
-				"port":     int64(6666),
-				"type":     "redis",
-			},
-			wantErr: false,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			c := conf.ConSelector{
-				ConnSelectorStr: tt.args.connectionSelector,
-			}
-			_ = c.Init()
-
-			got, err := c.ReadCfgFromYaml()
-			if (err != nil) != tt.wantErr {
-				t.Errorf("getConnectionConf() error = %v, wantErr %v", err, tt.wantErr)
-				return
-			}
-			if !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("getConnectionConf() got = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
diff --git a/internal/topo/connection/manager_test.go b/internal/topo/connection/manager_test.go
deleted file mode 100644
index bfaad2b2..00000000
--- a/internal/topo/connection/manager_test.go
+++ /dev/null
@@ -1,92 +0,0 @@
-package connection
-
-import (
-	"github.com/lf-edge/ekuiper/internal/conf"
-	"reflect"
-	"testing"
-)
-
-const TestClient = "MockClient"
-
-var GetCalledNumber int
-var CloseCalledNumber int
-
-type MockClient struct {
-	selector *conf.ConSelector
-
-	mockCon string
-}
-
-func (m *MockClient) CfgValidate(m2 map[string]interface{}) error {
-	return nil
-}
-
-func (m *MockClient) GetClient() (interface{}, error) {
-	GetCalledNumber = 1
-	m.mockCon = "MockClient"
-	return m.mockCon, nil
-}
-
-func (m *MockClient) CloseClient() error {
-	CloseCalledNumber = 1
-	m.mockCon = ""
-	return nil
-}
-
-func TestManager(t *testing.T) {
-	registerClientFactory("mqtt", func(super *conf.ConSelector) Client {
-		return &MockClient{selector: super}
-	})
-
-	conSelector := "mqtt.localConnection"
-
-	connection, err := GetConnection(conSelector)
-	if err != nil {
-		t.Errorf("GetConnection Error")
-	}
-	value := connection.(string)
-	if !reflect.DeepEqual(value, TestClient) {
-		t.Errorf("Error")
-	}
-
-	connection, err = GetConnection(conSelector)
-	if err != nil {
-		t.Errorf("GetConnection Error")
-	}
-
-	wrapper := m.clientMap[conSelector]
-	if !reflect.DeepEqual(wrapper.refCnt, uint32(2)) {
-		t.Errorf("Error, ectual=%v, want=%v", wrapper.refCnt, 2)
-	}
-
-	connection, err = GetConnection(conSelector)
-	if err != nil {
-		t.Errorf("GetConnection Error")
-	}
-	if !reflect.DeepEqual(wrapper.refCnt, uint32(3)) {
-		t.Errorf("Error")
-	}
-
-	ReleaseConnection(conSelector)
-	if !reflect.DeepEqual(wrapper.refCnt, uint32(2)) {
-		t.Errorf("Error")
-	}
-
-	ReleaseConnection(conSelector)
-	if !reflect.DeepEqual(wrapper.refCnt, uint32(1)) {
-		t.Errorf("Error")
-	}
-
-	ReleaseConnection(conSelector)
-	if !reflect.DeepEqual(wrapper.refCnt, uint32(0)) {
-		t.Errorf("Error")
-	}
-
-	if _, ok := m.clientMap[conSelector]; ok {
-		t.Errorf("Error")
-	}
-
-	if !reflect.DeepEqual(GetCalledNumber, CloseCalledNumber) || !reflect.DeepEqual(GetCalledNumber, 1) {
-		t.Errorf("Error")
-	}
-}
diff --git a/internal/topo/sink/edgex_sink_test.go b/internal/topo/sink/edgex_sink_test.go
index 1636e379..f4a7a30a 100644
--- a/internal/topo/sink/edgex_sink_test.go
+++ b/internal/topo/sink/edgex_sink_test.go
@@ -22,7 +22,6 @@ import (
 	"fmt"
 	v2 "github.com/edgexfoundry/go-mod-core-contracts/v2/common"
 	"github.com/edgexfoundry/go-mod-core-contracts/v2/dtos"
-	"github.com/edgexfoundry/go-mod-messaging/v2/messaging"
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/testx"
 	"github.com/lf-edge/ekuiper/internal/topo/context"
@@ -70,10 +69,6 @@ func TestConfigure(t *testing.T) {
 				"metadata": "meta",
 			},
 			expected: &EdgexConf{
-				Protocol:    "redis",
-				Host:        "localhost",
-				Port:        6379,
-				Type:        messaging.Redis,
 				MessageType: MessageTypeEvent,
 				ContentType: "application/json",
 				DeviceName:  "ekuiper",
@@ -94,10 +89,6 @@ func TestConfigure(t *testing.T) {
 				"contentType": "application/json",
 			},
 			expected: &EdgexConf{
-				Protocol:    "redis",
-				Host:        "edgex-redis",
-				Port:        6379,
-				Type:        messaging.Redis,
 				MessageType: MessageTypeEvent,
 				ContentType: "application/json",
 				DeviceName:  "ekuiper",
@@ -120,10 +111,6 @@ func TestConfigure(t *testing.T) {
 				},
 			},
 			expected: &EdgexConf{
-				Protocol:    "tcp",
-				Host:        "127.0.0.1",
-				Port:        1883,
-				Type:        messaging.MQTT,
 				MessageType: MessageTypeEvent,
 				ContentType: "application/json",
 				DeviceName:  "ekuiper",
@@ -131,11 +118,9 @@ func TestConfigure(t *testing.T) {
 				SourceName:  "",
 				Metadata:    "edgex_meta",
 				Topic:       "result",
-				Optional: map[string]string{
-					"ClientId": "edgex_message_bus_001",
-				},
 			},
-		}, { // 3
+		},
+		{ // 3
 			conf: map[string]interface{}{
 				"type":        "redis",
 				"protocol":    "redis",
@@ -146,10 +131,6 @@ func TestConfigure(t *testing.T) {
 				"contentType": "application/json",
 			},
 			expected: &EdgexConf{
-				Protocol:    "redis",
-				Host:        "edgex-redis",
-				Port:        6379,
-				Type:        messaging.Redis,
 				MessageType: MessageTypeRequest,
 				ContentType: "application/json",
 				DeviceName:  "ekuiper",
@@ -157,7 +138,8 @@ func TestConfigure(t *testing.T) {
 				SourceName:  "",
 				TopicPrefix: "edgex/events/device",
 			},
-		}, { // 4
+		},
+		{ // 4
 			conf: map[string]interface{}{
 				"type":        "redis",
 				"protocol":    "redis",
@@ -168,40 +150,8 @@ func TestConfigure(t *testing.T) {
 				"contentType": "application/json",
 			},
 			error: "specified wrong messageType value requests",
-		}, { // 5
-			conf: map[string]interface{}{
-				"type":        20,
-				"protocol":    "redis",
-				"host":        "edgex-redis",
-				"port":        6379,
-				"topicPrefix": "edgex/events/device",
-				"messageType": "requests",
-				"contentType": "application/json",
-			},
-			error: "read properties map[contentType:application/json host:edgex-redis messageType:requests port:6379 protocol:redis topicPrefix:edgex/events/device type:20] fail with error: 1 error(s) decoding:\n\n* 'type' expected type 'string', got unconvertible type 'int', value: '20'",
-		}, { // 6
-			conf: map[string]interface{}{
-				"type":        "redis",
-				"protocol":    "redis",
-				"host":        "edgex-redis",
-				"port":        -1,
-				"topicPrefix": "edgex/events/device",
-				"messageType": "requests",
-				"contentType": "application/json",
-			},
-			error: "specified wrong port value, expect positive integer but got -1",
-		}, { // 7
-			conf: map[string]interface{}{
-				"type":        "zmq",
-				"protocol":    "redis",
-				"host":        "edgex-redis",
-				"port":        6379,
-				"topicPrefix": "edgex/events/device",
-				"messageType": "requests",
-				"contentType": "application/json",
-			},
-			error: "specified wrong type value zmq",
-		}, { // 8
+		},
+		{ // 5
 			conf: map[string]interface{}{
 				"protocol":    "redis",
 				"host":        "edgex-redis",
diff --git a/internal/topo/source/edgex_source_test.go b/internal/topo/source/edgex_source_test.go
index adb4704b..cf8221e6 100644
--- a/internal/topo/source/edgex_source_test.go
+++ b/internal/topo/source/edgex_source_test.go
@@ -324,7 +324,6 @@ func TestPrintConf(t *testing.T) {
 		"Password": "fafsadfsadf=",
 		"Prob":     "Prob",
 	}}
-	printConf(mbconf)
 	if !reflect.DeepEqual(expMbconf, mbconf) {
 		t.Errorf("conf changed after printing")
 	}
diff --git a/internal/xsql/parser_test.go b/internal/xsql/parser_test.go
index 2b11eb2e..d9687dc4 100644
--- a/internal/xsql/parser_test.go
+++ b/internal/xsql/parser_test.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2022 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -1436,6 +1436,10 @@ func TestParser_ParseStatement(t *testing.T) {
 				Sources: []ast.Source{&ast.Table{Name: "tbl"}},
 			},
 		},
+		{
+			s:   `SELECT ".*(/)(?!.*\1)" FROM topic/sensor1 AS t1`,
+			err: `found "invalid string: \".*(/)(?!.*\\1)\"", expected expression.`,
+		},
 	}
 
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))