Repository: lf-edge/ekuiper
Pull request #2924 (2024-06-13 23:29:57)

fix: fix fastjson decode

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

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2924?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [internal/converter/json/converter.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2924?src=pr&el=tree&filepath=internal%2Fconverter%2Fjson%2Fconverter.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvY29udmVydGVyL2pzb24vY29udmVydGVyLmdv) | 85.71% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2924?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |

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


```diff
@@             Coverage Diff              @@
##             master-v2    #2924   +/-   ##
============================================
  Coverage             ?   69.13%           
============================================
  Files                ?      297           
  Lines                ?    28784           
  Branches             ?        0           
============================================
  Hits                 ?    19897           
  Misses               ?     7158           
  Partials             ?     1729           
```



</details>

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


Patch:
diff --git a/internal/converter/json/converter.go b/internal/converter/json/converter.go
index 32fe61f3..d81ea012 100644
--- a/internal/converter/json/converter.go
+++ b/internal/converter/json/converter.go
@@ -188,11 +188,14 @@ func (f *FastJsonConverter) decodeObject(obj *fastjson.Object, schema map[string
 		case fastjson.TypeNull:
 			m[key] = nil
 		case fastjson.TypeObject:
-			valid := f.checkSchema(key, "struct", schema)
+			add, valid := f.checkSchema(key, "struct", schema)
 			if !valid {
 				err = fmt.Errorf("%v has wrong type:%v, expect:%v", key, v.Type().String(), getType(schema[key]))
 				return
 			}
+			if !add {
+				return
+			}
 			childObj, err2 := v.Object()
 			if err2 != nil {
 				err = err2
@@ -211,11 +214,14 @@ func (f *FastJsonConverter) decodeObject(obj *fastjson.Object, schema map[string
 				m[key] = childMap
 			}
 		case fastjson.TypeArray:
-			valid := f.checkSchema(key, "array", schema)
+			add, valid := f.checkSchema(key, "array", schema)
 			if !valid {
 				err = fmt.Errorf("%v has wrong type:%v, expect:%v", key, v.Type().String(), getType(schema[key]))
 				return
 			}
+			if !add {
+				return
+			}
 			childArray, err2 := v.Array()
 			if err2 != nil {
 				err = err2
@@ -286,12 +292,22 @@ func (f *FastJsonConverter) decodeObject(obj *fastjson.Object, schema map[string
 	return m, nil
 }
 
-func (f *FastJsonConverter) checkSchema(key, typ string, schema map[string]*ast.JsonStreamField) (valid bool) {
-	v := schema[key]
+func (f *FastJsonConverter) checkSchema(key, typ string, schema map[string]*ast.JsonStreamField) (add, valid bool) {
+	// for schemaless, allow to decode the key value
+	if schema == nil {
+		return true, true
+	}
+	// for defined schema, skip to decode undefined key
+	v, ok := schema[key]
+	if !ok {
+		return false, true
+	}
+	// for the schema we didn't parse,allow to decode eg: results[0].a.b
 	if v == nil {
-		return true
+		return true, true
 	}
-	return v.Type == typ
+	// for the defined schema type, directly to check
+	return v.Type == typ, v.Type == typ
 }
 
 func (f *FastJsonConverter) extractNumberValue(name string, v *fastjson.Value, field *ast.JsonStreamField) (interface{}, error) {
diff --git a/internal/io/mqtt/client/client.go b/internal/io/mqtt/client/client.go
index fc006da1..460ee433 100644
--- a/internal/io/mqtt/client/client.go
+++ b/internal/io/mqtt/client/client.go
@@ -25,9 +25,9 @@ import (
 	"github.com/google/uuid"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cert"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
@@ -37,7 +37,6 @@ type Connection struct {
 	selId     string
 	logger    api.Logger
 	connected atomic.Bool
-	refCount  atomic.Int32
 	// key is the topic. Each topic will have only one connector
 	subscriptions map[string]*SubscriptionInfo
 }
@@ -90,18 +89,8 @@ func (conn *Connection) onReconnecting(_ pahoMqtt.Client, _ *pahoMqtt.ClientOpti
 	conn.logger.Infof("Reconnecting to mqtt broker")
 }
 
-// Do not call this directly. Call connection pool Attach method to get the connection
-func (conn *Connection) Attach(ctx api.StreamContext) {
-	conn.refCount.Add(1)
-}
-
-func (conn *Connection) Ref(ctx api.StreamContext) int {
-	return int(conn.refCount.Load())
-}
-
 // Do not call this directly. Call connection pool Detach method to release the connection
 func (conn *Connection) DetachSub(ctx api.StreamContext, props map[string]any) {
-	conn.refCount.Add(-1)
 	topic, err := getTopicFromProps(props)
 	if err != nil {
 		return
@@ -110,10 +99,6 @@ func (conn *Connection) DetachSub(ctx api.StreamContext, props map[string]any) {
 	conn.Client.Unsubscribe(topic)
 }
 
-func (conn *Connection) DetachPub(ctx api.StreamContext, props map[string]any) {
-	conn.refCount.Add(-1)
-}
-
 func (conn *Connection) Subscribe(topic string, info *SubscriptionInfo) error {
 	conn.subscriptions[topic] = info
 	token := conn.Client.Subscribe(topic, info.Qos, info.Handler)
@@ -132,12 +117,12 @@ func (conn *Connection) Ping(ctx api.StreamContext) error {
 	}
 }
 
-func CreateConnection(ctx api.StreamContext, selId string, props map[string]any) (modules.Connection, error) {
-	return CreateClient(ctx, selId, props)
+func CreateConnection(ctx api.StreamContext, props map[string]any) (modules.Connection, error) {
+	return CreateClient(ctx, props)
 }
 
 // CreateClient creates a new mqtt client. It is anonymous and does not require a name.
-func CreateClient(ctx api.StreamContext, selId string, props map[string]any) (*Connection, error) {
+func CreateClient(ctx api.StreamContext, props map[string]any) (*Connection, error) {
 	c, err := ValidateConfig(props)
 	if err != nil {
 		return nil, err
@@ -175,9 +160,6 @@ func CreateClient(ctx api.StreamContext, selId string, props map[string]any) (*C
 	}
 	ctx.GetLogger().Infof("new mqtt client created")
 	con.Client = cli
-	if len(selId) > 0 {
-		con.Attach(ctx)
-	}
 	return con, nil
 }
 
@@ -222,7 +204,7 @@ func ValidateConfig(props map[string]any) (*ConnectionConfig, error) {
 }
 
 func CreateAnonymousConnection(ctx api.StreamContext, props map[string]any) (*Connection, error) {
-	cli, err := CreateClient(ctx, "", props)
+	cli, err := CreateClient(ctx, props)
 	if err != nil {
 		return nil, err
 	}
diff --git a/internal/io/mqtt/sink.go b/internal/io/mqtt/sink.go
index 0384464a..2ef6fa37 100644
--- a/internal/io/mqtt/sink.go
+++ b/internal/io/mqtt/sink.go
@@ -19,9 +19,9 @@ import (
 	"strings"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/io/mqtt/client"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 )
 
 // AdConf is the advanced configuration for the mqtt sink
@@ -64,27 +64,17 @@ func (ms *Sink) Provision(_ api.StreamContext, ps map[string]any) error {
 
 func (ms *Sink) Connect(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("Connecting to mqtt server")
-	var cli *client.Connection
 	var err error
-	if len(ms.adconf.SelId) > 0 {
-		conn, err := connection.GetNameConnection(ms.adconf.SelId)
-		if err != nil {
-			return err
-		}
-		c, ok := conn.(*client.Connection)
-		if !ok {
-			return fmt.Errorf("connection %s should be mqtt connection", ms.adconf.SelId)
-		}
-		cli = c
-	} else {
-		id := fmt.Sprintf("%s-%s-%s-mqtt-sink", ctx.GetRuleId(), ctx.GetOpId(), ms.adconf.Tpc)
-		conn, err := connection.CreateNonStoredConnection(ctx, id, "mqtt", ms.config)
-		if err != nil {
-			return err
-		}
-		cli = conn.(*client.Connection)
+	id := fmt.Sprintf("%s-%s-%s-mqtt-sink", ctx.GetRuleId(), ctx.GetOpId(), ms.adconf.Tpc)
+	conn, err := connection.FetchConnection(ctx, id, "mqtt", ms.config)
+	if err != nil {
+		return err
+	}
+	c, ok := conn.(*client.Connection)
+	if !ok {
+		return fmt.Errorf("connection %s should be mqtt connection", ms.adconf.SelId)
 	}
-	ms.cli = cli
+	ms.cli = c
 	return err
 }
 
@@ -111,12 +101,8 @@ func (ms *Sink) Collect(ctx api.StreamContext, item api.RawTuple) error {
 func (ms *Sink) Close(ctx api.StreamContext) error {
 	ctx.GetLogger().Info("Closing mqtt sink connector")
 	if ms.cli != nil {
-		if len(ms.adconf.SelId) < 1 {
-			id := fmt.Sprintf("%s-%s-%s-mqtt-sink", ctx.GetRuleId(), ctx.GetOpId(), ms.adconf.Tpc)
-			connection.DropNonStoredConnection(ctx, id)
-		} else {
-			ms.cli.DetachPub(ctx, nil)
-		}
+		id := fmt.Sprintf("%s-%s-%s-mqtt-sink", ctx.GetRuleId(), ctx.GetOpId(), ms.adconf.Tpc)
+		connection.DetachConnection(ctx, id, ms.config)
 	}
 	return nil
 }
diff --git a/internal/io/mqtt/source.go b/internal/io/mqtt/source.go
index c446a7ce..18317c41 100644
--- a/internal/io/mqtt/source.go
+++ b/internal/io/mqtt/source.go
@@ -20,10 +20,10 @@ import (
 	pahoMqtt "github.com/eclipse/paho.mqtt.golang"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/io/mqtt/client"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
@@ -75,24 +75,12 @@ func (ms *SourceConnector) Connect(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("Connecting to mqtt server")
 	var cli *client.Connection
 	var err error
-	if len(ms.cfg.SelId) > 0 {
-		conn, err := connection.GetNameConnection(ms.cfg.SelId)
-		if err != nil {
-			return err
-		}
-		c, ok := conn.(*client.Connection)
-		if !ok {
-			return fmt.Errorf("connection %s should be mqtt connection", ms.cfg.SelId)
-		}
-		cli = c
-	} else {
-		id := fmt.Sprintf("%s-%s-%s-mqtt-source", ctx.GetRuleId(), ctx.GetOpId(), ms.tpc)
-		conn, err := connection.CreateNonStoredConnection(ctx, id, "mqtt", ms.props)
-		if err != nil {
-			return err
-		}
-		cli = conn.(*client.Connection)
+	id := fmt.Sprintf("%s-%s-%s-mqtt-source", ctx.GetRuleId(), ctx.GetOpId(), ms.tpc)
+	conn, err := connection.FetchConnection(ctx, id, "mqtt", ms.props)
+	if err != nil {
+		return err
 	}
+	cli = conn.(*client.Connection)
 	ms.cli = cli
 	return err
 }
@@ -126,12 +114,9 @@ func (ms *SourceConnector) onMessage(ctx api.StreamContext, msg pahoMqtt.Message
 func (ms *SourceConnector) Close(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("Closing mqtt source connector to topic %s.", ms.tpc)
 	if ms.cli != nil {
-		if len(ms.cfg.SelId) < 1 {
-			id := fmt.Sprintf("%s-%s-%s-mqtt-source", ctx.GetRuleId(), ctx.GetOpId(), ms.tpc)
-			connection.DropNonStoredConnection(ctx, id)
-		} else {
-			ms.cli.DetachSub(ctx, ms.props)
-		}
+		id := fmt.Sprintf("%s-%s-%s-mqtt-source", ctx.GetRuleId(), ctx.GetOpId(), ms.tpc)
+		connection.DetachConnection(ctx, id, ms.props)
+		ms.cli.DetachSub(ctx, ms.props)
 	}
 	return nil
 }
diff --git a/internal/server/connection.go b/internal/server/connection.go
index 841c2c1c..5157befe 100644
--- a/internal/server/connection.go
+++ b/internal/server/connection.go
@@ -21,8 +21,8 @@ import (
 
 	"github.com/gorilla/mux"
 
-	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 )
 
 type ConnectionRequest struct {
diff --git a/internal/server/server.go b/internal/server/server.go
index b3930489..ec648f82 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -35,7 +35,6 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/binder/io"
 	"github.com/lf-edge/ekuiper/v2/internal/binder/meta"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/keyedstate"
 	meta2 "github.com/lf-edge/ekuiper/v2/internal/meta"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
@@ -48,6 +47,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/topo/rule"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
diff --git a/internal/io/connection/connection_mock.go b/pkg/connection/connection_mock.go
similarity index 78%
rename from internal/io/connection/connection_mock.go
rename to pkg/connection/connection_mock.go
index 863ff238..7490fff7 100644
--- a/internal/io/connection/connection_mock.go
+++ b/pkg/connection/connection_mock.go
@@ -61,8 +61,8 @@ func (m *mockConnection) Ref(ctx api.StreamContext) int {
 	return m.ref
 }
 
-func CreateMockConnection(ctx api.StreamContext, id string, props map[string]any) (modules.Connection, error) {
-	m := &mockConnection{id: id, ref: 0}
+func CreateMockConnection(ctx api.StreamContext, props map[string]any) (modules.Connection, error) {
+	m := &mockConnection{ref: 0}
 	return m, nil
 }
 
@@ -76,22 +76,10 @@ func (m mockErrConnection) Close(ctx api.StreamContext) {
 	return
 }
 
-func (m mockErrConnection) Attach(ctx api.StreamContext) {
-	return
-}
-
 func (m mockErrConnection) DetachSub(ctx api.StreamContext, props map[string]any) {
 	return
 }
 
-func (m mockErrConnection) DetachPub(ctx api.StreamContext, props map[string]any) {
-	return
-}
-
-func (m mockErrConnection) Ref(ctx api.StreamContext) int {
-	return 0
-}
-
-func CreateMockErrConnection(ctx api.StreamContext, id string, props map[string]any) (modules.Connection, error) {
+func CreateMockErrConnection(ctx api.StreamContext, props map[string]any) (modules.Connection, error) {
 	return nil, errors.New("mockErr")
 }
diff --git a/internal/io/connection/connection_pool.go b/pkg/connection/connection_pool.go
similarity index 73%
rename from internal/io/connection/connection_pool.go
rename to pkg/connection/connection_pool.go
index c5e3c33e..cdd2daaa 100644
--- a/internal/io/connection/connection_pool.go
+++ b/pkg/connection/connection_pool.go
@@ -31,7 +31,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
 
-func storeConnection(plugin, id string, props map[string]interface{}) error {
+func storeConnectionMeta(plugin, id string, props map[string]interface{}) error {
 	err := conf.WriteCfgIntoKVStorage("connections", plugin, id, props)
 	failpoint.Inject("storeConnectionErr", func() {
 		err = errors.New("storeConnectionErr")
@@ -47,6 +47,16 @@ func dropConnectionStore(plugin, id string) error {
 	return err
 }
 
+func GetConnectionRef(id string) int {
+	globalConnectionManager.RLock()
+	defer globalConnectionManager.RUnlock()
+	meta, ok := globalConnectionManager.connectionPool[id]
+	if !ok {
+		return 0
+	}
+	return meta.refCount
+}
+
 func GetAllConnectionStatus(ctx api.StreamContext) map[string]ConnectionStatus {
 	globalConnectionManager.RLock()
 	defer globalConnectionManager.RUnlock()
@@ -83,26 +93,71 @@ func GetAllConnectionsID() []string {
 }
 
 func PingConnection(ctx api.StreamContext, id string) error {
-	conn, err := GetNameConnection(id)
-	if err != nil {
-		return err
+	if id == "" {
+		return fmt.Errorf("connection id should be defined")
 	}
-	return conn.Ping(ctx)
+	globalConnectionManager.RLock()
+	defer globalConnectionManager.RUnlock()
+	meta, ok := globalConnectionManager.connectionPool[id]
+	if !ok {
+		return fmt.Errorf("connection %s not existed", id)
+	}
+	return meta.conn.Ping(ctx)
 }
 
-func GetNameConnection(selId string) (modules.Connection, error) {
-	if selId == "" {
+func FetchConnection(ctx api.StreamContext, id, typ string, props map[string]interface{}) (modules.Connection, error) {
+	if id == "" {
 		return nil, fmt.Errorf("connection id should be defined")
 	}
-	globalConnectionManager.RLock()
-	defer globalConnectionManager.RUnlock()
-	meta, ok := globalConnectionManager.connectionPool[selId]
+	selID := extractSelID(props)
+	if len(selID) < 1 {
+		return CreateNonStoredConnection(ctx, id, typ, props)
+	}
+	globalConnectionManager.Lock()
+	defer globalConnectionManager.Unlock()
+	return attachConnection(selID)
+}
+
+func attachConnection(id string) (modules.Connection, error) {
+	if id == "" {
+		return nil, fmt.Errorf("connection id should be defined")
+	}
+	meta, ok := globalConnectionManager.connectionPool[id]
 	if !ok {
-		return nil, fmt.Errorf("connection %s not existed", selId)
+		return nil, fmt.Errorf("connection %s not existed", id)
 	}
+	meta.refCount++
 	return meta.conn, nil
 }
 
+func DetachConnection(ctx api.StreamContext, id string, props map[string]interface{}) error {
+	if id == "" {
+		return fmt.Errorf("connection id should be defined")
+	}
+	globalConnectionManager.Lock()
+	defer globalConnectionManager.Unlock()
+	selID := extractSelID(props)
+	if len(selID) < 1 {
+		return detachConnection(ctx, id, true)
+	}
+	return detachConnection(ctx, selID, false)
+}
+
+func detachConnection(ctx api.StreamContext, id string, remove bool) error {
+	meta, ok := globalConnectionManager.connectionPool[id]
+	if !ok {
+		return nil
+	}
+	if remove {
+		conn := meta.conn
+		conn.Close(ctx)
+		delete(globalConnectionManager.connectionPool, id)
+		return nil
+	}
+	meta.refCount--
+	return nil
+}
+
 func CreateNamedConnection(ctx api.StreamContext, id, typ string, props map[string]any) (modules.Connection, error) {
 	if id == "" || typ == "" {
 		return nil, fmt.Errorf("connection id and type should be defined")
@@ -113,12 +168,12 @@ func CreateNamedConnection(ctx api.StreamContext, id, typ string, props map[stri
 	if ok {
 		return nil, fmt.Errorf("connection %v already been created", id)
 	}
-	meta := ConnectionMeta{
+	meta := &ConnectionMeta{
 		ID:    id,
 		Typ:   typ,
 		Props: props,
 	}
-	if err := storeConnection(typ, id, props); err != nil {
+	if err := storeConnectionMeta(typ, id, props); err != nil {
 		return nil, err
 	}
 	conn, err := createNamedConnection(ctx, meta)
@@ -143,7 +198,7 @@ func CreateNonStoredConnection(ctx api.StreamContext, id, typ string, props map[
 	if ok {
 		return nil, fmt.Errorf("connection %v already been created", id)
 	}
-	meta := ConnectionMeta{
+	meta := &ConnectionMeta{
 		ID:    id,
 		Typ:   typ,
 		Props: props,
@@ -157,25 +212,9 @@ func CreateNonStoredConnection(ctx api.StreamContext, id, typ string, props map[
 	return conn, nil
 }
 
-func DropNonStoredConnection(ctx api.StreamContext, selId string) error {
-	if selId == "" {
-		return fmt.Errorf("connection id should be defined")
-	}
-	globalConnectionManager.Lock()
-	defer globalConnectionManager.Unlock()
-	meta, ok := globalConnectionManager.connectionPool[selId]
-	if !ok {
-		return nil
-	}
-	conn := meta.conn
-	conn.Close(ctx)
-	delete(globalConnectionManager.connectionPool, selId)
-	return nil
-}
-
 var mockErr = true
 
-func createNamedConnection(ctx api.StreamContext, meta ConnectionMeta) (modules.Connection, error) {
+func createNamedConnection(ctx api.StreamContext, meta *ConnectionMeta) (modules.Connection, error) {
 	var conn modules.Connection
 	var err error
 	connRegister, ok := modules.ConnectionRegister[strings.ToLower(meta.Typ)]
@@ -183,7 +222,7 @@ func createNamedConnection(ctx api.StreamContext, meta ConnectionMeta) (modules.
 		return nil, fmt.Errorf("unknown connection type")
 	}
 	err = backoff.Retry(func() error {
-		conn, err = connRegister(ctx, meta.ID, meta.Props)
+		conn, err = connRegister(ctx, meta.Props)
 		failpoint.Inject("createConnectionErr", func() {
 			if mockErr {
 				err = errorx.NewIOErr("createConnectionErr")
@@ -215,15 +254,14 @@ func DropNameConnection(ctx api.StreamContext, selId string) error {
 		}
 		return nil
 	}
-	conn := meta.conn
-	if conn.Ref(ctx) > 0 {
+	if meta.refCount > 0 {
 		return fmt.Errorf("connection %s can't be dropped due to reference", selId)
 	}
 	err := dropConnectionStore(meta.Typ, selId)
 	if err != nil {
 		return fmt.Errorf("drop connection %s failed, err:%v", selId, err)
 	}
-	conn.Close(ctx)
+	meta.conn.Close(ctx)
 	delete(globalConnectionManager.connectionPool, selId)
 	return nil
 }
@@ -238,7 +276,7 @@ func InitConnectionManager4Test() error {
 
 func InitConnectionManager() {
 	globalConnectionManager = &ConnectionManager{
-		connectionPool: make(map[string]ConnectionMeta),
+		connectionPool: make(map[string]*ConnectionMeta),
 		failConnection: make(map[string]string),
 	}
 	if conf.IsTesting {
@@ -259,7 +297,7 @@ func ReloadConnection() error {
 		}
 		typ := names[1]
 		id := names[2]
-		meta := ConnectionMeta{
+		meta := &ConnectionMeta{
 			ID:    id,
 			Typ:   typ,
 			Props: props,
@@ -278,15 +316,16 @@ func ReloadConnection() error {
 
 type ConnectionManager struct {
 	sync.RWMutex
-	connectionPool map[string]ConnectionMeta
+	connectionPool map[string]*ConnectionMeta
 	failConnection map[string]string
 }
 
 type ConnectionMeta struct {
-	ID    string             `json:"id"`
-	Typ   string             `json:"typ"`
-	Props map[string]any     `json:"props"`
-	conn  modules.Connection `json:"-"`
+	ID       string             `json:"id"`
+	Typ      string             `json:"typ"`
+	Props    map[string]any     `json:"props"`
+	conn     modules.Connection `json:"-"`
+	refCount int                `json:"-"`
 }
 
 func NewExponentialBackOff() *backoff.ExponentialBackOff {
@@ -313,3 +352,18 @@ type ConnectionStatus struct {
 	Status string
 	ErrMsg string
 }
+
+func extractSelID(props map[string]interface{}) string {
+	if len(props) < 1 {
+		return ""
+	}
+	v, ok := props["connectionSelector"]
+	if !ok {
+		return ""
+	}
+	id, ok := v.(string)
+	if !ok {
+		return ""
+	}
+	return id
+}
diff --git a/pkg/modules/connection.go b/pkg/modules/connection.go
index e693777b..d2e1fb13 100644
--- a/pkg/modules/connection.go
+++ b/pkg/modules/connection.go
@@ -19,13 +19,10 @@ import "github.com/lf-edge/ekuiper/contract/v2/api"
 type Connection interface {
 	Ping(ctx api.StreamContext) error
 	Close(ctx api.StreamContext)
-	Attach(ctx api.StreamContext)
 	DetachSub(ctx api.StreamContext, props map[string]any)
-	DetachPub(ctx api.StreamContext, props map[string]any)
-	Ref(ctx api.StreamContext) int
 }
 
-type ConnectionProvider func(ctx api.StreamContext, id string, props map[string]any) (Connection, error)
+type ConnectionProvider func(ctx api.StreamContext, props map[string]any) (Connection, error)
 
 var ConnectionRegister map[string]ConnectionProvider
 


Test patch:
diff --git a/internal/converter/json/converter_test.go b/internal/converter/json/converter_test.go
index 6124111d..3d012381 100644
--- a/internal/converter/json/converter_test.go
+++ b/internal/converter/json/converter_test.go
@@ -22,6 +22,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
@@ -657,3 +658,53 @@ func TestSchemaless(t *testing.T) {
 		require.Equal(t, tc.expect, v)
 	}
 }
+
+func TestIssue(t *testing.T) {
+	originSchema := map[string]*ast.JsonStreamField{
+		"results": nil,
+	}
+	f := NewFastJsonConverter(originSchema)
+	data := `{
+    "results": [
+        {
+            "location": {
+                "id": "WTMKQ069CCJ7",
+                "name": "杭州",
+                "country": "CN",
+                "path": "杭州,杭州,浙江,中国",
+                "timezone": "Asia/Shanghai",
+                "timezone_offset": "+08:00"
+            },
+            "now": {
+                "text": "多云",
+                "code": "4",
+                "temperature": "31",
+                "feels_like": "36",
+                "pressure": "997",
+                "humidity": "58",
+                "visibility": "9.7",
+                "wind_direction": "东",
+                "wind_direction_degree": "87",
+                "wind_speed": "14.0",
+                "wind_scale": "3",
+                "clouds": "100",
+                "dew_point": ""
+            },
+            "last_update": "2024-06-13T16:37:50+08:00"
+        }
+    ]
+}`
+	m, err := f.Decode(context.Background(), []byte(data))
+	require.NoError(t, err)
+	expected := make(map[string]interface{})
+	json.Unmarshal([]byte(data), &expected)
+	require.Equal(t, expected, m)
+
+	schmema2 := map[string]*ast.JsonStreamField{
+		"others": nil,
+	}
+	f2 := NewFastJsonConverter(schmema2)
+	m, err = f2.Decode(context.Background(), []byte(data))
+	require.NoError(t, err)
+	require.Len(t, m, 0)
+}
diff --git a/internal/io/mqtt/sink_test.go b/internal/io/mqtt/sink_test.go
index 487562b0..36694043 100644
--- a/internal/io/mqtt/sink_test.go
+++ b/internal/io/mqtt/sink_test.go
@@ -22,8 +22,8 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
diff --git a/internal/io/mqtt/source_sink_test.go b/internal/io/mqtt/source_sink_test.go
index f5890368..c4f4be3c 100644
--- a/internal/io/mqtt/source_sink_test.go
+++ b/internal/io/mqtt/source_sink_test.go
@@ -22,10 +22,10 @@ import (
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/topotest/mockclock"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	"github.com/lf-edge/ekuiper/v2/pkg/mock"
 	"github.com/lf-edge/ekuiper/v2/pkg/model"
 )
diff --git a/internal/io/mqtt/source_test.go b/internal/io/mqtt/source_test.go
index e347e42f..b9910b1b 100644
--- a/internal/io/mqtt/source_test.go
+++ b/internal/io/mqtt/source_test.go
@@ -22,10 +22,10 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/io/mqtt/client"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
diff --git a/internal/server/rest_test.go b/internal/server/rest_test.go
index 29162df0..06786809 100644
--- a/internal/server/rest_test.go
+++ b/internal/server/rest_test.go
@@ -32,13 +32,13 @@ import (
 	"github.com/stretchr/testify/suite"
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/meta"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/v2/internal/processor"
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/rule"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 )
 
diff --git a/internal/io/connection/connection_pool_test.go b/pkg/connection/connection_pool_test.go
similarity index 71%
rename from internal/io/connection/connection_pool_test.go
rename to pkg/connection/connection_pool_test.go
index f9019bb8..4b9f845c 100644
--- a/internal/io/connection/connection_pool_test.go
+++ b/pkg/connection/connection_pool_test.go
@@ -35,37 +35,52 @@ func TestConnection(t *testing.T) {
 	require.NoError(t, err)
 	require.NotNil(t, conn)
 	require.NoError(t, conn.Ping(ctx))
+	require.Equal(t, 0, GetConnectionRef("id1"))
 	_, err = CreateNamedConnection(ctx, "id1", "mock", nil)
 	require.Error(t, err)
-	require.Equal(t, 0, conn.Ref(ctx))
-	conn.Attach(ctx)
-	require.Equal(t, 1, conn.Ref(ctx))
-	conn.Attach(ctx)
-	require.Equal(t, 2, conn.Ref(ctx))
-	conn.DetachPub(ctx, nil)
-	require.Equal(t, 1, conn.Ref(ctx))
+	attachConnection("id1")
+	require.Equal(t, 1, GetConnectionRef("id1"))
+	attachConnection("id1")
+	require.Equal(t, 2, GetConnectionRef("id1"))
+	detachConnection(ctx, "id1", false)
+	require.Equal(t, 1, GetConnectionRef("id1"))
 	err = DropNameConnection(ctx, "id1")
 	require.Error(t, err)
-	conn2, err := GetNameConnection("id1")
-	require.NoError(t, err)
-	require.NotNil(t, conn2)
-	conn.DetachSub(ctx, nil)
-	require.Equal(t, 0, conn.Ref(ctx))
+	detachConnection(ctx, "id1", false)
+	require.Equal(t, 0, GetConnectionRef("id1"))
 	err = DropNameConnection(ctx, "id1")
 	require.NoError(t, err)
 	err = DropNameConnection(ctx, "id1")
 	require.NoError(t, err)
-	conn3, err := GetNameConnection("id1")
+	conn3, err := attachConnection("id1")
 	require.Error(t, err)
 	require.Nil(t, conn3)
 
+	conn, err = CreateNamedConnection(ctx, "id2", "mock", nil)
+	require.NoError(t, err)
+	require.NotNil(t, conn)
+
+	conn, err = FetchConnection(ctx, "2222", "mock", map[string]interface{}{"connectionSelector": "id2"})
+	require.NoError(t, err)
+	require.NotNil(t, conn)
+
+	require.Equal(t, 1, GetConnectionRef("id2"))
+}
+
+func TestConnectionErr(t *testing.T) {
+	dataDir, err := conf.GetDataLoc()
+	require.NoError(t, err)
+	require.NoError(t, store.SetupDefault(dataDir))
+	require.NoError(t, InitConnectionManager4Test())
+	ctx := context.Background()
+
 	_, err = CreateNamedConnection(ctx, "", "mock", nil)
 	require.Error(t, err)
 	err = DropNameConnection(ctx, "")
 	require.Error(t, err)
 	_, err = CreateNamedConnection(ctx, "12", "unknown", nil)
 	require.Error(t, err)
-	_, err = GetNameConnection("")
+	_, err = attachConnection("")
 	require.Error(t, err)
 	err = PingConnection(ctx, "")
 	require.Error(t, err)
@@ -77,41 +92,33 @@ func TestConnection(t *testing.T) {
 	require.NotNil(t, conn4)
 	_, err = CreateNonStoredConnection(ctx, "id2", "mock", nil)
 	require.Error(t, err)
-	err = DropNonStoredConnection(ctx, "")
+	err = DetachConnection(ctx, "", nil)
 	require.Error(t, err)
-	err = DropNonStoredConnection(ctx, "nonexists")
-	require.NoError(t, err)
-}
-
-func TestConnectionErr(t *testing.T) {
-	dataDir, err := conf.GetDataLoc()
+	err = DetachConnection(ctx, "nonexists", nil)
 	require.NoError(t, err)
-	require.NoError(t, store.SetupDefault(dataDir))
-	require.NoError(t, InitConnectionManager4Test())
-	ctx := context.Background()
 
-	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/io/connection/createConnectionErr", "return(true)")
-	conn, err := createNamedConnection(ctx, ConnectionMeta{
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/pkg/createConnectionErr", "return(true)")
+	conn, err := createNamedConnection(ctx, &ConnectionMeta{
 		ID:    "1",
 		Typ:   "mock",
 		Props: nil,
 	})
 	require.NoError(t, err)
 	require.NotNil(t, conn)
-	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/io/connection/createConnectionErr")
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/pkg/connection/createConnectionErr")
 
-	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/io/connection/storeConnectionErr", "return(true)")
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/pkg/connection/storeConnectionErr", "return(true)")
 	_, err = CreateNamedConnection(ctx, "qwe", "mock", nil)
 	require.Error(t, err)
-	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/io/connection/storeConnectionErr")
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/pkg/connection/storeConnectionErr")
 
 	_, err = CreateNamedConnection(ctx, "qwe", "mock", nil)
 	require.NoError(t, err)
 
-	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/io/connection/dropConnectionStoreErr", "return(true)")
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/pkg/connection/dropConnectionStoreErr", "return(true)")
 	err = DropNameConnection(ctx, "qwe")
 	require.Error(t, err)
-	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/io/connection/dropConnectionStoreErr")
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/pkg/connection/dropConnectionStoreErr")
 }
 
 func TestConnectionStatus(t *testing.T) {