Repository: lf-edge/ekuiper
Pull request #2867 (2024-05-22 09:14:38)

feat: support manage connection

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2867?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 `29.05405%` with `105 lines` in your changes are missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@3147d8c`). [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/2867?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [internal/io/connection/connection\_pool.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2867?src=pr&el=tree&filepath=internal%2Fio%2Fconnection%2Fconnection_pool.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vY29ubmVjdGlvbi9jb25uZWN0aW9uX3Bvb2wuZ28=) | 0.00% | [76 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2867?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/mqtt/client/client.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2867?src=pr&el=tree&filepath=internal%2Fio%2Fmqtt%2Fclient%2Fclient.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vbXF0dC9jbGllbnQvY2xpZW50Lmdv) | 70.45% | [11 Missing and 2 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2867?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/mqtt/sink.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2867?src=pr&el=tree&filepath=internal%2Fio%2Fmqtt%2Fsink.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vbXF0dC9zaW5rLmdv) | 41.67% | [5 Missing and 2 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2867?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/mqtt/source.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2867?src=pr&el=tree&filepath=internal%2Fio%2Fmqtt%2Fsource.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vbXF0dC9zb3VyY2UuZ28=) | 53.85% | [3 Missing and 3 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2867?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/server.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2867?src=pr&el=tree&filepath=internal%2Fserver%2Fserver.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL3NlcnZlci5nbw==) | 0.00% | [3 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2867?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    #2867   +/-   ##
============================================
  Coverage             ?   68.25%           
============================================
  Files                ?      273           
  Lines                ?    26056           
  Branches             ?        0           
============================================
  Hits                 ?    17783           
  Misses               ?     6663           
  Partials             ?     1610           
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2867?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/io/connection/connection_pool.go b/internal/io/connection/connection_pool.go
new file mode 100644
index 00000000..e829f880
--- /dev/null
+++ b/internal/io/connection/connection_pool.go
@@ -0,0 +1,258 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package connection
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+	"sync"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/kv"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+)
+
+var isTest bool
+
+func GetAllConnectionsID() []string {
+	globalConnectionManager.RLock()
+	defer globalConnectionManager.RUnlock()
+	ids := make([]string, 0)
+	for key := range globalConnectionManager.connectionPool {
+		ids = append(ids, key)
+	}
+	return ids
+}
+
+func PingConnection(ctx api.StreamContext, id string) error {
+	conn, err := GetNameConnection(id)
+	if err != nil {
+		return err
+	}
+	return conn.Ping(ctx)
+}
+
+func GetNameConnection(selId string) (modules.Connection, error) {
+	if selId == "" {
+		return nil, fmt.Errorf("connection id should be defined")
+	}
+	globalConnectionManager.RLock()
+	defer globalConnectionManager.RUnlock()
+	meta, ok := globalConnectionManager.connectionPool[selId]
+	if !ok {
+		return nil, fmt.Errorf("connection %s not existed", selId)
+	}
+	return meta.conn, 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")
+	}
+	globalConnectionManager.Lock()
+	defer globalConnectionManager.Unlock()
+	_, ok := globalConnectionManager.connectionPool[id]
+	if ok {
+		return nil, fmt.Errorf("connection %v already been created", id)
+	}
+	meta := ConnectionMeta{
+		ID:    id,
+		Typ:   typ,
+		Props: props,
+	}
+	if !isTest {
+		b, err := json.Marshal(meta)
+		if err != nil {
+			return nil, err
+		}
+		if err := globalConnectionManager.store.Set(id, string(b)); err != nil {
+			return nil, err
+		}
+	}
+	conn, err := createNamedConnection(ctx, meta)
+	if err != nil {
+		return nil, err
+	}
+	meta.conn = conn
+	globalConnectionManager.connectionPool[id] = meta
+	return conn, nil
+}
+
+func CreateNonStoredConnection(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")
+	}
+	globalConnectionManager.Lock()
+	defer globalConnectionManager.Unlock()
+	_, ok := globalConnectionManager.connectionPool[id]
+	if ok {
+		return nil, fmt.Errorf("connection %v already been created", id)
+	}
+	meta := ConnectionMeta{
+		ID:    id,
+		Typ:   typ,
+		Props: props,
+	}
+	conn, err := createNamedConnection(ctx, meta)
+	if err != nil {
+		return nil, err
+	}
+	meta.conn = conn
+	globalConnectionManager.connectionPool[id] = meta
+	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
+}
+
+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)]
+	if !ok {
+		return nil, fmt.Errorf("unknown connection type")
+	}
+	conn, err = connRegister(ctx, meta.ID, meta.Props)
+	if err != nil {
+		return nil, err
+	}
+	return conn, nil
+}
+
+func DropNameConnection(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
+	if conn.Ref(ctx) > 0 {
+		return fmt.Errorf("connection %s can't be dropped due to reference", selId)
+	}
+	if !isTest {
+		err := globalConnectionManager.store.Delete(selId)
+		if err != nil {
+			return fmt.Errorf("drop connection %s failed, err:%v", selId, err)
+		}
+	}
+	conn.Close(ctx)
+	delete(globalConnectionManager.connectionPool, selId)
+	return nil
+}
+
+var globalConnectionManager *ConnectionManager
+
+func InitConnectionManagerInTest() {
+	isTest = true
+	InitConnectionManager()
+}
+
+func InitConnectionManager() error {
+	globalConnectionManager = &ConnectionManager{
+		connectionPool: make(map[string]ConnectionMeta),
+	}
+	if !isTest {
+		globalConnectionManager.store, _ = store.GetKV("connectionMeta")
+		kvs, _ := globalConnectionManager.store.All()
+		for connectionID, raw := range kvs {
+			meta := ConnectionMeta{}
+			err := json.Unmarshal([]byte(raw), &meta)
+			if err != nil {
+				return fmt.Errorf("initialize connection:%v failed, err:%v", connectionID, err)
+			}
+			conn, err := createNamedConnection(context.Background(), meta)
+			if err != nil {
+				return fmt.Errorf("initialize connection:%v failed, err:%v", connectionID, err)
+			}
+			meta.conn = conn
+			globalConnectionManager.connectionPool[connectionID] = meta
+		}
+	}
+	return nil
+}
+
+type ConnectionManager struct {
+	sync.RWMutex
+	store          kv.KeyValue
+	connectionPool map[string]ConnectionMeta
+}
+
+type ConnectionMeta struct {
+	ID    string             `json:"id"`
+	Typ   string             `json:"typ"`
+	Props map[string]any     `json:"props"`
+	conn  modules.Connection `json:"-"`
+}
+
+type mockConnection struct {
+	id  string
+	ref int
+}
+
+func (m *mockConnection) Ping(ctx api.StreamContext) error {
+	return nil
+}
+
+func (m *mockConnection) Close(ctx api.StreamContext) {
+	return
+}
+
+func (m *mockConnection) Attach(ctx api.StreamContext) {
+	m.ref++
+	return
+}
+
+func (m *mockConnection) DetachSub(ctx api.StreamContext, props map[string]any) {
+	m.ref--
+	return
+}
+
+func (m *mockConnection) DetachPub(ctx api.StreamContext, props map[string]any) {
+	m.ref--
+	return
+}
+
+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}
+	return m, nil
+}
+
+func init() {
+	modules.ConnectionRegister["mock"] = CreateMockConnection
+}
diff --git a/internal/io/file/reader/csv.go b/internal/io/file/reader/csv.go
new file mode 100644
index 00000000..27e0bf75
--- /dev/null
+++ b/internal/io/file/reader/csv.go
@@ -0,0 +1,108 @@
+package reader
+
+import (
+	"encoding/csv"
+	"errors"
+	"fmt"
+	"io"
+	"strconv"
+	"strings"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+)
+
+func init() {
+	modules.RegisterFileStreamReader("csv", func(ctx api.StreamContext) modules.FileStreamReader {
+		return &CsvReader{}
+	})
+}
+
+type csvConf struct {
+	HasHeader bool     `json:"hasHeader"`
+	Columns   []string `json:"columns"`
+	Delimiter string   `json:"delimiter"`
+}
+
+type CsvReader struct {
+	csvR   *csv.Reader
+	config *csvConf
+
+	cols []string
+}
+
+func (r *CsvReader) Provision(ctx api.StreamContext, props map[string]any) error {
+	c := &csvConf{
+		Delimiter: ",",
+	}
+	e := cast.MapToStruct(props, c)
+	if e != nil {
+		return e
+	}
+	if c.Delimiter == "" {
+		return errors.New("invalid empty delimiter")
+	}
+	r.config = c
+	return nil
+}
+
+func (r *CsvReader) Bind(ctx api.StreamContext, fileStream io.Reader) error {
+	cr := csv.NewReader(fileStream)
+	cr.Comma = rune(r.config.Delimiter[0])
+	cr.TrimLeadingSpace = true
+	cr.FieldsPerRecord = -1
+	cols := r.config.Columns
+	if r.config.HasHeader {
+		var err error
+		ctx.GetLogger().Debug("Has header")
+		cols, err = cr.Read()
+		if err == io.EOF {
+			return fmt.Errorf("header not found")
+		}
+		if err != nil {
+			ctx.GetLogger().Warnf("Read file %s encounter error: %v", "fs.file", err)
+			return err
+		}
+		ctx.GetLogger().Debugf("Got header %v", cols)
+	}
+
+	r.csvR = cr
+	r.cols = cols
+	return nil
+}
+
+func (r *CsvReader) Read(ctx api.StreamContext) (any, error) {
+	record, err := r.csvR.Read()
+	if err == io.EOF {
+		return nil, err
+	}
+	if err != nil {
+		ctx.GetLogger().Warnf("Read file %s encounter error: %v", "fs.file", err)
+		return nil, err
+	}
+	ctx.GetLogger().Debugf("Read" + strings.Join(record, ","))
+
+	var m map[string]interface{}
+	if r.cols == nil {
+		m = make(map[string]interface{}, len(record))
+		for i, v := range record {
+			m["cols"+strconv.Itoa(i)] = v
+		}
+	} else {
+		m = make(map[string]interface{}, len(r.cols))
+		for i, v := range r.cols {
+			m[v] = record[i]
+		}
+	}
+
+	return m, nil
+}
+
+func (r *CsvReader) IsBytesReader() bool {
+	return false
+}
+
+func (r *CsvReader) Close(ctx api.StreamContext) error {
+	return nil
+}
diff --git a/internal/io/file/reader/lines.go b/internal/io/file/reader/lines.go
new file mode 100644
index 00000000..5c437cf8
--- /dev/null
+++ b/internal/io/file/reader/lines.go
@@ -0,0 +1,46 @@
+package reader
+
+import (
+	"bufio"
+	"io"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+)
+
+func init() {
+	modules.RegisterFileStreamReader("lines", func(ctx api.StreamContext) modules.FileStreamReader {
+		return &LinesReader{}
+	})
+}
+
+type LinesReader struct {
+	scanner *bufio.Scanner
+}
+
+func (r *LinesReader) Provision(ctx api.StreamContext, props map[string]any) error {
+	return nil
+}
+
+func (r *LinesReader) Bind(ctx api.StreamContext, fileStream io.Reader) error {
+	scanner := bufio.NewScanner(fileStream)
+	scanner.Split(bufio.ScanLines)
+	r.scanner = scanner
+	return nil
+}
+
+func (r *LinesReader) Read(ctx api.StreamContext) (any, error) {
+	succ := r.scanner.Scan()
+	if !succ {
+		return nil, io.EOF
+	}
+	return r.scanner.Bytes(), nil
+}
+
+func (r *LinesReader) IsBytesReader() bool {
+	return true
+}
+
+func (r *LinesReader) Close(ctx api.StreamContext) error {
+	return nil
+}
diff --git a/internal/io/file/source.go b/internal/io/file/source.go
index 25483b3c..1118d351 100644
--- a/internal/io/file/source.go
+++ b/internal/io/file/source.go
@@ -26,13 +26,16 @@ 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/file/reader"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/model"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
 type SourceConfig struct {
 	FileName         string        `json:"datasource"`
-	FileType         FileType      `json:"fileType"`
+	FileType         string        `json:"fileType"`
 	Path             string        `json:"path"`
 	Interval         time.Duration `json:"interval"`
 	IsTable          bool          `json:"isTable"`
@@ -42,28 +45,25 @@ type SourceConfig struct {
 	MoveTo           string        `json:"moveTo"`
 	IgnoreStartLines int           `json:"ignoreStartLines"`
 	IgnoreEndLines   int           `json:"ignoreEndLines"`
-	// For csv only
-	HasHeader bool     `json:"hasHeader"`
-	Columns   []string `json:"columns"`
-	Delimiter string   `json:"delimiter"`
 	// Only use for planning
 	Decompression string `json:"decompression"`
 }
 
 // Source load data from file system.
-// If file type is lines, it reads line by line.
+// Depending on file types, it may read line by line like lines, csv.
 // Otherwise, it reads the file as a whole and send to company reader node to read and split.
-// The planner need to plan according to the file type.(currently, it is hardcoded to deal with file source only)
+// The planner need to plan according to the file type.
 type Source struct {
 	file   string
 	isDir  bool
 	config *SourceConfig
+	reader modules.FileStreamReader
 	eof    api.EOFIngest
 }
 
 func (fs *Source) Provision(ctx api.StreamContext, props map[string]any) error {
 	cfg := &SourceConfig{
-		FileType: JSON_TYPE,
+		FileType: "json",
 	}
 	err := cast.MapToStruct(props, cfg)
 	if err != nil {
@@ -72,8 +72,15 @@ func (fs *Source) Provision(ctx api.StreamContext, props map[string]any) error {
 	if cfg.FileType == "" {
 		return errors.New("missing or invalid property fileType, must be 'json'")
 	}
-	if _, ok := fileTypes[cfg.FileType]; !ok {
-		return fmt.Errorf("invalid property fileType: %s", cfg.FileType)
+	reader, ok := modules.GetFileStreamReader(ctx, cfg.FileType)
+	if ok {
+		err = reader.Provision(ctx, props)
+		if err != nil {
+			return err
+		}
+		fs.reader = reader
+	} else {
+		ctx.GetLogger().Warnf("file type %s is not stream reader, will send out the whole file", cfg.FileType)
 	}
 	if cfg.Path == "" {
 		return errors.New("missing property Path")
@@ -125,10 +132,6 @@ func (fs *Source) Provision(ctx api.StreamContext, props map[string]any) error {
 			}
 		}
 	}
-	if cfg.Delimiter == "" {
-		cfg.Delimiter = ","
-	}
-
 	fs.config = cfg
 	return nil
 }
@@ -138,10 +141,10 @@ func (fs *Source) Connect(ctx api.StreamContext) error {
 	return nil
 }
 
-// Subscribe file source always ingest bytes
+// Subscribe file source may ingest bytes or tuple
 // For stream source, it ingest one line
 // For batch source, it ingest the whole file, thus it need a reader node to coordinate and read the content into lines/array
-func (fs *Source) Subscribe(ctx api.StreamContext, ingest api.BytesIngest, ingestError api.ErrorIngest) error {
+func (fs *Source) Subscribe(ctx api.StreamContext, ingest api.TupleIngest, ingestError api.ErrorIngest) error {
 	fs.Load(ctx, ingest, ingestError)
 	if fs.config.Interval > 0 {
 		ticker := time.NewTicker(fs.config.Interval)
@@ -175,7 +178,7 @@ func (fs *Source) Close(ctx api.StreamContext) error {
 	return nil
 }
 
-func (fs *Source) Load(ctx api.StreamContext, ingest api.BytesIngest, ingestError api.ErrorIngest) {
+func (fs *Source) Load(ctx api.StreamContext, ingest api.TupleIngest, ingestError api.ErrorIngest) {
 	if fs.isDir {
 		ctx.GetLogger().Debugf("Monitor dir %s", fs.file)
 		entries, err := os.ReadDir(fs.file)
@@ -220,7 +223,7 @@ func (fs *Source) Load(ctx api.StreamContext, ingest api.BytesIngest, ingestErro
 	ctx.GetLogger().Debug("All tuples sent")
 }
 
-func (fs *Source) parseFile(ctx api.StreamContext, file string, ingest api.BytesIngest, ingestError api.ErrorIngest) {
+func (fs *Source) parseFile(ctx api.StreamContext, file string, ingest api.TupleIngest, ingestError api.ErrorIngest) {
 	var (
 		err error
 		r   io.Reader
@@ -241,25 +244,25 @@ func (fs *Source) parseFile(ctx api.StreamContext, file string, ingest api.Bytes
 	}
 	meta := map[string]any{"file": file}
 	// Read line or read all
-	switch fs.config.FileType {
-	case LINES_TYPE: // read line by line
-		scanner := bufio.NewScanner(r)
-		scanner.Split(bufio.ScanLines)
-		for scanner.Scan() {
+	if fs.reader != nil {
+		err = fs.reader.Bind(ctx, r)
+		if err != nil {
+			ingestError(ctx, err)
+			return
+		}
+		for {
+			line, err := fs.reader.Read(ctx)
+			if err != nil {
+				break
+			}
 			rcvTime := timex.GetNow()
-			fmt.Println("received", rcvTime.UnixMilli())
-			line := scanner.Bytes()
 			ingest(ctx, line, meta, rcvTime)
 			if fs.config.SendInterval > 0 {
 				time.Sleep(fs.config.SendInterval)
 			}
 		}
-		// Check for any errors during scanning
-		if err := scanner.Err(); err != nil {
-			ingestError(ctx, err)
-			return
-		}
-	default: // read all
+		_ = fs.reader.Close(ctx)
+	} else {
 		rcvTime := timex.GetNow()
 		content, err := os.ReadFile(file)
 		if err != nil {
@@ -270,6 +273,7 @@ func (fs *Source) parseFile(ctx api.StreamContext, file string, ingest api.Bytes
 			ingest(ctx, content, meta, rcvTime)
 		}
 	}
+
 	ctx.GetLogger().Debugf("Finish loading from file %s", file)
 	switch fs.config.ActionAfterRead {
 	case 1:
@@ -337,11 +341,28 @@ func ignoreLines(ctx api.StreamContext, reader io.Reader, ignoreStartLines int,
 	return r
 }
 
+func (fs *Source) Info() (i model.NodeInfo) {
+	if fs.reader == nil { // output batch raw, so need encrypt/decompress as a whole and then decode as a whole
+		i.NeedBatchDecode = true
+		i.NeedDecode = true
+	} else if fs.reader.IsBytesReader() { // decrypt/decompress in scan and output raw
+		i.NeedDecode = true
+		i.HasCompress = true
+		i.HasInterval = true
+	} else { // decrypt/decompress in scan and output decoded tuple
+		i.HasCompress = true
+		i.HasInterval = true
+	}
+	return
+}
+
 func GetSource() api.Source {
 	return &Source{}
 }
 
 var (
-	_ api.BytesSource = &Source{}
+	// ingest possibly []byte and tuple
+	_ api.TupleSource = &Source{}
 	_ api.Bounded     = &Source{}
+	_ model.InfoNode  = &Source{}
 )
diff --git a/internal/io/mqtt/connection.go b/internal/io/mqtt/client/client.go
similarity index 78%
rename from internal/io/mqtt/connection.go
rename to internal/io/mqtt/client/client.go
index 95132e1f..de87a527 100644
--- a/internal/io/mqtt/connection.go
+++ b/internal/io/mqtt/client/client.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package mqtt
+package client
 
 import (
 	"crypto/tls"
@@ -25,12 +25,16 @@ import (
 	"github.com/google/uuid"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cert"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
 
+func init() {
+	modules.ConnectionRegister["mqtt"] = CreateConnection
+}
+
 type Connection struct {
 	pahoMqtt.Client
 	selId     string
@@ -90,27 +94,27 @@ func (conn *Connection) onReconnecting(_ pahoMqtt.Client, _ *pahoMqtt.ClientOpti
 }
 
 // Do not call this directly. Call connection pool Attach method to get the connection
-func (conn *Connection) attach() {
+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(topic string) bool {
+func (conn *Connection) DetachSub(ctx api.StreamContext, props map[string]any) {
+	conn.refCount.Add(-1)
+	topic, err := getTopicFromProps(props)
+	if err != nil {
+		return
+	}
 	delete(conn.subscriptions, topic)
 	conn.Client.Unsubscribe(topic)
-	if conn.refCount.Add(-1) == 0 {
-		go conn.Close()
-		return true
-	}
-	return false
 }
 
-func (conn *Connection) detachPub() bool {
-	if conn.refCount.Add(-1) == 0 {
-		go conn.Close()
-		return true
-	}
-	return false
+func (conn *Connection) DetachPub(ctx api.StreamContext, props map[string]any) {
+	conn.refCount.Add(-1)
 }
 
 func (conn *Connection) Subscribe(topic string, info *SubscriptionInfo) error {
@@ -119,11 +123,11 @@ func (conn *Connection) Subscribe(topic string, info *SubscriptionInfo) error {
 	return handleToken(token)
 }
 
-func (conn *Connection) Close() {
+func (conn *Connection) Close(ctx api.StreamContext) {
 	conn.Client.Disconnect(1)
 }
 
-func (conn *Connection) Ping() error {
+func (conn *Connection) Ping(ctx api.StreamContext) error {
 	if conn.Client.IsConnected() {
 		return nil
 	} else {
@@ -131,22 +135,13 @@ func (conn *Connection) Ping() error {
 	}
 }
 
+func CreateConnection(ctx api.StreamContext, selId string, props map[string]any) (modules.Connection, error) {
+	return CreateClient(ctx, selId, 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) {
-	if selId != "" {
-		selectCfg := &conf.ConSelector{
-			ConnSelectorStr: selId,
-		}
-		if err := selectCfg.Init(); err != nil {
-			return nil, err
-		}
-		cf, err := selectCfg.ReadCfgFromYaml()
-		if err != nil {
-			return nil, err
-		}
-		props = cf
-	}
-	c, err := validateConfig(props)
+	c, err := ValidateConfig(props)
 	if err != nil {
 		return nil, err
 	}
@@ -180,7 +175,9 @@ func CreateClient(ctx api.StreamContext, selId string, props map[string]any) (*C
 	}
 	ctx.GetLogger().Infof("new mqtt client created")
 	con.Client = cli
-	con.attach()
+	if len(selId) > 0 {
+		con.Attach(ctx)
+	}
 	return con, nil
 }
 
@@ -193,7 +190,7 @@ func handleToken(token pahoMqtt.Token) error {
 	return nil
 }
 
-func validateConfig(props map[string]any) (*ConnectionConfig, error) {
+func ValidateConfig(props map[string]any) (*ConnectionConfig, error) {
 	c := &ConnectionConfig{PVersion: "3.1.1"}
 	err := cast.MapToStruct(props, c)
 	if err != nil {
@@ -224,6 +221,22 @@ func validateConfig(props map[string]any) (*ConnectionConfig, error) {
 	return c, nil
 }
 
-func (conn *Connection) GetClientId() string {
-	return conn.selId
+func CreateAnonymousConnection(ctx api.StreamContext, props map[string]any) (*Connection, error) {
+	cli, err := CreateClient(ctx, "", props)
+	if err != nil {
+		return nil, err
+	}
+	return cli, nil
+}
+
+const (
+	dataSourceProp = "datasource"
+)
+
+func getTopicFromProps(props map[string]any) (string, error) {
+	v, ok := props[dataSourceProp]
+	if ok {
+		return v.(string), nil
+	}
+	return "", fmt.Errorf("topic or datasource not defined")
 }
diff --git a/internal/io/mqtt/connection_pool.go b/internal/io/mqtt/connection_pool.go
deleted file mode 100644
index e63e1258..00000000
--- a/internal/io/mqtt/connection_pool.go
+++ /dev/null
@@ -1,59 +0,0 @@
-// Copyright 2024 EMQ Technologies Co., Ltd.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package mqtt
-
-import (
-	"sync"
-
-	"github.com/lf-edge/ekuiper/contract/v2/api"
-)
-
-var (
-	connectionPool = make(map[string]*Connection)
-	lock           sync.RWMutex
-)
-
-func GetConnection(ctx api.StreamContext, selId string, props map[string]any) (*Connection, error) {
-	if selId == "" {
-		return CreateClient(ctx, "", props)
-	}
-	lock.Lock()
-	defer lock.Unlock()
-	if conn, ok := connectionPool[selId]; ok {
-		conn.attach()
-		return conn, nil
-	} else {
-		cli, err := CreateClient(ctx, selId, props)
-		if err != nil {
-			return nil, err
-		}
-		connectionPool[selId] = cli
-		return cli, nil
-	}
-}
-
-func DetachConnection(conn *Connection, selId string, subscribedTopic string) {
-	var closed bool
-	if subscribedTopic != "" {
-		closed = conn.detachSub(subscribedTopic)
-	} else {
-		closed = conn.detachPub()
-	}
-	lock.Lock()
-	defer lock.Unlock()
-	if _, ok := connectionPool[selId]; closed && ok {
-		delete(connectionPool, selId)
-	}
-}
diff --git a/internal/io/mqtt/sink.go b/internal/io/mqtt/sink.go
index 791afe4f..0384464a 100644
--- a/internal/io/mqtt/sink.go
+++ b/internal/io/mqtt/sink.go
@@ -19,6 +19,8 @@ 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"
 )
 
@@ -33,11 +35,11 @@ type AdConf struct {
 type Sink struct {
 	adconf *AdConf
 	config map[string]interface{}
-	cli    *Connection
+	cli    *client.Connection
 }
 
 func (ms *Sink) Provision(_ api.StreamContext, ps map[string]any) error {
-	_, err := validateConfig(ps)
+	_, err := client.ValidateConfig(ps)
 	if err != nil {
 		return err
 	}
@@ -62,7 +64,26 @@ 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")
-	cli, err := GetConnection(ctx, ms.adconf.SelId, ms.config)
+	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)
+	}
 	ms.cli = cli
 	return err
 }
@@ -90,7 +111,12 @@ 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 {
-		DetachConnection(ms.cli, ms.adconf.SelId, "")
+		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)
+		}
 	}
 	return nil
 }
diff --git a/internal/io/mqtt/source.go b/internal/io/mqtt/source.go
index 8c132bef..c446a7ce 100644
--- a/internal/io/mqtt/source.go
+++ b/internal/io/mqtt/source.go
@@ -20,6 +20,8 @@ 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/timex"
@@ -32,7 +34,7 @@ type SourceConnector struct {
 	cfg   *Conf
 	props map[string]any
 
-	cli *Connection
+	cli *client.Connection
 }
 
 type Conf struct {
@@ -50,7 +52,7 @@ func (ms *SourceConnector) Provision(ctx api.StreamContext, props map[string]any
 	if cfg.Topic == "" {
 		return fmt.Errorf("topic is required")
 	}
-	_, err = validateConfig(props)
+	_, err = client.ValidateConfig(props)
 	if err != nil {
 		return err
 	}
@@ -60,18 +62,37 @@ func (ms *SourceConnector) Provision(ctx api.StreamContext, props map[string]any
 	return nil
 }
 
-func (ms *SourceConnector) Ping(props map[string]interface{}) error {
-	cli, err := CreateClient(context.Background(), "", ms.props)
+func (ms *SourceConnector) Ping(ctx api.StreamContext, props map[string]interface{}) error {
+	cli, err := client.CreateAnonymousConnection(context.Background(), props)
 	if err != nil {
 		return err
 	}
-	defer cli.Close()
-	return cli.Ping()
+	defer cli.Close(ctx)
+	return cli.Ping(ctx)
 }
 
 func (ms *SourceConnector) Connect(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("Connecting to mqtt server")
-	cli, err := GetConnection(ctx, ms.cfg.SelId, ms.props)
+	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)
+	}
 	ms.cli = cli
 	return err
 }
@@ -79,7 +100,7 @@ func (ms *SourceConnector) Connect(ctx api.StreamContext) error {
 // Subscribe is a one time only operation for source. It connects to the mqtt broker and subscribe to the topic
 // Run open before subscribe
 func (ms *SourceConnector) Subscribe(ctx api.StreamContext, ingest api.BytesIngest, ingestError api.ErrorIngest) error {
-	return ms.cli.Subscribe(ms.tpc, &SubscriptionInfo{
+	return ms.cli.Subscribe(ms.tpc, &client.SubscriptionInfo{
 		Qos: byte(ms.cfg.Qos),
 		Handler: func(client pahoMqtt.Client, message pahoMqtt.Message) {
 			ms.onMessage(ctx, message, ingest)
@@ -105,7 +126,12 @@ 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 {
-		DetachConnection(ms.cli, ms.cfg.SelId, ms.tpc)
+		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)
+		}
 	}
 	return nil
 }
diff --git a/internal/server/connection.go b/internal/server/connection.go
new file mode 100644
index 00000000..ee60d7a2
--- /dev/null
+++ b/internal/server/connection.go
@@ -0,0 +1,89 @@
+// Copyright 2021-2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+
+	"github.com/gorilla/mux"
+
+	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
+)
+
+type ConnectionRequest struct {
+	ID    string                 `json:"id"`
+	Typ   string                 `json:"typ"`
+	Props map[string]interface{} `json:"props"`
+}
+
+func connectionsHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	switch r.Method {
+	case http.MethodPost:
+		body, err := io.ReadAll(r.Body)
+		if err != nil {
+			handleError(w, err, "Invalid body", logger)
+			return
+		}
+		req := &ConnectionRequest{}
+		if err := json.Unmarshal(body, req); err != nil {
+			handleError(w, err, "Invalid body", logger)
+			return
+		}
+		_, err = connection.CreateNamedConnection(context.Background(), req.ID, req.Typ, req.Props)
+		if err != nil {
+			handleError(w, err, "create connection failed", logger)
+			return
+		}
+		w.WriteHeader(http.StatusCreated)
+		w.Write([]byte("success"))
+	case http.MethodGet:
+		ids := connection.GetAllConnectionsID()
+		w.WriteHeader(http.StatusOK)
+		jsonResponse(ids, w, logger)
+	}
+}
+
+type ConnectionResponse struct {
+	ID  string `json:"id"`
+	Err string `json:"err"`
+}
+
+func connectionHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	id := mux.Vars(r)["id"]
+	switch r.Method {
+	case http.MethodGet:
+		res := &ConnectionResponse{
+			ID: id,
+		}
+		err := connection.PingConnection(context.Background(), id)
+		if err != nil {
+			res.Err = err.Error()
+		}
+		w.WriteHeader(http.StatusOK)
+		jsonResponse(res, w, logger)
+	case http.MethodDelete:
+		if err := connection.DropNameConnection(context.Background(), id); err != nil {
+			handleError(w, err, "drop connection failed", logger)
+			return
+		}
+		w.WriteHeader(http.StatusOK)
+		w.Write([]byte("success"))
+	}
+}
diff --git a/internal/server/rest.go b/internal/server/rest.go
index f505508f..6267e803 100644
--- a/internal/server/rest.go
+++ b/internal/server/rest.go
@@ -175,6 +175,8 @@ func createRestServer(ip string, port int, needToken bool) *http.Server {
 	r.HandleFunc("/data/export", configurationExportHandler).Methods(http.MethodGet, http.MethodPost)
 	r.HandleFunc("/data/import", configurationImportHandler).Methods(http.MethodPost)
 	r.HandleFunc("/data/import/status", configurationStatusHandler).Methods(http.MethodGet)
+	r.HandleFunc("/connections", connectionsHandler).Methods(http.MethodGet, http.MethodPost)
+	r.HandleFunc("/connection/{id}", connectionHandler).Methods(http.MethodGet, http.MethodDelete)
 	// r.HandleFunc("/connection/websocket", connectionHandler).Methods(http.MethodGet, http.MethodPost, http.MethodDelete)
 	// Register extended routes
 	for k, v := range components {
diff --git a/internal/server/server.go b/internal/server/server.go
index bff7b80a..f23c2395 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -35,6 +35,7 @@ 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"
@@ -150,6 +151,10 @@ func StartUp(Version string) {
 	keyedstate.InitKeyedStateKV()
 
 	meta2.InitYamlConfigManager()
+	if err := connection.InitConnectionManager(); err != nil {
+		panic(err)
+	}
+
 	ruleProcessor = processor.NewRuleProcessor()
 	streamProcessor = processor.NewStreamProcessor()
 	rulesetProcessor = processor.NewRulesetProcessor(ruleProcessor, streamProcessor)
diff --git a/internal/topo/node/concurrent.go b/internal/topo/node/concurrent.go
index b9cdfec3..f15131ba 100644
--- a/internal/topo/node/concurrent.go
+++ b/internal/topo/node/concurrent.go
@@ -15,6 +15,8 @@
 package node
 
 import (
+	"time"
+
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
 )
@@ -25,6 +27,10 @@ import (
 type workerFunc func(ctx api.StreamContext, item any) []any
 
 func runWithOrder(ctx api.StreamContext, node *defaultSinkNode, numWorkers int, wf workerFunc) {
+	runWithOrderAndInterval(ctx, node, numWorkers, wf, 0)
+}
+
+func runWithOrderAndInterval(ctx api.StreamContext, node *defaultSinkNode, numWorkers int, wf workerFunc, sendInterval time.Duration) {
 	workerChans := make([]chan any, numWorkers)
 	workerOutChans := make([]chan []any, numWorkers)
 	for i := range workerChans {
@@ -38,14 +44,14 @@ func runWithOrder(ctx api.StreamContext, node *defaultSinkNode, numWorkers int,
 	}
 	// start merger goroutine
 	output := make(chan any)
-	go merge(ctx, node, output, workerOutChans...)
+	go merge(ctx, node, sendInterval, output, workerOutChans...)
 
 	// Distribute input data to workers
 	distribute(ctx, node, numWorkers, workerChans)
 }
 
 // Merge multiple channels into one preserving the order
-func merge(ctx api.StreamContext, node *defaultSinkNode, output chan any, channels ...chan []any) {
+func merge(ctx api.StreamContext, node *defaultSinkNode, sendInterval time.Duration, output chan any, channels ...chan []any) {
 	defer close(output)
 	// Start a goroutine for each input channel
 	for {
@@ -64,6 +70,9 @@ func merge(ctx api.StreamContext, node *defaultSinkNode, output chan any, channe
 					default:
 						node.statManager.IncTotalRecordsOut()
 					}
+					if sendInterval > 0 {
+						time.Sleep(sendInterval)
+					}
 				}
 				node.statManager.IncTotalMessagesProcessed(1)
 			case <-ctx.Done():
diff --git a/internal/topo/node/decode_op.go b/internal/topo/node/decode_op.go
index 8d663085..0b3944b9 100644
--- a/internal/topo/node/decode_op.go
+++ b/internal/topo/node/decode_op.go
@@ -16,6 +16,7 @@ package node
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/converter"
@@ -23,14 +24,23 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/infra"
 	"github.com/lf-edge/ekuiper/v2/pkg/message"
 )
 
+// DecodeOp manages the format decoding (employ schema) and sending frequency (for batch decode, like a json array)
 type DecodeOp struct {
 	*defaultSinkNode
 	converter message.Converter
 	sLayer    *schemaLayer.SchemaLayer
+	// When receiving list, send them one by one, this is the sending interval between each
+	// Typically set by file source
+	sendInterval time.Duration
+}
+
+type dconf struct {
+	SendInterval time.Duration `json:"sendInterval"`
 }
 
 func (o *DecodeOp) AttachSchema(ctx api.StreamContext, dataSource string, schema map[string]*ast.JsonStreamField, isWildcard bool) {
@@ -57,7 +67,7 @@ func (o *DecodeOp) DetachSchema(ctx api.StreamContext, ruleId string) {
 	}
 }
 
-func NewDecodeOp(name, StreamName string, ruleId string, rOpt *def.RuleOption, options *ast.Options, isWildcard, isSchemaless bool, schema map[string]*ast.JsonStreamField) (*DecodeOp, error) {
+func NewDecodeOp(name, StreamName string, ruleId string, rOpt *def.RuleOption, options *ast.Options, isWildcard, isSchemaless bool, schema map[string]*ast.JsonStreamField, props map[string]any) (*DecodeOp, error) {
 	options.Schema = nil
 	options.IsWildCard = isWildcard
 	options.IsSchemaLess = isSchemaless
@@ -71,10 +81,16 @@ func NewDecodeOp(name, StreamName string, ruleId string, rOpt *def.RuleOption, o
 		msg := fmt.Sprintf("cannot get converter from format %s, schemaId %s: %v", options.FORMAT, options.SCHEMAID, err)
 		return nil, fmt.Errorf(msg)
 	}
+	dc := &dconf{}
+	e := cast.MapToStruct(props, dc)
+	if e != nil {
+		return nil, e
+	}
 	return &DecodeOp{
 		defaultSinkNode: newDefaultSinkNode(name, rOpt),
 		converter:       converterTool,
 		sLayer:          schemaLayer.NewSchemaLayer(ruleId, StreamName, schema, isWildcard),
+		sendInterval:    dc.SendInterval,
 	}, nil
 }
 
@@ -86,7 +102,7 @@ func (o *DecodeOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 			o.Close()
 		}()
 		err := infra.SafeRun(func() error {
-			runWithOrder(ctx, o.defaultSinkNode, o.concurrency, o.Worker)
+			runWithOrderAndInterval(ctx, o.defaultSinkNode, o.concurrency, o.Worker, o.sendInterval)
 			return nil
 		})
 		if err != nil {
diff --git a/internal/topo/node/source_node.go b/internal/topo/node/source_node.go
index fb3fe4bc..6b67f3fd 100644
--- a/internal/topo/node/source_node.go
+++ b/internal/topo/node/source_node.go
@@ -87,6 +87,10 @@ func (m *SourceNode) ingestAnyTuple(ctx api.StreamContext, data any, meta map[st
 		for _, mm := range mess {
 			m.ingestMap(mm, meta, ts)
 		}
+	// expected from file which send out any tuple type
+	case []byte:
+		tuple := &xsql.RawTuple{Emitter: m.name, Rawdata: mess, Timestamp: ts, Metadata: meta}
+		m.Broadcast(tuple)
 	// Source tuples are expected from memory
 	case *xsql.Tuple:
 		m.ingestTuple(mess, ts)
diff --git a/internal/topo/planner/planner_source.go b/internal/topo/planner/planner_source.go
index b8f3becc..798722f2 100644
--- a/internal/topo/planner/planner_source.go
+++ b/internal/topo/planner/planner_source.go
@@ -26,6 +26,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/topo/operator"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/model"
 )
 
 func transformSourceNode(ctx api.StreamContext, t *DataSourcePlan, mockSourcesProp map[string]map[string]any, ruleId string, options *def.RuleOption, index int) (node.DataSourceNode, []node.OperatorNode, int, error) {
@@ -100,8 +101,14 @@ func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, option
 		}
 		srcConnNode = srcSubtopo
 	}
-
-	_, isBytesSource := ss.(api.BytesSource)
+	// Make sure it is provisioned. Already done in NewSourceNode
+	info := checkByteSource(ss)
+	if info.HasCompress {
+		sp.Decompression = ""
+	}
+	if info.HasInterval {
+		delete(props, "sendInterval")
+	}
 
 	if err != nil {
 		return nil, nil, 0, err
@@ -110,7 +117,7 @@ func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, option
 	var ops []node.OperatorNode
 
 	if sp.Decompression != "" {
-		if isBytesSource {
+		if info.NeedBatchDecode {
 			dco, err := node.NewDecompressOp(fmt.Sprintf("%d_decompress", index), options, sp.Decompression)
 			if err != nil {
 				return nil, nil, 0, err
@@ -118,13 +125,13 @@ func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, option
 			index++
 			ops = append(ops, dco)
 		} else {
-			return nil, nil, 0, fmt.Errorf("source %s does not support decompression", t.name)
+			ctx.GetLogger().Warnf("source %s does not support decompression", t.name)
 		}
 	}
 
-	if isBytesSource {
+	if info.NeedDecode {
 		// Create the decode node
-		decodeNode, err := node.NewDecodeOp(fmt.Sprintf("%d_decoder", index), string(t.streamStmt.Name), ruleId, options, t.streamStmt.Options, t.isWildCard, t.isSchemaless, t.streamFields)
+		decodeNode, err := node.NewDecodeOp(fmt.Sprintf("%d_decoder", index), string(t.streamStmt.Name), ruleId, options, t.streamStmt.Options, t.isWildCard, t.isSchemaless, t.streamFields, props)
 		if err != nil {
 			return nil, nil, 0, err
 		}
@@ -155,3 +162,17 @@ func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, option
 	}
 	return srcConnNode, ops, 0, nil
 }
+
+func checkByteSource(ss api.Source) model.NodeInfo {
+	switch st := ss.(type) {
+	case model.InfoNode:
+		return st.Info()
+	case api.BytesSource:
+		return model.NodeInfo{
+			NeedDecode:      true,
+			NeedBatchDecode: true,
+		}
+	default:
+		return model.NodeInfo{}
+	}
+}
diff --git a/pkg/model/io.go b/pkg/model/io.go
index 38fc8f3a..fc23b255 100644
--- a/pkg/model/io.go
+++ b/pkg/model/io.go
@@ -25,3 +25,19 @@ import (
 type StreamWriter interface {
 	CreateWriter(ctx api.StreamContext, currWriter io.Writer, compression string, encryption string) (io.Writer, error)
 }
+
+type StreamReader interface {
+	CreateWriter(ctx api.StreamContext, currWriter io.Writer, compression string, encryption string) (io.Writer, error)
+}
+
+// InfoNode explain the node itself. Mainly used for planner to decide the split of source/sink
+type InfoNode interface {
+	Info() NodeInfo
+}
+
+type NodeInfo struct {
+	NeedDecode      bool
+	NeedBatchDecode bool // like decrypt, decompress as a whole
+	HasCompress     bool
+	HasInterval     bool
+}
diff --git a/pkg/modules/file_roll_hook.go b/pkg/modules/connection.go
similarity index 57%
rename from pkg/modules/file_roll_hook.go
rename to pkg/modules/connection.go
index 1546a71f..feab1007 100644
--- a/pkg/modules/file_roll_hook.go
+++ b/pkg/modules/connection.go
@@ -16,23 +16,19 @@ package modules
 
 import "github.com/lf-edge/ekuiper/contract/v2/api"
 
-type RollHook interface {
-	Provision(ctx api.StreamContext, props map[string]any) error
-	RollDone(ctx api.StreamContext, filePath string) error
-	api.Closable
+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 RollHookProvider func() RollHook
+type RegisterConnection func(ctx api.StreamContext, id string, props map[string]any) (Connection, error)
 
-var fileRollHooks = map[string]RollHookProvider{}
+var ConnectionRegister map[string]RegisterConnection
 
-func RegisterFileRollHook(name string, provider RollHookProvider) {
-	fileRollHooks[name] = provider
-}
-
-func GetFileRollHook(name string) (RollHook, bool) {
-	if p, ok := fileRollHooks[name]; ok {
-		return p(), ok
-	}
-	return nil, false
+func init() {
+	ConnectionRegister = map[string]RegisterConnection{}
 }
diff --git a/pkg/modules/file.go b/pkg/modules/file.go
new file mode 100644
index 00000000..d8623c0b
--- /dev/null
+++ b/pkg/modules/file.go
@@ -0,0 +1,71 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package modules
+
+import (
+	"io"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+)
+
+type RollHook interface {
+	Provision(ctx api.StreamContext, props map[string]any) error
+	RollDone(ctx api.StreamContext, filePath string) error
+	api.Closable
+}
+
+type RollHookProvider func() RollHook
+
+var fileRollHooks = map[string]RollHookProvider{}
+
+func RegisterFileRollHook(name string, provider RollHookProvider) {
+	fileRollHooks[name] = provider
+}
+
+func GetFileRollHook(name string) (RollHook, bool) {
+	if p, ok := fileRollHooks[name]; ok {
+		return p(), ok
+	}
+	return nil, false
+}
+
+// FileStreamReader reads a type of file line by line. Avoid to load the full file
+// If need to load full file, just extend converter to decode the full bytes
+type FileStreamReader interface {
+	// Provision Set up the static properties
+	Provision(ctx api.StreamContext, props map[string]any) error
+	// Bind set the file stream. Make sure the previous read has done
+	Bind(ctx api.StreamContext, fileStream io.Reader) error
+	// Read the next record. Returns EOF when the input has reached its end.
+	Read(ctx api.StreamContext) (any, error)
+	// IsBytesReader If is bytes reader, Read must return []byte, otherwise return map or []map
+	IsBytesReader() bool
+	api.Closable
+}
+
+type FileStreamReaderProvider func(ctx api.StreamContext) FileStreamReader
+
+var fileStreamReaders = map[string]FileStreamReaderProvider{}
+
+func RegisterFileStreamReader(name string, provider FileStreamReaderProvider) {
+	fileStreamReaders[name] = provider
+}
+
+func GetFileStreamReader(ctx api.StreamContext, name string) (FileStreamReader, bool) {
+	if p, ok := fileStreamReaders[name]; ok {
+		return p(ctx), true
+	}
+	return nil, false
+}


Test patch:
diff --git a/internal/io/connection/connection_pool_test.go b/internal/io/connection/connection_pool_test.go
new file mode 100644
index 00000000..dd0f4f9d
--- /dev/null
+++ b/internal/io/connection/connection_pool_test.go
@@ -0,0 +1,78 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package connection
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
+)
+
+func TestConnection(t *testing.T) {
+	InitConnectionManagerInTest()
+	ctx := context.Background()
+	conn, err := CreateNamedConnection(ctx, "id1", "mock", nil)
+	require.NoError(t, err)
+	require.NotNil(t, conn)
+	require.NoError(t, conn.Ping(ctx))
+	_, 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))
+	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))
+	err = DropNameConnection(ctx, "id1")
+	require.NoError(t, err)
+	err = DropNameConnection(ctx, "id1")
+	require.NoError(t, err)
+	conn3, err := GetNameConnection("id1")
+	require.Error(t, err)
+	require.Nil(t, conn3)
+
+	_, 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("")
+	require.Error(t, err)
+	err = PingConnection(ctx, "")
+	require.Error(t, err)
+	_, err = CreateNonStoredConnection(ctx, "", "mock", nil)
+	require.Error(t, err)
+
+	conn4, err := CreateNonStoredConnection(ctx, "id2", "mock", nil)
+	require.NoError(t, err)
+	require.NotNil(t, conn4)
+	_, err = CreateNonStoredConnection(ctx, "id2", "mock", nil)
+	require.Error(t, err)
+	err = DropNonStoredConnection(ctx, "")
+	require.Error(t, err)
+	err = DropNonStoredConnection(ctx, "nonexists")
+	require.NoError(t, err)
+}
diff --git a/internal/io/file/source_test.go b/internal/io/file/source_test.go
index 76402b3e..d09970e2 100644
--- a/internal/io/file/source_test.go
+++ b/internal/io/file/source_test.go
@@ -67,10 +67,9 @@ func TestSourceProvision(t *testing.T) {
 				"path":       path,
 			},
 			c: &SourceConfig{
-				FileName:  name,
-				Path:      path,
-				FileType:  JSON_TYPE,
-				Delimiter: ",",
+				FileName: name,
+				Path:     path,
+				FileType: string(JSON_TYPE),
 			},
 		},
 		{
@@ -92,13 +91,14 @@ func TestSourceProvision(t *testing.T) {
 			e: "missing or invalid property fileType, must be 'json'",
 		},
 		{
-			name: "invalid file type",
+			name: "invalid reader prop",
 			props: map[string]any{
 				"datasource": name,
 				"path":       path,
-				"fileType":   "aaa",
+				"fileType":   "csv",
+				"hasHeader":  "uvw",
 			},
-			e: "invalid property fileType: aaa",
+			e: "1 error(s) decoding:\n\n* 'hasHeader' expected type 'bool', got unconvertible type 'string', value: 'uvw'",
 		},
 		{
 			name: "missing path",
@@ -134,8 +134,7 @@ func TestSourceProvision(t *testing.T) {
 			c: &SourceConfig{
 				FileName:         "",
 				Path:             relPath,
-				FileType:         JSON_TYPE,
-				Delimiter:        ",",
+				FileType:         string(JSON_TYPE),
 				IgnoreStartLines: 0,
 				IgnoreEndLines:   0,
 			},
@@ -179,8 +178,7 @@ func TestSourceProvision(t *testing.T) {
 			c: &SourceConfig{
 				FileName:        name,
 				Path:            path,
-				FileType:        JSON_TYPE,
-				Delimiter:       ",",
+				FileType:        string(JSON_TYPE),
 				ActionAfterRead: 2,
 				MoveTo:          filepath.Join(path, "ddd"),
 			},
@@ -227,7 +225,7 @@ func TestLines(t *testing.T) {
 		model.NewDefaultRawTuple([]byte("{\"id\": 3,\"name\": \"John Smith\"}"), meta, mc.Now()),
 		model.NewDefaultRawTuple([]byte("[{\"id\": 4,\"name\": \"John Smith\"},{\"id\": 5,\"name\": \"John Smith\"}]"), meta, mc.Now()),
 	}
-	r := GetSource().(api.BytesSource)
+	r := GetSource().(api.TupleSource)
 	mock.TestSourceConnector(t, r, map[string]any{
 		"path":            path,
 		"fileType":        "lines",
@@ -238,6 +236,40 @@ func TestLines(t *testing.T) {
 	})
 }
 
+func TestCSVBatch(t *testing.T) {
+	path, err := os.Getwd()
+	if err != nil {
+		t.Fatal(err)
+	}
+	path = filepath.Join(path, "test")
+
+	meta1 := map[string]any{
+		"file": filepath.Join(path, "csv", "a.csv"),
+	}
+	meta2 := map[string]any{
+		"file": filepath.Join(path, "csv", "b.csv"),
+	}
+	mc := timex.Clock
+	exp := []api.MessageTuple{
+		model.NewDefaultSourceTuple(map[string]any{"@": "#", "id": "1", "ts": "1670170500", "value": "161.927872"}, meta1, mc.Now()),
+		model.NewDefaultSourceTuple(map[string]any{"@": "#", "id": "2", "ts": "1670170900", "value": "176"}, meta1, mc.Now()),
+		model.NewDefaultSourceTuple(map[string]any{"id": "33", "ts": "1670270500", "humidity": "89"}, meta2, mc.Now()),
+		model.NewDefaultSourceTuple(map[string]any{"id": "44", "ts": "1670270900", "humidity": "76"}, meta2, mc.Now()),
+	}
+	r := GetSource().(api.TupleSource)
+	mock.TestSourceConnector(t, r, map[string]any{
+		"path":             path,
+		"fileType":         "csv",
+		"datasource":       "csv",
+		"hasHeader":        true,
+		"delimiter":        "\t",
+		"ignoreStartLines": 3,
+		"ignoreEndLines":   1,
+	}, exp, func() {
+		// do nothing
+	})
+}
+
 func TestBatch(t *testing.T) {
 	path, err := os.Getwd()
 	if err != nil {
@@ -252,7 +284,7 @@ func TestBatch(t *testing.T) {
 	exp := []api.MessageTuple{
 		model.NewDefaultRawTuple([]byte("[{\"id\": 1,\"name\": \"John Doe\",\"height\": 1.82},{\"id\": 2,\"name\": \"Jane Doe\",\"height\": 1.65}]"), meta, mc.Now()),
 	}
-	r := GetSource().(api.BytesSource)
+	r := GetSource().(api.TupleSource)
 	mock.TestSourceConnector(t, r, map[string]any{
 		"path":       path,
 		"fileType":   "json",
@@ -281,7 +313,7 @@ func TestIgnoreLines(t *testing.T) {
 		model.NewDefaultRawTuple([]byte("{\"id\": 2,\"name\": \"Jane Doe\"}"), meta, mc.Now()),
 		model.NewDefaultRawTuple([]byte("{\"id\": 3,\"name\": \"John Smith\"}"), meta, mc.Now()),
 	}
-	r := GetSource().(api.BytesSource)
+	r := GetSource().(api.TupleSource)
 	r.(api.Bounded).SetEofIngest(func(ctx api.StreamContext) {
 		fmt.Printf("eof")
 	})
@@ -328,7 +360,7 @@ func TestIntervalAndDir(t *testing.T) {
 		model.NewDefaultRawTuple([]byte("[{\"id\": 4,\"name\": \"John Smith\"},{\"id\": 5,\"name\": \"John Smith\"}]"), meta, ts),
 		model.NewDefaultRawTuple([]byte("{\"id\": 2,\"name\": \"Jane Doe\"}"), meta, ts),
 	}
-	r := GetSource().(api.BytesSource)
+	r := GetSource().(api.TupleSource)
 	mock.TestSourceConnector(t, r, map[string]any{
 		"path":             path,
 		"fileType":         "lines",
diff --git a/internal/io/file/test/csv/a.csv b/internal/io/file/test/csv/a.csv
new file mode 100644
index 00000000..16d0da3a
--- /dev/null
+++ b/internal/io/file/test/csv/a.csv
@@ -0,0 +1,7 @@
+//	some irrelevant content
+<!	Version=1.0	Code=UTF-8	!>
+<special content>
+@	id	ts	value
+#	1	1670170500	161.927872
+#	2	1670170900	176
+<special content>
diff --git a/internal/io/file/test/csv/b.csv b/internal/io/file/test/csv/b.csv
new file mode 100644
index 00000000..de354274
--- /dev/null
+++ b/internal/io/file/test/csv/b.csv
@@ -0,0 +1,7 @@
+//	some irrelevant content
+<!	Version=1.0	Code=UTF-8	!>
+<special content>
+id	ts	humidity
+33	1670270500	89
+44	1670270900	76
+<special content>
diff --git a/internal/io/mqtt/client/client_test.go b/internal/io/mqtt/client/client_test.go
new file mode 100644
index 00000000..b4e180bf
--- /dev/null
+++ b/internal/io/mqtt/client/client_test.go
@@ -0,0 +1,55 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package client
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// NOTICE!!! Need to run a MQTT broker in localhost:1883 for this test or change the url to your broker
+const url = "tcp://127.0.0.1:1883"
+
+func TestValidate(t *testing.T) {
+	tests := []struct {
+		name  string
+		props map[string]any
+		err   string
+	}{
+		{
+			name: "No server",
+			props: map[string]any{
+				"server": "",
+			},
+			err: "missing server property",
+		},
+		{
+			name: "invalid protocol",
+			props: map[string]any{
+				"server":          url,
+				"protocolVersion": "5.0",
+			},
+			err: "unsupported protocol version 5.0",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := ValidateConfig(tt.props)
+			assert.Error(t, err)
+			assert.Equal(t, tt.err, err.Error())
+		})
+	}
+}
diff --git a/internal/io/mqtt/connection_test.go b/internal/io/mqtt/connection_test.go
deleted file mode 100644
index 42704f30..00000000
--- a/internal/io/mqtt/connection_test.go
+++ /dev/null
@@ -1,94 +0,0 @@
-// Copyright 2024 EMQ Technologies Co., Ltd.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package mqtt
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-
-	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
-)
-
-func TestConnectionLC(t *testing.T) {
-	propsShared := map[string]any{
-		"connectionSelector": "mqtt.localConnection",
-	}
-	propsNormal := map[string]any{
-		"server": url,
-	}
-	propsInvalid := map[string]any{
-		"server": "abc",
-	}
-	ctx := mockContext.NewMockContext("test", "op")
-	connShared1, err := GetConnection(ctx, "mqtt.localConnection", propsShared)
-	assert.NoError(t, err)
-	assert.Equal(t, 1, len(connectionPool))
-	_, err = GetConnection(ctx, "", propsNormal)
-	assert.NoError(t, err)
-	assert.Equal(t, 1, len(connectionPool))
-	connShared2, err := GetConnection(ctx, "mqtt.localConnection", propsShared)
-	assert.NoError(t, err)
-	assert.Equal(t, 1, len(connectionPool))
-	assert.Equal(t, connShared1, connShared2)
-	_, err = GetConnection(ctx, "", propsInvalid)
-	assert.Error(t, err)
-	assert.Equal(t, err.Error(), "found error when connecting for abc: network Error : dial tcp: address abc: missing port in address")
-	assert.Equal(t, 1, len(connectionPool))
-
-	c := connShared1.refCount.Load()
-	assert.Equal(t, int32(2), c)
-
-	err = connShared1.Ping()
-	assert.NoError(t, err)
-
-	// Test subscribe in the connector test.
-
-	DetachConnection(connShared1, "mqtt.localConnection", "")
-	assert.Equal(t, 1, len(connectionPool))
-	DetachConnection(connShared2, "mqtt.localConnection", "")
-	assert.Equal(t, 0, len(connectionPool))
-}
-
-func TestValidate(t *testing.T) {
-	tests := []struct {
-		name  string
-		props map[string]any
-		err   string
-	}{
-		{
-			name: "No server",
-			props: map[string]any{
-				"server": "",
-			},
-			err: "missing server property",
-		},
-		{
-			name: "invalid protocol",
-			props: map[string]any{
-				"server":          url,
-				"protocolVersion": "5.0",
-			},
-			err: "unsupported protocol version 5.0",
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			_, err := validateConfig(tt.props)
-			assert.Error(t, err)
-			assert.Equal(t, tt.err, err.Error())
-		})
-	}
-}
diff --git a/internal/io/mqtt/sink_test.go b/internal/io/mqtt/sink_test.go
index 903efd3e..5ce5af52 100644
--- a/internal/io/mqtt/sink_test.go
+++ b/internal/io/mqtt/sink_test.go
@@ -21,10 +21,12 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
 func TestSinkConfigure(t *testing.T) {
+	connection.InitConnectionManagerInTest()
 	tests := []struct {
 		name           string
 		input          map[string]interface{}
@@ -135,6 +137,7 @@ func TestSinkConfigure(t *testing.T) {
 }
 
 func TestValidateMQTTSinkConf(t *testing.T) {
+	connection.InitConnectionManagerInTest()
 	testcases := []struct {
 		topic       string
 		expectError bool
diff --git a/internal/io/mqtt/source_sink_test.go b/internal/io/mqtt/source_sink_test.go
index 7a203db8..a7d4aeaf 100644
--- a/internal/io/mqtt/source_sink_test.go
+++ b/internal/io/mqtt/source_sink_test.go
@@ -20,12 +20,14 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/topotest/mockclock"
 	"github.com/lf-edge/ekuiper/v2/pkg/mock"
 	"github.com/lf-edge/ekuiper/v2/pkg/model"
 )
 
 func TestSourceSink(t *testing.T) {
+	connection.InitConnectionManagerInTest()
 	sc := GetSource().(api.BytesSource)
 	sk := GetSink().(api.BytesCollector)
 	mc := mockclock.GetMockClock()
@@ -57,10 +59,14 @@ func TestSourceSink(t *testing.T) {
 	mock.TestSourceConnector(t, sc, map[string]any{
 		"server":     url,
 		"datasource": "demo",
+		"qos":        0,
+		"topic":      "demo",
 	}, result, func() {
 		err := mock.RunBytesSinkCollect(sk, data, map[string]any{
-			"server": url,
-			"topic":  "demo",
+			"server":   url,
+			"topic":    "demo",
+			"qos":      0,
+			"retained": false,
 		})
 		assert.NoError(t, err)
 	})
diff --git a/internal/io/mqtt/source_test.go b/internal/io/mqtt/source_test.go
index 2bc05f24..6b74691e 100644
--- a/internal/io/mqtt/source_test.go
+++ b/internal/io/mqtt/source_test.go
@@ -21,6 +21,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
@@ -33,6 +34,7 @@ func init() {
 }
 
 func TestProvision(t *testing.T) {
+	connection.InitConnectionManagerInTest()
 	tests := []struct {
 		name  string
 		props map[string]any
diff --git a/internal/server/rest_test.go b/internal/server/rest_test.go
index 9e2b4119..df6239a3 100644
--- a/internal/server/rest_test.go
+++ b/internal/server/rest_test.go
@@ -32,6 +32,7 @@ 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"
@@ -91,6 +92,8 @@ func (suite *RestTestSuite) SetupTest() {
 	r.HandleFunc("/data/export", configurationExportHandler).Methods(http.MethodGet, http.MethodPost)
 	r.HandleFunc("/data/import", configurationImportHandler).Methods(http.MethodPost)
 	r.HandleFunc("/data/import/status", configurationStatusHandler).Methods(http.MethodGet)
+	r.HandleFunc("/connections", connectionsHandler).Methods(http.MethodGet, http.MethodPost)
+	r.HandleFunc("/connection/{id}", connectionHandler).Methods(http.MethodGet, http.MethodDelete)
 	// r.HandleFunc("/connection/websocket", connectionHandler).Methods(http.MethodGet, http.MethodPost, http.MethodDelete)
 	r.HandleFunc("/metadata/sinks/{name}/confKeys/{confKey}", sinkConfKeyHandler).Methods(http.MethodDelete, http.MethodPut)
 	suite.r = r
@@ -802,3 +805,32 @@ func (suite *RestTestSuite) TestCreateDuplicateRule() {
 	returnVal, _ = io.ReadAll(w2.Result().Body)
 	require.Equal(suite.T(), `{"error":1000,"message":"rule test12345 already exists"}`+"\n", string(returnVal))
 }
+
+func (suite *RestTestSuite) TestConnection() {
+	connection.InitConnectionManagerInTest()
+	buf1 := bytes.NewBuffer([]byte(`{"id":"id1","typ":"mock","props":{}}`))
+	req1, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/connections", buf1)
+	w1 := httptest.NewRecorder()
+	suite.r.ServeHTTP(w1, req1)
+	require.Equal(suite.T(), http.StatusCreated, w1.Code)
+
+	req, _ := http.NewRequest(http.MethodGet, "http://localhost:8080/connections", bytes.NewBufferString("any"))
+	w := httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	assert.Equal(suite.T(), http.StatusOK, w.Code)
+	returnVal, _ := io.ReadAll(w.Result().Body)
+	require.Equal(suite.T(), `["id1"]`, string(returnVal))
+
+	req, _ = http.NewRequest(http.MethodGet, "http://localhost:8080/connection/id1", bytes.NewBufferString("any"))
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	assert.Equal(suite.T(), http.StatusOK, w.Code)
+
+	returnVal, _ = io.ReadAll(w.Result().Body)
+	require.Equal(suite.T(), `{"id":"id1","err":""}`, string(returnVal))
+
+	req, _ = http.NewRequest(http.MethodDelete, "http://localhost:8080/connection/id1", bytes.NewBufferString("any"))
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	assert.Equal(suite.T(), http.StatusOK, w.Code)
+}
diff --git a/internal/topo/node/decode_op_test.go b/internal/topo/node/decode_op_test.go
index e8dcf5c5..2480b134 100644
--- a/internal/topo/node/decode_op_test.go
+++ b/internal/topo/node/decode_op_test.go
@@ -26,6 +26,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
 func TestJSON(t *testing.T) {
@@ -38,7 +39,9 @@ func TestJSON(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			op, err := NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: tt.concurrency}, &ast.Options{FORMAT: "json"}, false, true, nil)
+			op, err := NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: tt.concurrency}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{
+				"sendInterval": "10ms",
+			})
 			assert.NoError(t, err)
 			out := make(chan any, 100)
 			err = op.AddOutput(out, "test")
@@ -68,7 +71,7 @@ func TestJSON(t *testing.T) {
 				{errors.New("unsupported decode result: hello")},
 				{errors.New("unsupported data received: invalid")},
 			}
-
+			timex.Add(2 * time.Second)
 			for i, c := range cases {
 				op.input <- c
 				for _, e := range expects[i] {
@@ -89,7 +92,7 @@ func TestJSON(t *testing.T) {
 // Concurrency 10 - BenchmarkThrougput-16           1000000000               0.1553 ns/op
 // This is useful when a node is much slower
 func BenchmarkThrougput(b *testing.B) {
-	op, err := NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: 10, Debug: true}, &ast.Options{FORMAT: "mock"}, false, true, nil)
+	op, err := NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: 10, Debug: true}, &ast.Options{FORMAT: "mock"}, false, true, nil, nil)
 	assert.NoError(b, err)
 	out := make(chan any, 100)
 	err = op.AddOutput(out, "test")
@@ -130,7 +133,7 @@ func TestJSONWithSchema(t *testing.T) {
 				"a": {
 					Type: "bigint",
 				},
-			})
+			}, nil)
 			assert.NoError(t, err)
 			out := make(chan any, 100)
 			err = op.AddOutput(out, "test")
@@ -220,7 +223,13 @@ func TestJSONWithSchema(t *testing.T) {
 }
 
 func TestValidate(t *testing.T) {
-	_, err := NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "cann"}, false, true, nil)
+	_, err := NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "cann"}, false, true, nil, nil)
 	assert.Error(t, err)
 	assert.Equal(t, "cannot get converter from format cann, schemaId : format type cann not supported", err.Error())
+	_, err = NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{"sendInterval": "none"})
+	assert.Error(t, err)
+	assert.EqualError(t, err, "1 error(s) decoding:\n\n* error decoding 'sendInterval': time: invalid duration \"none\"")
+	do, err := NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{"sendInterval": "12s"})
+	assert.NoError(t, err)
+	assert.Equal(t, 12*time.Second, do.sendInterval)
 }
diff --git a/internal/topo/planner/planner_source_test.go b/internal/topo/planner/planner_source_test.go
new file mode 100644
index 00000000..a395006b
--- /dev/null
+++ b/internal/topo/planner/planner_source_test.go
@@ -0,0 +1,292 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package planner
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"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/xsql"
+	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+)
+
+func TestPlanTopo(t *testing.T) {
+	kv, err := store.GetKV("stream")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	streamSqls := map[string]string{
+		"src1":     `CREATE STREAM src1 () WITH (DATASOURCE="src1", FORMAT="json", TYPE="mqtt");`,
+		"src2":     `CREATE STREAM src2 () WITH (DATASOURCE="src1", FORMAT="json", TYPE="mqtt", SHARED="true");`,
+		"src3":     `CREATE STREAM src3 () WITH (DATASOURCE="topic1", FORMAT="json", TYPE="mqtt", CONF_KEY="testSel");`,
+		"src4":     `CREATE STREAM src4 () WITH (DATASOURCE="topic1", FORMAT="json", TYPE="mqtt", CONF_KEY="testSel",SHARED="true");`,
+		"filesrc1": `CREATE STREAM fs1 () WITH (FORMAT="json", TYPE="file",CONF_KEY="lines");`,
+		"filesrc2": `CREATE STREAM fs2 () WITH (FORMAT="delimited", TYPE="file",CONF_KEY="csv");`,
+		"filesrc3": `CREATE STREAM fs3 () WITH (FORMAT="json",TYPE="file",CONF_KEY="json");`,
+	}
+	for name, sql := range streamSqls {
+		s, err := json.Marshal(&xsql.StreamInfo{
+			StreamType: ast.TypeStream,
+			Statement:  sql,
+		})
+		assert.NoError(t, err)
+		err = kv.Set(name, string(s))
+		assert.NoError(t, err)
+	}
+	confs := []struct {
+		conf map[string]any
+		p    string
+		k    string
+	}{
+		{
+			conf: map[string]any{
+				"connectionSelector": "mqtt.localConnection",
+			},
+			p: "mqtt",
+			k: "testSel",
+		},
+		{
+			conf: map[string]any{
+				"fileType":      "csv",
+				"decompression": "gzip",
+			},
+			p: "file",
+			k: "csv",
+		},
+		{
+			conf: map[string]any{
+				"fileType":      "lines",
+				"decompression": "gzip",
+			},
+			p: "file",
+			k: "lines",
+		},
+		{
+			conf: map[string]any{
+				"decompression": "gzip",
+			},
+			p: "file",
+			k: "json",
+		},
+	}
+	meta.InitYamlConfigManager()
+	dataDir, _ := conf.GetDataLoc()
+	err = os.MkdirAll(filepath.Join(dataDir, "sources"), 0o755)
+	assert.NoError(t, err)
+
+	for _, cc := range confs {
+		p, k := cc.p, cc.k
+		bs, err := json.Marshal(cc.conf)
+		assert.NoError(t, err)
+		err = meta.AddSourceConfKey(p, k, "", bs)
+		assert.NoError(t, err)
+		// intended to run at last
+		defer func() {
+			err = meta.DelSourceConfKey(p, k, "")
+			assert.NoError(t, err)
+		}()
+	}
+
+	tests := []struct {
+		name string
+		sql  string
+		topo *def.PrintableTopo
+	}{
+		{
+			name: "testMqttSplit",
+			sql:  `SELECT * FROM src1`,
+			topo: &def.PrintableTopo{
+				Sources: []string{"source_src1"},
+				Edges: map[string][]any{
+					"source_src1": {
+						"op_2_decoder",
+					},
+					"op_2_decoder": {
+						"op_3_project",
+					},
+					"op_3_project": {
+						"op_logToMemory_0_0_transform",
+					},
+					"op_logToMemory_0_0_transform": {
+						"op_logToMemory_0_1_encode",
+					},
+					"op_logToMemory_0_1_encode": {
+						"sink_logToMemory_0",
+					},
+				},
+			},
+		},
+		{
+			name: "testSharedMqttSplit",
+			sql:  `SELECT * FROM src2`,
+			topo: &def.PrintableTopo{
+				Sources: []string{"source_src2"},
+				Edges: map[string][]any{
+					"source_src2": {
+						"op_src2_2_decoder",
+					},
+					"op_src2_2_decoder": {
+						"op_3_project",
+					},
+					"op_3_project": {
+						"op_logToMemory_0_0_transform",
+					},
+					"op_logToMemory_0_0_transform": {
+						"op_logToMemory_0_1_encode",
+					},
+					"op_logToMemory_0_1_encode": {
+						"sink_logToMemory_0",
+					},
+				},
+			},
+		},
+		{
+			name: "testSharedConnSplit",
+			sql:  `SELECT * FROM src3`,
+			topo: &def.PrintableTopo{
+				Sources: []string{"source_mqtt.localConnection/topic1"},
+				Edges: map[string][]any{
+					"source_mqtt.localConnection/topic1": {
+						"op_2_decoder",
+					},
+					"op_2_decoder": {
+						"op_3_project",
+					},
+					"op_3_project": {
+						"op_logToMemory_0_0_transform",
+					},
+					"op_logToMemory_0_0_transform": {
+						"op_logToMemory_0_1_encode",
+					},
+					"op_logToMemory_0_1_encode": {
+						"sink_logToMemory_0",
+					},
+				},
+			},
+		},
+		{
+			name: "testSharedNodeWithSharedConnSplit",
+			sql:  `SELECT * FROM src4`,
+			topo: &def.PrintableTopo{
+				Sources: []string{"source_mqtt.localConnection/topic1"},
+				Edges: map[string][]any{
+					"source_mqtt.localConnection/topic1": {
+						"op_src4_2_decoder",
+					},
+					"op_src4_2_decoder": {
+						"op_3_project",
+					},
+					"op_3_project": {
+						"op_logToMemory_0_0_transform",
+					},
+					"op_logToMemory_0_0_transform": {
+						"op_logToMemory_0_1_encode",
+					},
+					"op_logToMemory_0_1_encode": {
+						"sink_logToMemory_0",
+					},
+				},
+			},
+		},
+		{
+			name: "test lines file",
+			sql:  `SELECT * FROM filesrc1`,
+			topo: &def.PrintableTopo{
+				Sources: []string{"source_fs1"},
+				Edges: map[string][]any{
+					"source_fs1": {
+						"op_2_decoder",
+					},
+					"op_2_decoder": {
+						"op_3_project",
+					},
+					"op_3_project": {
+						"op_logToMemory_0_0_transform",
+					},
+					"op_logToMemory_0_0_transform": {
+						"op_logToMemory_0_1_encode",
+					},
+					"op_logToMemory_0_1_encode": {
+						"sink_logToMemory_0",
+					},
+				},
+			},
+		},
+		{
+			name: "test csv file",
+			sql:  `SELECT * FROM filesrc2`,
+			topo: &def.PrintableTopo{
+				Sources: []string{"source_fs2"},
+				Edges: map[string][]any{
+					"source_fs2": {
+						"op_2_project",
+					},
+					"op_2_project": {
+						"op_logToMemory_0_0_transform",
+					},
+					"op_logToMemory_0_0_transform": {
+						"op_logToMemory_0_1_encode",
+					},
+					"op_logToMemory_0_1_encode": {
+						"sink_logToMemory_0",
+					},
+				},
+			},
+		},
+		{
+			name: "test json file",
+			sql:  `SELECT * FROM filesrc3`,
+			topo: &def.PrintableTopo{
+				Sources: []string{"source_fs3"},
+				Edges: map[string][]any{
+					"source_fs3": {
+						"op_2_decompress",
+					},
+					"op_2_decompress": {
+						"op_3_decoder",
+					},
+					"op_3_decoder": {
+						"op_4_project",
+					},
+					"op_4_project": {
+						"op_logToMemory_0_0_transform",
+					},
+					"op_logToMemory_0_0_transform": {
+						"op_logToMemory_0_1_encode",
+					},
+					"op_logToMemory_0_1_encode": {
+						"sink_logToMemory_0",
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tp, err := PlanSQLWithSourcesAndSinks(def.GetDefaultRule(tt.name, tt.sql), nil)
+			assert.NoError(t, err)
+			assert.Equal(t, tt.topo, tp.GetTopo())
+		})
+	}
+}
diff --git a/internal/topo/planner/planner_test.go b/internal/topo/planner/planner_test.go
index fe9a927d..6ead3104 100644
--- a/internal/topo/planner/planner_test.go
+++ b/internal/topo/planner/planner_test.go
@@ -4785,11 +4785,11 @@ func TestTransformSourceNode(t *testing.T) {
 	props := nodeConf.GetSourceConf("mqtt", &ast.Options{TYPE: "mqtt", DATASOURCE: "topic1"})
 	srcNode, err := node.NewSourceNode(tp.GetContext(), "test", &mqtt.SourceConnector{}, props, &def.RuleOption{SendError: false})
 	assert.NoError(t, err)
-	decodeNode, err := node.NewDecodeOp("2_decoder", "test", "test", &def.RuleOption{SendError: false}, &ast.Options{TYPE: "mqtt"}, false, false, schema)
+	decodeNode, err := node.NewDecodeOp("2_decoder", "test", "test", &def.RuleOption{SendError: false}, &ast.Options{TYPE: "mqtt"}, false, false, schema, nil)
 	assert.NoError(t, err)
 	decomNode, err := node.NewDecompressOp("2_decompressor", &def.RuleOption{SendError: false}, "gzip")
 	assert.NoError(t, err)
-	decodeNode2, err := node.NewDecodeOp("3_decoder", "test", "test", &def.RuleOption{SendError: false}, &ast.Options{TYPE: "mqtt"}, false, false, schema)
+	decodeNode2, err := node.NewDecodeOp("3_decoder", "test", "test", &def.RuleOption{SendError: false}, &ast.Options{TYPE: "mqtt"}, false, false, schema, nil)
 	assert.NoError(t, err)
 	props2 := nodeConf.GetSourceConf("mqtt", &ast.Options{TYPE: "mqtt", CONF_KEY: "testCom", DATASOURCE: "topic2"})
 	srcNode2, err := node.NewSourceNode(tp.GetContext(), "test", &mqtt.SourceConnector{}, props2, &def.RuleOption{SendError: false})
@@ -4967,151 +4967,3 @@ func TestGetLogicalPlanForExplain(t *testing.T) {
 		}
 	}
 }
-
-func TestPlanTopo(t *testing.T) {
-	kv, err := store.GetKV("stream")
-	if err != nil {
-		t.Error(err)
-		return
-	}
-	streamSqls := map[string]string{
-		"src1": `CREATE STREAM src1 () WITH (DATASOURCE="src1", FORMAT="json", TYPE="mqtt");`,
-		"src2": `CREATE STREAM src2 () WITH (DATASOURCE="src1", FORMAT="json", TYPE="mqtt", SHARED="true");`,
-		"src3": `CREATE STREAM src3 () WITH (DATASOURCE="topic1", FORMAT="json", TYPE="mqtt", CONF_KEY="testSel");`,
-		"src4": `CREATE STREAM src4 () WITH (DATASOURCE="topic1", FORMAT="json", TYPE="mqtt", CONF_KEY="testSel",SHARED="true");`,
-	}
-	for name, sql := range streamSqls {
-		s, err := json.Marshal(&xsql.StreamInfo{
-			StreamType: ast.TypeStream,
-			Statement:  sql,
-		})
-		assert.NoError(t, err)
-		err = kv.Set(name, string(s))
-		assert.NoError(t, err)
-	}
-	// add connectionSelector for meta
-	a1 := map[string]interface{}{
-		"connectionSelector": "mqtt.localConnection",
-	}
-	bs, err := json.Marshal(a1)
-	assert.NoError(t, err)
-	meta.InitYamlConfigManager()
-	dataDir, _ := conf.GetDataLoc()
-	err = os.MkdirAll(filepath.Join(dataDir, "sources"), 0o755)
-	assert.NoError(t, err)
-	err = meta.AddSourceConfKey("mqtt", "testSel", "", bs)
-	assert.NoError(t, err)
-	defer func() {
-		err = meta.DelSourceConfKey("mqtt", "testSel", "")
-		assert.NoError(t, err)
-	}()
-	tests := []struct {
-		name string
-		sql  string
-		topo *def.PrintableTopo
-	}{
-		{
-			name: "testMqttSplit",
-			sql:  `SELECT * FROM src1`,
-			topo: &def.PrintableTopo{
-				Sources: []string{"source_src1"},
-				Edges: map[string][]any{
-					"source_src1": {
-						"op_2_decoder",
-					},
-					"op_2_decoder": {
-						"op_3_project",
-					},
-					"op_3_project": {
-						"op_logToMemory_0_0_transform",
-					},
-					"op_logToMemory_0_0_transform": {
-						"op_logToMemory_0_1_encode",
-					},
-					"op_logToMemory_0_1_encode": {
-						"sink_logToMemory_0",
-					},
-				},
-			},
-		},
-		{
-			name: "testSharedMqttSplit",
-			sql:  `SELECT * FROM src2`,
-			topo: &def.PrintableTopo{
-				Sources: []string{"source_src2"},
-				Edges: map[string][]any{
-					"source_src2": {
-						"op_src2_2_decoder",
-					},
-					"op_src2_2_decoder": {
-						"op_3_project",
-					},
-					"op_3_project": {
-						"op_logToMemory_0_0_transform",
-					},
-					"op_logToMemory_0_0_transform": {
-						"op_logToMemory_0_1_encode",
-					},
-					"op_logToMemory_0_1_encode": {
-						"sink_logToMemory_0",
-					},
-				},
-			},
-		},
-		{
-			name: "testSharedConnSplit",
-			sql:  `SELECT * FROM src3`,
-			topo: &def.PrintableTopo{
-				Sources: []string{"source_mqtt.localConnection/topic1"},
-				Edges: map[string][]any{
-					"source_mqtt.localConnection/topic1": {
-						"op_2_decoder",
-					},
-					"op_2_decoder": {
-						"op_3_project",
-					},
-					"op_3_project": {
-						"op_logToMemory_0_0_transform",
-					},
-					"op_logToMemory_0_0_transform": {
-						"op_logToMemory_0_1_encode",
-					},
-					"op_logToMemory_0_1_encode": {
-						"sink_logToMemory_0",
-					},
-				},
-			},
-		},
-		{
-			name: "testSharedNodeWithSharedConnSplit",
-			sql:  `SELECT * FROM src4`,
-			topo: &def.PrintableTopo{
-				Sources: []string{"source_mqtt.localConnection/topic1"},
-				Edges: map[string][]any{
-					"source_mqtt.localConnection/topic1": {
-						"op_src4_2_decoder",
-					},
-					"op_src4_2_decoder": {
-						"op_3_project",
-					},
-					"op_3_project": {
-						"op_logToMemory_0_0_transform",
-					},
-					"op_logToMemory_0_0_transform": {
-						"op_logToMemory_0_1_encode",
-					},
-					"op_logToMemory_0_1_encode": {
-						"sink_logToMemory_0",
-					},
-				},
-			},
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			tp, err := PlanSQLWithSourcesAndSinks(def.GetDefaultRule(tt.name, tt.sql), nil)
-			assert.NoError(t, err)
-			assert.Equal(t, tt.topo, tp.GetTopo())
-		})
-	}
-}
diff --git a/pkg/mock/test_source.go b/pkg/mock/test_source.go
index 1b22feb7..62fff1ad 100644
--- a/pkg/mock/test_source.go
+++ b/pkg/mock/test_source.go
@@ -78,7 +78,22 @@ func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expec
 				panic(err)
 			})
 		case api.TupleSource:
-			panic("added later")
+			err = ss.Subscribe(ctx, func(ctx api.StreamContext, message any, meta map[string]any, ts time.Time) {
+				switch mt := message.(type) {
+				case []byte:
+					result = append(result, model.NewDefaultRawTuple(mt, meta, ts))
+				case map[string]any:
+					result = append(result, model.NewDefaultSourceTuple(mt, meta, ts))
+				default:
+					panic("not supported yet")
+				}
+				limit--
+				if limit == 0 {
+					wg.Done()
+				}
+			}, func(ctx api.StreamContext, err error) {
+				panic(err)
+			})
 		default:
 			panic("wrong source type")
 		}