Repository: lf-edge/ekuiper
Pull request #2996 (2024-07-08 10:11:58)

feat: support export metadata in yaml


Patch:
diff --git a/extension/sql/client/dburl.go b/extensions/impl/sql/client/dburl.go
similarity index 100%
rename from extension/sql/client/dburl.go
rename to extensions/impl/sql/client/dburl.go
diff --git a/extension/sql/sink.go b/extensions/impl/sql/sink.go
similarity index 99%
rename from extension/sql/sink.go
rename to extensions/impl/sql/sink.go
index 39f3e56a..230d7947 100644
--- a/extension/sql/sink.go
+++ b/extensions/impl/sql/sink.go
@@ -20,7 +20,7 @@ import (
 	"strings"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/extension/sql/client"
+	"github.com/lf-edge/ekuiper/v2/extensions/impl/sql/client"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/connection"
diff --git a/extension/sql/source.go b/extensions/impl/sql/source.go
similarity index 96%
rename from extension/sql/source.go
rename to extensions/impl/sql/source.go
index 7c45a023..98585fa0 100644
--- a/extension/sql/source.go
+++ b/extensions/impl/sql/source.go
@@ -25,7 +25,7 @@ import (
 	"github.com/pingcap/failpoint"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/extension/sql/client"
+	client2 "github.com/lf-edge/ekuiper/v2/extensions/impl/sql/client"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	"github.com/lf-edge/ekuiper/v2/pkg/sqldatabase/sqlgen"
@@ -34,7 +34,7 @@ import (
 type SQLSourceConnector struct {
 	conf          *SQLConf
 	Query         sqlgen.SqlQueryGenerator
-	conn          *client.SQLConnection
+	conn          *client2.SQLConnection
 	props         map[string]any
 	needReconnect bool
 }
@@ -61,7 +61,7 @@ func (s *SQLSourceConnector) Provision(ctx api.StreamContext, props map[string]a
 	}
 	s.conf = cfg
 	s.props = props
-	sqlDriver, err := client.ParseDriver(cfg.DBUrl)
+	sqlDriver, err := client2.ParseDriver(cfg.DBUrl)
 	if err != nil {
 		return fmt.Errorf("dburl.Parse %s fail with error: %v", cfg.DBUrl, err)
 	}
@@ -78,14 +78,14 @@ func (s *SQLSourceConnector) Provision(ctx api.StreamContext, props map[string]a
 
 func (s *SQLSourceConnector) Connect(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("Connecting to sql server")
-	var cli *client.SQLConnection
+	var cli *client2.SQLConnection
 	var err error
 	id := s.conf.DBUrl
 	conn, err := connection.FetchConnection(ctx, id, "sql", s.props)
 	if err != nil {
 		return err
 	}
-	cli = conn.(*client.SQLConnection)
+	cli = conn.(*client2.SQLConnection)
 	s.conn = cli
 	return err
 }
diff --git a/extension/sql/testx/mysql_server.go b/extensions/impl/sql/testx/mysql_server.go
similarity index 100%
rename from extension/sql/testx/mysql_server.go
rename to extensions/impl/sql/testx/mysql_server.go
diff --git a/extension/video/source.go b/extensions/impl/video/source.go
similarity index 100%
rename from extension/video/source.go
rename to extensions/impl/video/source.go
diff --git a/extension/video/source_test.go b/extensions/impl/video/source_test.go
similarity index 100%
rename from extension/video/source_test.go
rename to extensions/impl/video/source_test.go
diff --git a/extensions/sinks/sql/sql.go b/extensions/sinks/sql/sql.go
new file mode 100644
index 00000000..38972dbe
--- /dev/null
+++ b/extensions/sinks/sql/sql.go
@@ -0,0 +1,24 @@
+// Copyright 2022-2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/extensions/impl/sql"
+)
+
+func Sql() api.Sink {
+	return sql.GetSink()
+}
diff --git a/extensions/sinks/sql/sql.json b/extensions/sinks/sql/sql.json
new file mode 100644
index 00000000..6bc893c2
--- /dev/null
+++ b/extensions/sinks/sql/sql.json
@@ -0,0 +1,76 @@
+{
+  "about": {
+    "trial": true,
+    "author": {
+      "name": "EMQ",
+      "email": "contact@emqx.io",
+      "company": "EMQ Technologies Co., Ltd",
+      "website": "https://www.emqx.io"
+    },
+    "helpUrl": {
+      "en_US": "https://ekuiper.org/docs/en/latest/guide/sinks/plugin/sql.html",
+      "zh_CN": "https://ekuiper.org/docs/zh/latest/guide/sinks/plugin/sql.html"
+    },
+    "description": {
+      "en_US": "This a sink plugin for Sql based Database, it can be used for saving the analysis data into Sql based database.",
+      "zh_CN": "本插件为 SQL Database 的持久化插件，可以用于将分析数据存入支持 SQL 语法的数据库中"
+    }
+  },
+  "libs": [
+  ],
+  "properties": [
+    {
+      "name": "url",
+      "default": "",
+      "optional": false,
+      "control": "text",
+      "type": "string",
+      "hint": {
+        "en_US": "The url of the database, it is important to note that the password in the URL is url-encoded characters, and you need to re-enter the password when editing. For example, the database address of mysql is mysql://username:password@127.0.0.1:3306/testdb?parseTime=true, if the original password is 123:#?, the password here should be filled in with 123%3A%23%3F",
+        "zh_CN": "数据库服务器的 URL，需要特别注意的是 URL 中的密码为 url 编码后的字符，并且编辑时需要重新输入密码。例如，mysql 的数据库地址为 mysql://username:password@127.0.0.1:3306/testdb?parseTime=true，若原始密码为 123:#?，则这里的 password 应填入 123%3A%23%3F"
+      },
+      "label": {
+        "en_US": "server address",
+        "zh_CN": "数据库地址"
+      }
+    },
+    {
+      "name": "table",
+      "default": "",
+      "optional": false,
+      "control": "text",
+      "type": "string",
+      "hint": {
+        "en_US": "Table name",
+        "zh_CN": "表名"
+      },
+      "label": {
+        "en_US": "Table name",
+        "zh_CN": "表名"
+      }
+    },
+    {
+      "name": "fields",
+      "default": [],
+      "optional": true,
+      "control": "list",
+      "type": "list_string",
+      "hint": {
+        "en_US": "Tag fields written to the database",
+        "zh_CN": "写入数据库的标签字段"
+      },
+      "label": {
+        "en_US": "Tag Fields",
+        "zh_CN": "标签字段"
+      }
+    }
+  ],
+  "node": {
+    "category": "sink",
+    "icon": "iconPath",
+    "label": {
+      "en": "SQL",
+      "zh": "SQL"
+    }
+  }
+}
diff --git a/extension/random/ext/export.go b/extensions/sources/random/random.go
similarity index 92%
rename from extension/random/ext/export.go
rename to extensions/sources/random/random.go
index 9d76a969..b47e501c 100644
--- a/extension/random/ext/export.go
+++ b/extensions/sources/random/random.go
@@ -16,7 +16,7 @@ package main
 
 import (
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/extension/random"
+	"github.com/lf-edge/ekuiper/v2/extensions/impl/random"
 )
 
 func Random() api.Source {
diff --git a/extension/random/random.json b/extensions/sources/random/random.json
similarity index 100%
rename from extension/random/random.json
rename to extensions/sources/random/random.json
diff --git a/extension/random/random.yaml b/extensions/sources/random/random.yaml
similarity index 100%
rename from extension/random/random.yaml
rename to extensions/sources/random/random.yaml
diff --git a/extensions/sources/sql/sql.go b/extensions/sources/sql/sql.go
new file mode 100644
index 00000000..a46392ed
--- /dev/null
+++ b/extensions/sources/sql/sql.go
@@ -0,0 +1,24 @@
+// Copyright 2022-2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/extensions/impl/sql"
+)
+
+func Sql() api.Source {
+	return sql.GetSource()
+}
diff --git a/extensions/sources/sql/sql.json b/extensions/sources/sql/sql.json
new file mode 100644
index 00000000..db92c36b
--- /dev/null
+++ b/extensions/sources/sql/sql.json
@@ -0,0 +1,348 @@
+{
+  "about": {
+    "trial": true,
+    "author": {
+      "name": "EMQ",
+      "email": "contact@emqx.io",
+      "company": "EMQ Technologies Co., Ltd",
+      "website": "https://www.emqx.io"
+    },
+    "helpUrl": {
+      "en_US": "https://ekuiper.org/docs/en/latest/guide/sources/plugin/sql.html",
+      "zh_CN": "https://ekuiper.org/docs/zh/latest/guide/sources/plugin/sql.html"
+    },
+    "description": {
+      "en_US": "Read message from sql database",
+      "zh_CN": "从数据库中读取消息"
+    }
+  },
+  "dataSource": {
+    "default": "table1",
+    "hint": {
+      "en_US": "The SQL database to connect to, e.g. table1",
+      "zh_CN": "将要连接的关系数据库表，例如 table1"
+    },
+    "label": {
+      "en_US": "Data Source (Table)",
+      "zh_CN": "数据源（表名）"
+    }
+  },
+  "libs": [],
+  "properties": {
+    "default": [
+      {
+        "name": "url",
+        "default": "",
+        "optional": false,
+        "control": "text",
+        "type": "string",
+        "hint": {
+          "en_US": "The url of the database, it is important to note that the password in the URL is url-encoded characters, and you need to re-enter the password when editing. For example, the database address of mysql is mysql://username:password@127.0.0.1:3306/testdb?parseTime=true, if the original password is 123:#?, the password here should be filled in with 123%3A%23%3F",
+          "zh_CN": "数据库服务器的 URL，需要特别注意的是 URL 中的密码为 url 编码后的字符，并且编辑时需要重新输入密码。例如，mysql 的数据库地址为 mysql://username:password@127.0.0.1:3306/testdb?parseTime=true，若原始密码为 123:#?，则这里的 password 应填入 123%3A%23%3F"
+        },
+        "label": {
+          "en_US": "Server address",
+          "zh_CN": "数据库地址"
+        }
+      },
+      {
+        "name": "interval",
+        "default": 1000,
+        "optional": true,
+        "control": "text",
+        "type": "int",
+        "hint": {
+          "en_US": "The interval (ms) to issue a query",
+          "zh_CN": "发出消息的间隔（毫秒）"
+        },
+        "label": {
+          "en_US": "Interval",
+          "zh_CN": "间隔时间"
+        }
+      },
+      {
+        "name": "lookup",
+        "default": {
+          "cache": {
+            "name": "cache",
+            "default": true,
+            "optional": true,
+            "control": "radio",
+            "type": "bool",
+            "hint": {
+              "en_US": "Whether to enable cache",
+              "zh_CN": "是否启用缓存"
+            },
+            "label": {
+              "en_US": "Enable lookup cache",
+              "zh_CN": "开启查询缓存"
+            }
+          },
+          "cacheTtl": {
+            "name": "cacheTtl",
+            "default": 600,
+            "optional": true,
+            "control": "text",
+            "type": "int",
+            "hint": {
+              "en_US": "The time to live of the cache in seconds",
+              "zh_CN": "缓存的生存时间（s)"
+            },
+            "label": {
+              "en_US": "Cache TTL",
+              "zh_CN": "缓存时间"
+            }
+          },
+          "cacheMissingKey": {
+            "name": "cacheMissingKey",
+            "default": true,
+            "optional": true,
+            "control": "radio",
+            "type": "bool",
+            "hint": {
+              "en_US": "Whether to cache nil value for a key",
+              "zh_CN": "是否对空值进行缓存"
+            },
+            "label": {
+              "en_US": "Cache missing key",
+              "zh_CN": "缓存未命中的 Key"
+            }
+          }
+        },
+        "optional": true,
+        "control": "list",
+        "type": "object",
+        "hint": {
+          "en_US": "Lookup table configuration, only effective when using as a lookup table",
+          "zh_CN": "查询表配置，仅在作为查询表使用时生效"
+        },
+        "label": {
+          "en_US": "Lookup table configuration",
+          "zh_CN": "查询表配置"
+        }
+      },
+      {
+        "name": "internalSqlQueryCfg",
+        "default": {
+          "table": {
+            "name": "table",
+            "default": "",
+            "optional": true,
+            "control": "text",
+            "type": "string",
+            "hint": {
+              "en_US": "table name to query",
+              "zh_CN": "指定查询的数据库表名"
+            },
+            "label": {
+              "en_US": "Table Name",
+              "zh_CN": "表名"
+            }
+          },
+          "indexField": {
+            "name": "indexField",
+            "default": "",
+            "optional": true,
+            "control": "text",
+            "type": "string",
+            "hint": {
+              "en_US": "index field",
+              "zh_CN": "索引字段名"
+            },
+            "label": {
+              "en_US": "Index Field",
+              "zh_CN": "索引字段名"
+            }
+          },
+          "indexValue": {
+            "name": "indexValue",
+            "default": "",
+            "optional": true,
+            "control": "text",
+            "type": "string",
+            "hint": {
+              "en_US": "index init value",
+              "zh_CN": "索引字段初始值"
+            },
+            "label": {
+              "en_US": "Index init value",
+              "zh_CN": "索引字段初始值"
+            }
+          },
+          "limit": {
+            "name": "limit",
+            "default": 10,
+            "optional": true,
+            "control": "text",
+            "type": "int",
+            "hint": {
+              "en_US": "query result limit",
+              "zh_CN": "查询结果条数限制"
+            },
+            "label": {
+              "en_US": "Limit",
+              "zh_CN": "查询条数限制"
+            }
+          },
+          "indexFieldType": {
+            "name": "indexFieldType",
+            "default": "",
+            "optional": true,
+            "control": "select",
+            "type": "string",
+            "values": [
+              "DATETIME",
+              "INT"
+            ],
+            "hint": {
+              "en_US": "Column type for the indexField",
+              "zh_CN": "索引字段的列类型"
+            },
+            "label": {
+              "en_US": "Index Field Type",
+              "zh_CN": "索引字段格式"
+            }
+          },
+          "dateTimeFormat": {
+            "name": "dateTimeFormat",
+            "default": "",
+            "optional": true,
+            "control": "text",
+            "type": "string",
+            "hint": {
+              "en_US": "dateTimeFormat",
+              "zh_CN": "dateTimeFormat"
+            },
+            "label": {
+              "en_US": "DateTime Format",
+              "zh_CN": "时间格式"
+            }
+          }
+        },
+        "optional": true,
+        "control": "list",
+        "type": "object",
+        "hint": {
+          "en_US": "basic configuration for the query",
+          "zh_CN": "查询基础配置"
+        },
+        "label": {
+          "en_US": "Query Parameters",
+          "zh_CN": "查询参数"
+        }
+      },
+      {
+        "name": "templateSqlQueryCfg",
+        "default": {
+          "TemplateSql": {
+            "name": "TemplateSql",
+            "default": "",
+            "optional": false,
+            "control": "textarea",
+            "type": "string",
+            "hint": {
+              "en_US": "SQL statement template",
+              "zh_CN": "SQL 语句模版"
+            },
+            "label": {
+              "en_US": "Query template",
+              "zh_CN": "查询语句模版"
+            }
+          },
+          "indexField": {
+            "name": "indexField",
+            "default": "",
+            "optional": true,
+            "control": "text",
+            "type": "string",
+            "hint": {
+              "en_US": "Which column for the table act as index to record the offset",
+              "zh_CN": "表的哪一列作为索引来记录偏移量"
+            },
+            "label": {
+              "en_US": "Index Field",
+              "zh_CN": "索引字段名"
+            }
+          },
+          "indexValue": {
+            "name": "indexValue",
+            "default": "",
+            "optional": true,
+            "control": "text",
+            "type": "string",
+            "hint": {
+              "en_US": "Initial index value",
+              "zh_CN": "初始索引值"
+            },
+            "label": {
+              "en_US": "Index init value",
+              "zh_CN": "索引字段初始值"
+            }
+          },
+          "indexFieldType": {
+            "name": "indexFieldType",
+            "default": "",
+            "optional": true,
+            "control": "select",
+            "type": "string",
+            "values": [
+              "DATETIME",
+              "INT"
+            ],
+            "hint": {
+              "en_US": "Column type for the indexField",
+              "zh_CN": "索引字段的列类型"
+            },
+            "label": {
+              "en_US": "Index Field Type",
+              "zh_CN": "索引字段格式"
+            }
+          },
+          "dateTimeFormat": {
+            "name": "dateTimeFormat",
+            "default": "",
+            "optional": true,
+            "control": "text",
+            "type": "string",
+            "hint": {
+              "en_US": "Data time format for the index field",
+              "zh_CN": "索引字段的时间格式"
+            },
+            "label": {
+              "en_US": "DateTime Format",
+              "zh_CN": "时间格式"
+            }
+          }
+        },
+        "optional": true,
+        "control": "list",
+        "type": "object",
+        "hint": {
+          "en_US": "Query Template",
+          "zh_CN": "查询模版"
+        },
+        "label": {
+          "en_US": "Query Template",
+          "zh_CN": "查询模版"
+        }
+      }
+    ]
+  },
+  "outputs": [
+    {
+      "label": {
+        "en_US": "Output",
+        "zh_CN": "输出"
+      },
+      "value": "signal"
+    }
+  ],
+  "node": {
+    "category": "source",
+    "icon": "iconPath",
+    "label": {
+      "en_US": "SQL",
+      "zh_CN": "SQL"
+    }
+  }
+}
diff --git a/extensions/sources/sql/sql.yaml b/extensions/sources/sql/sql.yaml
new file mode 100644
index 00000000..9de68325
--- /dev/null
+++ b/extensions/sources/sql/sql.yaml
@@ -0,0 +1,17 @@
+default:
+  url: mysql://username:password@127.0.0.1:3306/testdb?parseTime=true
+
+mysql_config:
+  url: mysql://username:password@127.0.0.1:3306/testdb?parseTime=true
+
+postgresql_config:
+  url: postgres://username:password@127.0.0.1:5432/testdb
+
+sqlite_config:
+  url: sqlite:/tmp/test.db
+
+sqlserver_config:
+  url: sqlserver://username:password@127.0.0.1:1433/testdb
+
+oracle_config:
+  url: oracle://username:password@127.0.0.1:1521/testdb
diff --git a/extensions/sources/video/install.sh b/extensions/sources/video/install.sh
new file mode 100644
index 00000000..3cdc2737
--- /dev/null
+++ b/extensions/sources/video/install.sh
@@ -0,0 +1,63 @@
+#!/bin/sh
+#
+# 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.
+#
+
+set +e -x -u
+
+DISTRO='unknow'
+
+Get_Dist_Name()
+{
+    if grep -Eqii "CentOS" /etc/issue || grep -Eq "CentOS" /etc/*-release; then
+        DISTRO='CentOS'
+    elif grep -Eqi "Red Hat Enterprise Linux Server" /etc/issue || grep -Eq "Red Hat Enterprise Linux Server" /etc/*-release; then
+        DISTRO='RHEL'
+    elif grep -Eqi "Aliyun" /etc/issue || grep -Eq "Aliyun" /etc/*-release; then
+        DISTRO='Aliyun'
+    elif grep -Eqi "Fedora" /etc/issue || grep -Eq "Fedora" /etc/*-release; then
+        DISTRO='Fedora'
+    elif grep -Eqi "Debian" /etc/issue || grep -Eq "Debian" /etc/*-release; then
+        DISTRO='Debian'
+    elif grep -Eqi "Ubuntu" /etc/issue || grep -Eq "Ubuntu" /etc/*-release; then
+        DISTRO='Ubuntu'
+    elif grep -Eqi "Raspbian" /etc/issue || grep -Eq "Raspbian" /etc/*-release; then
+        DISTRO='Raspbian'
+    elif grep -Eqi "Alpine" /etc/issue || grep -Eq "Alpine" /etc/*-release; then
+        DISTRO='Alpine'
+    else
+        DISTRO='unknow'
+    fi
+    echo $DISTRO;
+}
+
+
+Get_Dist_Name
+
+case $DISTRO in \
+    Debian|Ubuntu|Raspbian ) \
+	apt update \
+	&& apt upgrade \
+        && apt install -y ffmpeg 2> /dev/null \
+    ;; \
+    Alpine ) \
+        apk add ffmpeg 2> /dev/null \
+    ;; \
+    *) \
+        yum install -y ffmpeg 2> /dev/null \
+    ;; \
+esac
+    
+echo "install success";
diff --git a/extensions/sources/video/video.go b/extensions/sources/video/video.go
new file mode 100644
index 00000000..9808bd8b
--- /dev/null
+++ b/extensions/sources/video/video.go
@@ -0,0 +1,24 @@
+// Copyright 2022-2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/extensions/impl/video"
+)
+
+func Video() api.Source {
+	return video.GetSource()
+}
diff --git a/extensions/sources/video/video.json b/extensions/sources/video/video.json
new file mode 100644
index 00000000..e27ab964
--- /dev/null
+++ b/extensions/sources/video/video.json
@@ -0,0 +1,72 @@
+{
+  "libs": [],
+  "about": {
+    "trial": false,
+    "author": {
+      "name": "EMQ",
+      "email": "contact@emqx.io",
+      "company": "EMQ Technologies Co., Ltd",
+      "website": "https://www.emqx.io"
+    },
+    "helpUrl": {
+      "en_US": "https://ekuiper.org/docs/en/latest/guide/sources/plugin/video.html",
+      "zh_CN": "https://ekuiper.org/docs/zh/latest/guide/sources/plugin/video.html"
+    },
+    "description": {
+      "en_US": "Pull images from a video stream",
+      "zh_CN": "可从视频流提取图片消息并输入到处理管道。"
+    }
+  },
+  "dataSource": {},
+  "properties": {
+    "default": [
+      {
+        "name": "url",
+        "default": "127.0.0.1:5536",
+        "optional": false,
+        "control": "text",
+        "type": "string",
+        "hint": {
+          "en_US": "The URL where to get the result.",
+          "zh_CN": "获取结果的 URL"
+        },
+        "label": {
+          "en_US": "URL",
+          "zh_CN": "路径"
+        }
+      },
+      {
+        "name": "interval",
+        "default": 1000,
+        "optional": true,
+        "control": "text",
+        "type": "int",
+        "hint": {
+          "en_US": "The interval between the requests, time unit is ms.",
+          "zh_CN": "请求之间的间隔时间，单位为 ms"
+        },
+        "label": {
+          "en_US": "Interval",
+          "zh_CN": "间隔时间"
+        }
+      }
+    ]
+  },
+  "outputs": [
+    {
+      "label": {
+        "en_US": "Output",
+        "zh_CN": "输出"
+      },
+      "value": "signal"
+    }
+  ],
+  "node": {
+    "category": "source",
+    "icon": "iconPath",
+    "label": {
+      "en_US": "Video",
+      "zh_CN": "视频"
+    }
+  }
+}
diff --git a/extensions/sources/video/video.yaml b/extensions/sources/video/video.yaml
new file mode 100644
index 00000000..3485f860
--- /dev/null
+++ b/extensions/sources/video/video.yaml
@@ -0,0 +1,8 @@
+#Global video source configurations
+default:
+  # url of the request server address
+  # username password, can be this format: rtsp://xxxxxx:xxxxxx@192.168.0.64:554/h265/ch1/main/av_stream
+  url: http://localhost
+  # The interval between the requests, time unit is ms
+  interval: 10000
+  # The timeout for http request, time unit is ms
\ No newline at end of file
diff --git a/internal/binder/io/builtin.go b/internal/binder/io/builtin.go
index eb7cc113..b2c8be2c 100644
--- a/internal/binder/io/builtin.go
+++ b/internal/binder/io/builtin.go
@@ -23,6 +23,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/io/mqtt"
 	mqttCon "github.com/lf-edge/ekuiper/v2/internal/io/mqtt/client"
 	"github.com/lf-edge/ekuiper/v2/internal/io/neuron"
+	"github.com/lf-edge/ekuiper/v2/internal/io/simulator"
 	"github.com/lf-edge/ekuiper/v2/internal/io/sink"
 	"github.com/lf-edge/ekuiper/v2/internal/io/websocket"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
@@ -37,7 +38,7 @@ func init() {
 	modules.RegisterSource("memory", func() api.Source { return memory.GetSource() })
 	modules.RegisterSource("neuron", neuron.GetSource)
 	modules.RegisterSource("websocket", func() api.Source { return websocket.GetSource() })
-	// modules.RegisterSource("simulator", func() api.Source { return &simulator.Source{} })
+	modules.RegisterSource("simulator", func() api.Source { return simulator.GetSource() })
 
 	modules.RegisterSink("log", sink.NewLogSink)
 	modules.RegisterSink("logToMemory", sink.NewLogSinkToMemory)
diff --git a/internal/binder/io/ext_full.go b/internal/binder/io/ext_full.go
index 6f79423d..15b773eb 100644
--- a/internal/binder/io/ext_full.go
+++ b/internal/binder/io/ext_full.go
@@ -18,9 +18,9 @@ package io
 
 import (
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/extension/sql"
-	"github.com/lf-edge/ekuiper/v2/extension/sql/client"
-	"github.com/lf-edge/ekuiper/v2/extension/video"
+	sql2 "github.com/lf-edge/ekuiper/v2/extensions/impl/sql"
+	"github.com/lf-edge/ekuiper/v2/extensions/impl/sql/client"
+	"github.com/lf-edge/ekuiper/v2/extensions/impl/video"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
 
@@ -33,7 +33,7 @@ func init() {
 	//modules.RegisterSink("influx", func() api.Sink { return influx.GetSink() })
 	//modules.RegisterSink("influx2", func() api.Sink { return influx2.GetSink() })
 	//modules.RegisterSink("kafka", func() api.Sink { return kafka.GetSink() })
-	modules.RegisterSource("sql", sql.GetSource)
-	modules.RegisterSink("sql", sql.GetSink)
+	modules.RegisterSource("sql", sql2.GetSource)
+	modules.RegisterSink("sql", sql2.GetSink)
 	modules.RegisterConnection("sql", client.CreateConnection)
 }
diff --git a/internal/io/http/httpserver/websocketConn.go b/internal/io/http/httpserver/websocketConn.go
index c106b974..603f911f 100644
--- a/internal/io/http/httpserver/websocketConn.go
+++ b/internal/io/http/httpserver/websocketConn.go
@@ -17,13 +17,22 @@ package httpserver
 import (
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/cert"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
 
 type WebsocketConnection struct {
 	RecvTopic string
 	SendTopic string
-	cfg       *connectionCfg
+	props     map[string]any
+	cfg       *wscConfig
+	isServer  bool
+	client    *WebsocketClient
+}
+
+type wscConfig struct {
+	Datasource string `json:"datasource"`
+	Addr       string `json:"addr"`
 }
 
 func (w *WebsocketConnection) Ping(ctx api.StreamContext) error {
@@ -31,10 +40,14 @@ func (w *WebsocketConnection) Ping(ctx api.StreamContext) error {
 }
 
 func (w *WebsocketConnection) DetachSub(ctx api.StreamContext, props map[string]any) {
-	UnRegisterWebSocketEndpoint(w.cfg.Datasource)
 }
 
 func (w *WebsocketConnection) Close(ctx api.StreamContext) error {
+	if w.isServer {
+		UnRegisterWebSocketEndpoint(w.cfg.Datasource)
+	} else {
+		w.client.Close(ctx)
+	}
 	return nil
 }
 
@@ -43,17 +56,40 @@ func CreateWebsocketConnection(ctx api.StreamContext, props map[string]any) (mod
 }
 
 func createWebsocketServerConnection(ctx api.StreamContext, props map[string]any) (*WebsocketConnection, error) {
-	cfg := &connectionCfg{}
+	cfg := &wscConfig{}
 	if err := cast.MapToStruct(props, cfg); err != nil {
 		return nil, err
 	}
-	rTopic, sTopic, err := RegisterWebSocketEndpoint(ctx, cfg.Datasource)
-	if err != nil {
-		return nil, err
+	wc := &WebsocketConnection{
+		props:    props,
+		cfg:      cfg,
+		isServer: getWsType(cfg),
+	}
+	if wc.isServer {
+		rTopic, sTopic, err := RegisterWebSocketEndpoint(ctx, cfg.Datasource)
+		if err != nil {
+			return nil, err
+		}
+		wc.RecvTopic = rTopic
+		wc.SendTopic = sTopic
+	} else {
+		tlsConfig, err := cert.GenTLSConfig(props, "websocket")
+		if err != nil {
+			return nil, err
+		}
+		c := NewWebsocketClient(cfg.Addr, cfg.Datasource, tlsConfig)
+		if err := c.Connect(); err != nil {
+			return nil, err
+		}
+		wc.client = c
+		wc.RecvTopic, wc.SendTopic = c.Run(ctx)
+	}
+	return wc, nil
+}
+
+func getWsType(cfg *wscConfig) bool {
+	if len(cfg.Addr) < 1 {
+		return true
 	}
-	return &WebsocketConnection{
-		RecvTopic: rTopic,
-		SendTopic: sTopic,
-		cfg:       cfg,
-	}, nil
+	return false
 }
diff --git a/internal/io/http/httpserver/websocket_client.go b/internal/io/http/httpserver/websocket_client.go
new file mode 100644
index 00000000..668011ad
--- /dev/null
+++ b/internal/io/http/httpserver/websocket_client.go
@@ -0,0 +1,90 @@
+// 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 httpserver
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"net/url"
+	"sync"
+	"time"
+
+	"github.com/gorilla/websocket"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/io/memory/pubsub"
+)
+
+type WebsocketClient struct {
+	RecvTopic string
+	SendTopic string
+
+	addr      string
+	path      string
+	tlsConfig *tls.Config
+	conn      *websocket.Conn
+	wg        *sync.WaitGroup
+	cancel    context.CancelFunc
+}
+
+func NewWebsocketClient(addr, path string, tlsConfig *tls.Config) *WebsocketClient {
+	return &WebsocketClient{
+		addr:      addr,
+		path:      path,
+		tlsConfig: tlsConfig,
+		wg:        &sync.WaitGroup{},
+	}
+}
+
+func (c *WebsocketClient) Connect() error {
+	d := &websocket.Dialer{
+		HandshakeTimeout: 3 * time.Second,
+		TLSClientConfig:  c.tlsConfig,
+	}
+	if len(c.addr) < 1 {
+		return fmt.Errorf("addr should be defined")
+	}
+	u := url.URL{Scheme: "ws", Host: c.addr, Path: c.path}
+	conn, _, err := d.Dial(u.String(), nil)
+	if err != nil {
+		return err
+	}
+	c.conn = conn
+	return nil
+}
+
+func (c *WebsocketClient) Run(ctx api.StreamContext) (string, string) {
+	c.RecvTopic = recvTopic(c.path, false)
+	c.SendTopic = sendTopic(c.path, false)
+	pubsub.CreatePub(c.RecvTopic)
+	c.handleProcess(ctx)
+	return c.RecvTopic, c.SendTopic
+}
+
+func (c *WebsocketClient) handleProcess(parCtx api.StreamContext) {
+	ctx, cancel := parCtx.WithCancel()
+	c.cancel = cancel
+	c.wg.Add(2)
+	go recvProcess(ctx, c.RecvTopic, c.conn, cancel, c.wg)
+	go sendProcess(ctx, c.SendTopic, "", c.conn, cancel, c.wg)
+}
+
+func (c *WebsocketClient) Close(ctx api.StreamContext) error {
+	pubsub.RemovePub(c.RecvTopic)
+	c.cancel()
+	c.wg.Wait()
+	return nil
+}
diff --git a/internal/io/http/httpserver/websocket_server.go b/internal/io/http/httpserver/websocket_server.go
index fbef1d61..f03fd9a2 100644
--- a/internal/io/http/httpserver/websocket_server.go
+++ b/internal/io/http/httpserver/websocket_server.go
@@ -32,12 +32,20 @@ const (
 	WebsocketTopicPrefix = "$$websocket/"
 )
 
-func recvTopic(endpoint string) string {
-	return fmt.Sprintf("recv/%s/%s", WebsocketTopicPrefix, endpoint)
+func recvTopic(endpoint string, isServer bool) string {
+	if isServer {
+		return fmt.Sprintf("%s/server/recv/%s", WebsocketTopicPrefix, endpoint)
+	} else {
+		return fmt.Sprintf("%s/client/recv/%s", WebsocketTopicPrefix, endpoint)
+	}
 }
 
-func sendTopic(endpoint string) string {
-	return fmt.Sprintf("send/%s/%s", WebsocketTopicPrefix, endpoint)
+func sendTopic(endpoint string, isServer bool) string {
+	if isServer {
+		return fmt.Sprintf("%s/server/send/%s", WebsocketTopicPrefix, endpoint)
+	} else {
+		return fmt.Sprintf("%s/client/send/%s", WebsocketTopicPrefix, endpoint)
+	}
 }
 
 type websocketEndpointContext struct {
@@ -64,21 +72,17 @@ func (m *GlobalServerManager) handleProcess(ctx api.StreamContext, endpoint stri
 	}()
 	subWg := &sync.WaitGroup{}
 	subWg.Add(2)
-	go m.recvProcess(ctx, endpoint, c, cancel, subWg)
-	go m.sendProcess(ctx, endpoint, instanceID, c, cancel, subWg)
+	go recvProcess(ctx, recvTopic(endpoint, true), c, cancel, subWg)
+	go sendProcess(ctx, sendTopic(endpoint, true), fmt.Sprintf("ws/send/%v", instanceID), c, cancel, subWg)
 	subWg.Wait()
 }
 
-func (m *GlobalServerManager) sendProcess(ctx api.StreamContext, endpoint string, instanceID int, c *websocket.Conn, cancel context.CancelFunc, wg *sync.WaitGroup) {
-	conf.Log.Infof("websocket endpoint %v start sendProcess", endpoint)
-	topic := sendTopic(endpoint)
-	sourceID := fmt.Sprintf("ws/send/%v", instanceID)
+func sendProcess(ctx api.StreamContext, topic, sourceID string, c *websocket.Conn, cancel context.CancelFunc, wg *sync.WaitGroup) {
 	defer func() {
 		pubsub.CloseSourceConsumerChannel(topic, sourceID)
 		cancel()
 		c.Close()
 		wg.Done()
-		conf.Log.Infof("websocket send endpoint %v stop sendProcess", endpoint)
 	}()
 	ch := pubsub.CreateSub(topic, nil, sourceID, 1024)
 	for {
@@ -89,24 +93,19 @@ func (m *GlobalServerManager) sendProcess(ctx api.StreamContext, endpoint string
 			data := d.([]byte)
 			if err := c.WriteMessage(websocket.TextMessage, data); err != nil {
 				if websocket.IsCloseError(err) || strings.Contains(err.Error(), "close") {
-					conf.Log.Infof("websocket endpoint %s connection get closed: %v", endpoint, err)
 					return
 				}
-				conf.Log.Warnf("websocket endpoint %v send data meet error: %v", endpoint, err)
 			}
 		}
 	}
 }
 
-func (m *GlobalServerManager) recvProcess(ctx api.StreamContext, endpoint string, c *websocket.Conn, cancel context.CancelFunc, wg *sync.WaitGroup) {
+func recvProcess(ctx api.StreamContext, topic string, c *websocket.Conn, cancel context.CancelFunc, wg *sync.WaitGroup) {
 	defer func() {
 		cancel()
 		c.Close()
 		wg.Done()
-		conf.Log.Infof("websocket recv endpoint %v stop recvProcess", endpoint)
 	}()
-	conf.Log.Infof("websocket endpoint %v start recvProcess", endpoint)
-	topic := recvTopic(endpoint)
 	for {
 		select {
 		case <-ctx.Done():
@@ -116,17 +115,14 @@ func (m *GlobalServerManager) recvProcess(ctx api.StreamContext, endpoint string
 		msgType, data, err := c.ReadMessage()
 		if err != nil {
 			if websocket.IsCloseError(err) || strings.Contains(err.Error(), "close") {
-				conf.Log.Infof("websocket endpoint %s connection get closed: %v", endpoint, err)
 				return
 			}
-			conf.Log.Errorf("websocket endpoint %s recv error %s", endpoint, err)
 			continue
 		}
 		switch msgType {
 		case websocket.TextMessage:
 			pubsub.ProduceAny(ctx, topic, data)
 		default:
-			conf.Log.Infof("websocker endpoint %v recv other message typ %v", endpoint, msgType)
 		}
 	}
 }
@@ -135,8 +131,8 @@ func (m *GlobalServerManager) RegisterWebSocketEndpoint(ctx api.StreamContext, e
 	conf.Log.Infof("websocket endpoint %v register", endpoint)
 	m.Lock()
 	defer m.Unlock()
-	rTopic := recvTopic(endpoint)
-	sTopic := sendTopic(endpoint)
+	rTopic := recvTopic(endpoint, true)
+	sTopic := sendTopic(endpoint, true)
 	pubsub.CreatePub(rTopic)
 	m.router.HandleFunc(endpoint, func(w http.ResponseWriter, r *http.Request) {
 		c, err := m.upgrader.Upgrade(w, r, nil)
@@ -155,7 +151,7 @@ func (m *GlobalServerManager) RegisterWebSocketEndpoint(ctx api.StreamContext, e
 
 func (m *GlobalServerManager) UnRegisterWebSocketEndpoint(endpoint string) *websocketEndpointContext {
 	conf.Log.Infof("websocket endpoint %v unregister", endpoint)
-	pubsub.RemovePub(recvTopic(endpoint))
+	pubsub.RemovePub(recvTopic(endpoint, true))
 	m.Lock()
 	defer m.Unlock()
 	wctx, ok := m.websocketEndpoint[endpoint]
diff --git a/internal/io/simulator/simulator.go b/internal/io/simulator/simulator.go
new file mode 100644
index 00000000..2c8c2883
--- /dev/null
+++ b/internal/io/simulator/simulator.go
@@ -0,0 +1,69 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package simulator
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+)
+
+type SimulatorSource struct {
+	index int
+	cfg   *sConfig
+}
+
+type sConfig struct {
+	Data []map[string]any `json:"data"`
+	Loop bool             `json:"loop"`
+}
+
+func (s *SimulatorSource) Provision(ctx api.StreamContext, configs map[string]any) error {
+	cfg := &sConfig{}
+	if err := cast.MapToStruct(configs, cfg); err != nil {
+		return err
+	}
+	s.cfg = cfg
+	return nil
+}
+
+func (s SimulatorSource) Close(ctx api.StreamContext) error {
+	return nil
+}
+
+func (s *SimulatorSource) Connect(ctx api.StreamContext) error {
+	return nil
+}
+
+func (s *SimulatorSource) Pull(ctx api.StreamContext, trigger time.Time, ingest api.TupleIngest, ingestError api.ErrorIngest) {
+	if s.index >= len(s.cfg.Data) {
+		if s.cfg.Loop {
+			s.index = 0
+		} else {
+			ingestError(ctx, fmt.Errorf("simulator source message running out"))
+			return
+		}
+	}
+	ingest(ctx, s.cfg.Data[s.index], nil, trigger)
+	s.index++
+}
+
+func GetSource() api.Source {
+	return &SimulatorSource{}
+}
+
+var _ api.PullTupleSource = &SimulatorSource{}
diff --git a/internal/io/websocket/websocket_sink.go b/internal/io/websocket/websocket_sink.go
index 613a21d5..ad56d6ac 100644
--- a/internal/io/websocket/websocket_sink.go
+++ b/internal/io/websocket/websocket_sink.go
@@ -32,6 +32,7 @@ type WebsocketSink struct {
 }
 
 func (w *WebsocketSink) Provision(ctx api.StreamContext, configs map[string]any) error {
+	configs = solveProps(configs)
 	cfg := &WebsocketConfig{}
 	if err := cast.MapToStruct(configs, cfg); err != nil {
 		return err
diff --git a/internal/io/websocket/websocket_source.go b/internal/io/websocket/websocket_source.go
index 5dc6d2a1..8e06e27c 100644
--- a/internal/io/websocket/websocket_source.go
+++ b/internal/io/websocket/websocket_source.go
@@ -39,6 +39,7 @@ type WebsocketConfig struct {
 }
 
 func (w *WebsocketSource) Provision(ctx api.StreamContext, configs map[string]any) error {
+	configs = solveProps(configs)
 	cfg := &WebsocketConfig{}
 	if err := cast.MapToStruct(configs, cfg); err != nil {
 		return err
@@ -89,6 +90,15 @@ func (w *WebsocketSource) Subscribe(ctx api.StreamContext, ingest api.BytesInges
 	return nil
 }
 
+func solveProps(props map[string]any) map[string]any {
+	v1, ok1 := props["path"]
+	_, ok2 := props["datasource"]
+	if ok1 && !ok2 {
+		props["datasource"] = v1
+	}
+	return props
+}
+
 func GetSource() api.Source {
 	return &WebsocketSource{}
 }
diff --git a/internal/pkg/def/rule.go b/internal/pkg/def/rule.go
index 0184bba2..2d6ea646 100644
--- a/internal/pkg/def/rule.go
+++ b/internal/pkg/def/rule.go
@@ -21,27 +21,27 @@ import (
 )
 
 type RuleOption struct {
-	Debug              bool              `json:"debug" yaml:"debug"`
-	LogFilename        string            `json:"logFilename" yaml:"logFilename"`
-	IsEventTime        bool              `json:"isEventTime" yaml:"isEventTime"`
-	LateTol            cast.DurationConf `json:"lateTolerance" yaml:"lateTolerance"`
+	Debug              bool              `json:"debug,omitempty" yaml:"debug,omitempty"`
+	LogFilename        string            `json:"logFilename,omitempty" yaml:"logFilename,omitempty"`
+	IsEventTime        bool              `json:"isEventTime,omitempty" yaml:"isEventTime,omitempty"`
+	LateTol            cast.DurationConf `json:"lateTolerance,omitempty" yaml:"lateTolerance,omitempty"`
 	Concurrency        int               `json:"concurrency" yaml:"concurrency"`
 	BufferLength       int               `json:"bufferLength" yaml:"bufferLength"`
-	SendMetaToSink     bool              `json:"sendMetaToSink" yaml:"sendMetaToSink"`
-	SendError          bool              `json:"sendError" yaml:"sendError"`
-	Qos                Qos               `json:"qos" yaml:"qos"`
-	CheckpointInterval cast.DurationConf `json:"checkpointInterval" yaml:"checkpointInterval"`
-	Restart            *RestartStrategy  `json:"restartStrategy" yaml:"restartStrategy"`
-	Cron               string            `json:"cron" yaml:"cron"`
-	Duration           string            `json:"duration" yaml:"duration"`
-	CronDatetimeRange  []DatetimeRange   `json:"cronDatetimeRange" yaml:"cronDatetimeRange"`
+	SendMetaToSink     bool              `json:"sendMetaToSink,omitempty" yaml:"sendMetaToSink,omitempty"`
+	SendError          bool              `json:"sendError,omitempty" yaml:"sendError,omitempty"`
+	Qos                Qos               `json:"qos,omitempty" yaml:"qos,omitempty"`
+	CheckpointInterval cast.DurationConf `json:"checkpointInterval,omitempty" yaml:"checkpointInterval,omitempty"`
+	Restart            *RestartStrategy  `json:"restartStrategy,omitempty" yaml:"restartStrategy,omitempty"`
+	Cron               string            `json:"cron,omitempty" yaml:"cron,omitempty"`
+	Duration           string            `json:"duration,omitempty" yaml:"duration,omitempty"`
+	CronDatetimeRange  []DatetimeRange   `json:"cronDatetimeRange,omitempty" yaml:"cronDatetimeRange,omitempty"`
 }
 
 type DatetimeRange struct {
 	Begin          string `json:"begin" yaml:"begin"`
 	End            string `json:"end" yaml:"end"`
-	BeginTimestamp int64  `json:"beginTimestamp"`
-	EndTimestamp   int64  `json:"endTimestamp"`
+	BeginTimestamp int64  `json:"beginTimestamp" yaml:"beginTimestamp"`
+	EndTimestamp   int64  `json:"endTimestamp" yaml:"endTimestamp"`
 }
 
 type RestartStrategy struct {
@@ -53,16 +53,16 @@ type RestartStrategy struct {
 }
 
 type PrintableTopo struct {
-	Sources []string                 `json:"sources"`
-	Edges   map[string][]interface{} `json:"edges"`
+	Sources []string                 `json:"sources" yaml:"sources"`
+	Edges   map[string][]interface{} `json:"edges" yaml:"edges"`
 }
 
 type GraphNode struct {
-	Type     string                 `json:"type"`
-	NodeType string                 `json:"nodeType"`
-	Props    map[string]interface{} `json:"props"`
+	Type     string                 `json:"type" yaml:"type"`
+	NodeType string                 `json:"nodeType" yaml:"nodeType"`
+	Props    map[string]interface{} `json:"props" yaml:"props"`
 	// UI is a placeholder for ui properties
-	UI map[string]interface{} `json:"ui"`
+	UI map[string]interface{} `json:"ui" yaml:"ui"`
 }
 
 // SourceMeta is the metadata of a source node. It describes what existed stream/table to refer to.
@@ -73,20 +73,20 @@ type SourceMeta struct {
 }
 
 type RuleGraph struct {
-	Nodes map[string]*GraphNode `json:"nodes"`
-	Topo  *PrintableTopo        `json:"topo"`
+	Nodes map[string]*GraphNode `json:"nodes" yaml:"nodes"`
+	Topo  *PrintableTopo        `json:"topo" yaml:"topo"`
 }
 
 // Rule the definition of the business logic
 // Sql and Graph are mutually exclusive, at least one of them should be set
 type Rule struct {
-	Triggered bool                     `json:"triggered"`
-	Id        string                   `json:"id,omitempty"`
-	Name      string                   `json:"name,omitempty"` // The display name of a rule
-	Sql       string                   `json:"sql,omitempty"`
-	Graph     *RuleGraph               `json:"graph,omitempty"`
-	Actions   []map[string]interface{} `json:"actions,omitempty"`
-	Options   *RuleOption              `json:"options,omitempty"`
+	Triggered bool                     `json:"triggered,omitempty" yaml:"triggered,omitempty"`
+	Id        string                   `json:"id,omitempty" yaml:"id,omitempty"`
+	Name      string                   `json:"name,omitempty" yaml:"name,omitempty"` // The display name of a rule
+	Sql       string                   `json:"sql,omitempty" yaml:"sql,omitempty"`
+	Graph     *RuleGraph               `json:"graph,omitempty" yaml:"graph,omitempty"`
+	Actions   []map[string]interface{} `json:"actions,omitempty" yaml:"actions,omitempty"`
+	Options   *RuleOption              `json:"options,omitempty" yaml:"options,omitempty"`
 }
 
 func (r *Rule) IsLongRunningScheduleRule() bool {
diff --git a/internal/plugin/plugin.go b/internal/plugin/plugin.go
index f7a49a65..aa3cd6d7 100644
--- a/internal/plugin/plugin.go
+++ b/internal/plugin/plugin.go
@@ -45,9 +45,9 @@ type Plugin interface {
 
 // IOPlugin Unify model. Flat all properties for each kind.
 type IOPlugin struct {
-	Name       string   `json:"name"`
-	File       string   `json:"file"`
-	ShellParas []string `json:"shellParas"`
+	Name       string   `json:"name" yaml:"name"`
+	File       string   `json:"file" yaml:"file"`
+	ShellParas []string `json:"shellParas,omitempty" yaml:"shellParas,omitempty"`
 }
 
 func (p *IOPlugin) GetName() string {
diff --git a/internal/schema/schema.go b/internal/schema/schema.go
index 07e303ee..bdab489d 100644
--- a/internal/schema/schema.go
+++ b/internal/schema/schema.go
@@ -22,11 +22,11 @@ import (
 )
 
 type Info struct {
-	Type     def.SchemaType `json:"type"`
-	Name     string         `json:"name"`
-	Content  string         `json:"content"`
-	FilePath string         `json:"file"`
-	SoPath   string         `json:"soFile"`
+	Type     def.SchemaType `json:"type" yaml:"type"`
+	Name     string         `json:"name" yaml:"name"`
+	Content  string         `json:"content,omitempty" yaml:"content,omitempty"`
+	FilePath string         `json:"file,omitempty" yaml:"filePath,omitempty"`
+	SoPath   string         `json:"soFile,omitempty" yaml:"soPath,omitempty"`
 }
 
 func (i *Info) InstallScript() string {
diff --git a/internal/server/rest.go b/internal/server/rest.go
index 350f58ab..c6e5341a 100644
--- a/internal/server/rest.go
+++ b/internal/server/rest.go
@@ -179,6 +179,9 @@ func createRestServer(ip string, port int, needToken bool) *http.Server {
 	r.HandleFunc("/connections", connectionsHandler).Methods(http.MethodGet, http.MethodPost)
 	r.HandleFunc("/connections/status", connectionsStatusHandler).Methods(http.MethodGet)
 	r.HandleFunc("/connection/{id}", connectionHandler).Methods(http.MethodGet, http.MethodDelete)
+
+	r.HandleFunc("/v2/data/export", yamlConfigurationExportHandler).Methods(http.MethodGet)
+
 	// r.HandleFunc("/connection/websocket", connectionHandler).Methods(http.MethodGet, http.MethodPost, http.MethodDelete)
 	// Register extended routes
 	for k, v := range components {
@@ -203,9 +206,9 @@ func createRestServer(ip string, port int, needToken bool) *http.Server {
 }
 
 type fileContent struct {
-	Name     string `json:"name"`
-	Content  string `json:"content"`
-	FilePath string `json:"file"`
+	Name     string `json:"name" yaml:"name"`
+	Content  string `json:"content,omitempty" yaml:"content,omitempty"`
+	FilePath string `json:"file,omitempty" yaml:"filePath,omitempty"`
 }
 
 func (f *fileContent) InstallScript() string {
diff --git a/internal/server/yaml_import_export.go b/internal/server/yaml_import_export.go
new file mode 100644
index 00000000..601105e6
--- /dev/null
+++ b/internal/server/yaml_import_export.go
@@ -0,0 +1,252 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"time"
+
+	"github.com/pingcap/failpoint"
+	"gopkg.in/yaml.v3"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
+	"github.com/lf-edge/ekuiper/v2/internal/plugin"
+	"github.com/lf-edge/ekuiper/v2/internal/schema"
+	"github.com/lf-edge/ekuiper/v2/internal/service"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+)
+
+type MetaConfiguration struct {
+	SourceConfig     map[string]map[string]any `json:"sourceConfig,omitempty" yaml:"sourceConfig,omitempty"`
+	SinkConfig       map[string]map[string]any `json:"sinkConfig,omitempty" yaml:"sinkConfig,omitempty"`
+	ConnectionConfig map[string]map[string]any `json:"connectionConfig,omitempty" yaml:"connectionConfig,omitempty"`
+	// plugins
+	PortablePlugins map[string]*plugin.IOPlugin `json:"portablePlugins,omitempty" yaml:"portablePlugins,omitempty"`
+	// others
+	Service map[string]*service.ServiceCreationRequest `json:"service,omitempty" yaml:"service,omitempty"`
+	Schema  map[string]*schema.Info                    `json:"schema,omitempty" yaml:"schema,omitempty"`
+	Uploads map[string]*fileContent                    `json:"uploads,omitempty" yaml:"uploads,omitempty"`
+	// rules related
+	Streams map[string]*DatasourceExport `json:"streams" yaml:"streams"`
+	Tables  map[string]*DatasourceExport `json:"tables" yaml:"tables"`
+	Rules   map[string]*def.Rule         `json:"rules" yaml:"rules"`
+}
+
+type DatasourceExport struct {
+	SQL string `json:"sql" yaml:"sql"`
+}
+
+func GenMetaConfiguration() (*MetaConfiguration, error) {
+	m := &MetaConfiguration{
+		Streams: make(map[string]*DatasourceExport),
+		Tables:  make(map[string]*DatasourceExport),
+		Rules:   map[string]*def.Rule{},
+	}
+	var err error
+	if err = addConfiguration(m); err != nil {
+		return nil, err
+	}
+	if err = addPlugins(m); err != nil {
+		return nil, err
+	}
+	if err = addService(m); err != nil {
+		return nil, err
+	}
+	if err = addSchema(m); err != nil {
+		return nil, err
+	}
+	if err = addUploads(m); err != nil {
+		return nil, err
+	}
+
+	rset := rulesetProcessor.ExportRuleSet()
+	for key, sql := range rset.Streams {
+		m.Streams[key] = &DatasourceExport{
+			SQL: sql,
+		}
+	}
+	for key, sql := range rset.Tables {
+		m.Tables[key] = &DatasourceExport{
+			SQL: sql,
+		}
+	}
+	for key, v := range rset.Rules {
+		jm := make(map[string]any)
+		if err := json.Unmarshal([]byte(v), &jm); err != nil {
+			return nil, err
+		}
+		d := &def.Rule{}
+		if err := cast.MapToStruct(jm, d); err != nil {
+			return nil, err
+		}
+		m.Rules[key] = d
+	}
+	return m, nil
+}
+
+func addConfiguration(m *MetaConfiguration) error {
+	var err error
+	m.SourceConfig, err = conf.GetCfgFromKVStorage("sources", "", "")
+	if err != nil {
+		return err
+	}
+	m.SinkConfig, err = conf.GetCfgFromKVStorage("sinks", "", "")
+	if err != nil {
+		return err
+	}
+	m.ConnectionConfig, err = conf.GetCfgFromKVStorage("connections", "", "")
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func addPlugins(m *MetaConfiguration) error {
+	if managers["portable"] != nil {
+		want := make(map[string]*plugin.IOPlugin)
+		pm := managers["portable"].Export()
+		failpoint.Inject("mockYamlExport", func() {
+			e := map[string]*plugin.IOPlugin{
+				"p1": {
+					Name: "p1",
+					File: "path",
+				},
+			}
+			b, _ := json.Marshal(e)
+			pm = map[string]string{
+				"p1": string(b),
+			}
+		})
+		for k, v := range pm {
+			p := &plugin.IOPlugin{}
+			if err := json.Unmarshal([]byte(v), p); err != nil {
+				return err
+			}
+			want[k] = p
+		}
+		m.PortablePlugins = want
+	}
+	return nil
+}
+
+func addSchema(m *MetaConfiguration) error {
+	if managers["schema"] != nil {
+		want := make(map[string]*schema.Info)
+		pm := managers["schema"].Export()
+		failpoint.Inject("mockYamlExport", func() {
+			e := map[string]*schema.Info{
+				"p1": {
+					Name:     "p1",
+					FilePath: "path",
+				},
+			}
+			b, _ := json.Marshal(e)
+			pm = map[string]string{
+				"p1": string(b),
+			}
+		})
+		for k, v := range pm {
+			s := &schema.Info{}
+			if err := json.Unmarshal([]byte(v), s); err != nil {
+				return err
+			}
+			want[k] = s
+		}
+		m.Schema = want
+	}
+	return nil
+}
+
+func addUploads(m *MetaConfiguration) error {
+	ue := uploadsExport()
+	failpoint.Inject("mockYamlExport", func() {
+		e := map[string]*fileContent{
+			"p1": {
+				Name:     "p1",
+				FilePath: "path",
+			},
+		}
+		b, _ := json.Marshal(e)
+		ue = map[string]string{
+			"p1": string(b),
+		}
+	})
+	if len(ue) > 0 {
+		want := make(map[string]*fileContent)
+		for k, v := range ue {
+			f := &fileContent{}
+			if err := json.Unmarshal([]byte(v), f); err != nil {
+				return err
+			}
+			want[k] = f
+		}
+		m.Uploads = want
+	}
+	return nil
+}
+
+func addService(m *MetaConfiguration) error {
+	if managers["service"] != nil {
+		want := make(map[string]*service.ServiceCreationRequest)
+		pm := managers["service"].Export()
+		failpoint.Inject("mockYamlExport", func() {
+			e := map[string]*service.ServiceCreationRequest{
+				"p1": {
+					Name: "p1",
+					File: "path",
+				},
+			}
+			b, _ := json.Marshal(e)
+			pm = map[string]string{
+				"p1": string(b),
+			}
+		})
+		for k, v := range pm {
+			s := &service.ServiceCreationRequest{}
+			if err := json.Unmarshal([]byte(v), s); err != nil {
+				return err
+			}
+			want[k] = s
+		}
+		m.Service = want
+	}
+	return nil
+}
+
+func yamlConfigurationExportHandler(w http.ResponseWriter, r *http.Request) {
+	var yamlBytes []byte
+	var err error
+	var m *MetaConfiguration
+	const name = "ekuiper_export.yaml"
+	switch r.Method {
+	case http.MethodGet:
+		m, err = GenMetaConfiguration()
+		if err != nil {
+			handleError(w, err, "", logger)
+			return
+		}
+		yamlBytes, err = yaml.Marshal(m)
+		if err != nil {
+			handleError(w, err, "", logger)
+			return
+		}
+	}
+	w.Header().Set("Content-Type", "application/octet-stream")
+	w.Header().Add("Content-Disposition", "Attachment")
+	http.ServeContent(w, r, name, time.Now(), bytes.NewReader(yamlBytes))
+}
diff --git a/internal/service/manager.go b/internal/service/manager.go
index 296040a4..0a79cc5d 100644
--- a/internal/service/manager.go
+++ b/internal/service/manager.go
@@ -329,8 +329,8 @@ func (m *Manager) deleteFunc(service, name string) error {
 // ** CRUD of the service files **
 
 type ServiceCreationRequest struct {
-	Name string `json:"name"`
-	File string `json:"file"`
+	Name string `json:"name" yaml:"name"`
+	File string `json:"file" yaml:"file"`
 }
 
 func (s *ServiceCreationRequest) InstallScript() string {


Test patch:
diff --git a/.github/workflows/run_test_case.yaml b/.github/workflows/run_test_case.yaml
index b93c70bb..92447601 100644
--- a/.github/workflows/run_test_case.yaml
+++ b/.github/workflows/run_test_case.yaml
@@ -41,11 +41,11 @@ jobs:
         mkdir -p plugins/sinks plugins/sources plugins/functions
         sed -i -r "s/debug: .*/debug: true/1" etc/kuiper.yaml
         
-        go build --buildmode=plugin -trimpath --cover -covermode=atomic -coverpkg=./... -o plugins/sources/Random@v2.0.0.so extension/random/ext/export.go
-        cp extension/random/random.yaml etc/sources/random.yaml
-        go build --buildmode=plugin -trimpath --cover -covermode=atomic -coverpkg=./... -o plugins/functions/Echo.so extension/function/echo/echo.go
-        go build --buildmode=plugin -trimpath --cover -covermode=atomic -coverpkg=./... -o plugins/functions/CountPlusOne@v1.0.0.so extension/function/countPlusOne/countPlusOne.go
-        go build --buildmode=plugin -trimpath --cover -covermode=atomic -coverpkg=./... -o plugins/functions/AccumulateWordCount@v1.0.0.so extension/function/accumulateWordCount/accumulateWordCount.go
+        go build --buildmode=plugin -trimpath --cover -covermode=atomic -coverpkg=./... -o plugins/sources/Random@v2.0.0.so extensions/sources/random/random.go
+        cp extensions/sources/random/random.yaml etc/sources/random.yaml
+        go build --buildmode=plugin -trimpath --cover -covermode=atomic -coverpkg=./... -o plugins/functions/Echo.so extensions/functions/echo/echo.go
+        go build --buildmode=plugin -trimpath --cover -covermode=atomic -coverpkg=./... -o plugins/functions/CountPlusOne@v1.0.0.so extensions/functions/countPlusOne/countPlusOne.go
+        go build --buildmode=plugin -trimpath --cover -covermode=atomic -coverpkg=./... -o plugins/functions/AccumulateWordCount@v1.0.0.so extensions/functions/accumulateWordCount/accumulateWordCount.go
         
         mkdir -p plugins/portable/mirror
         cd sdk/go/example/mirror
diff --git a/extension/function/accumulateWordCount/accumulateWordCount.go b/extensions/functions/accumulateWordCount/accumulateWordCount.go
similarity index 100%
rename from extension/function/accumulateWordCount/accumulateWordCount.go
rename to extensions/functions/accumulateWordCount/accumulateWordCount.go
diff --git a/extension/function/accumulateWordCount/accumulateWordCount.json b/extensions/functions/accumulateWordCount/accumulateWordCount.json
similarity index 100%
rename from extension/function/accumulateWordCount/accumulateWordCount.json
rename to extensions/functions/accumulateWordCount/accumulateWordCount.json
diff --git a/extension/function/countPlusOne/countPlusOne.go b/extensions/functions/countPlusOne/countPlusOne.go
similarity index 100%
rename from extension/function/countPlusOne/countPlusOne.go
rename to extensions/functions/countPlusOne/countPlusOne.go
diff --git a/extension/function/countPlusOne/countPlusOne.json b/extensions/functions/countPlusOne/countPlusOne.json
similarity index 100%
rename from extension/function/countPlusOne/countPlusOne.json
rename to extensions/functions/countPlusOne/countPlusOne.json
diff --git a/extension/function/echo/echo.go b/extensions/functions/echo/echo.go
similarity index 100%
rename from extension/function/echo/echo.go
rename to extensions/functions/echo/echo.go
diff --git a/extension/function/echo/echo.json b/extensions/functions/echo/echo.json
similarity index 100%
rename from extension/function/echo/echo.json
rename to extensions/functions/echo/echo.json
diff --git a/extension/random/random.go b/extensions/impl/random/random.go
similarity index 100%
rename from extension/random/random.go
rename to extensions/impl/random/random.go
diff --git a/extension/sql/client/client.go b/extensions/impl/sql/client/client.go
similarity index 100%
rename from extension/sql/client/client.go
rename to extensions/impl/sql/client/client.go
diff --git a/extension/sql/client/client_test.go b/extensions/impl/sql/client/client_test.go
similarity index 95%
rename from extension/sql/client/client_test.go
rename to extensions/impl/sql/client/client_test.go
index 1a0cd272..16da8f04 100644
--- a/extension/sql/client/client_test.go
+++ b/extensions/impl/sql/client/client_test.go
@@ -20,7 +20,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 
-	"github.com/lf-edge/ekuiper/v2/extension/sql/testx"
+	"github.com/lf-edge/ekuiper/v2/extensions/impl/sql/testx"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
diff --git a/extension/sql/sink_test.go b/extensions/impl/sql/sink_test.go
similarity index 99%
rename from extension/sql/sink_test.go
rename to extensions/impl/sql/sink_test.go
index 9c23e2d2..85b21a69 100644
--- a/extension/sql/sink_test.go
+++ b/extensions/impl/sql/sink_test.go
@@ -23,7 +23,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 
-	"github.com/lf-edge/ekuiper/v2/extension/sql/testx"
+	"github.com/lf-edge/ekuiper/v2/extensions/impl/sql/testx"
 	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
diff --git a/extension/sql/source_test.go b/extensions/impl/sql/source_test.go
similarity index 98%
rename from extension/sql/source_test.go
rename to extensions/impl/sql/source_test.go
index c6d55da1..caf2df3c 100644
--- a/extension/sql/source_test.go
+++ b/extensions/impl/sql/source_test.go
@@ -24,8 +24,8 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/extension/sql/client"
-	"github.com/lf-edge/ekuiper/v2/extension/sql/testx"
+	"github.com/lf-edge/ekuiper/v2/extensions/impl/sql/client"
+	"github.com/lf-edge/ekuiper/v2/extensions/impl/sql/testx"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
 	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
diff --git a/internal/io/http/httpserver/websocketConn_test.go b/internal/io/http/httpserver/websocketConn_test.go
index d0eebf42..3daff97e 100644
--- a/internal/io/http/httpserver/websocketConn_test.go
+++ b/internal/io/http/httpserver/websocketConn_test.go
@@ -15,8 +15,14 @@
 package httpserver
 
 import (
+	"context"
+	"log"
+	"net/http"
+	"net/http/httptest"
+	"sync"
 	"testing"
 
+	"github.com/gorilla/websocket"
 	"github.com/stretchr/testify/require"
 
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
@@ -35,5 +41,95 @@ func TestWebsocketConn(t *testing.T) {
 	require.NoError(t, err)
 	require.NoError(t, conn.Ping(ctx))
 	conn.DetachSub(ctx, props)
+	require.NoError(t, conn.Close(ctx))
+}
+
+func TestWebsocketClientConn(t *testing.T) {
+	tc := newTC()
+	s := createWServer(tc)
+	defer func() {
+		s.Close()
+	}()
+	ctx := mockContext.NewMockContext("1", "2")
+	props := map[string]any{
+		"datasource": "/ws",
+		"addr":       s.URL[len("http://"):],
+	}
+	conn, err := createWebsocketServerConnection(ctx, props)
+	require.NoError(t, err)
 	require.NoError(t, conn.Ping(ctx))
+	conn.DetachSub(ctx, props)
+	require.NoError(t, conn.Close(ctx))
+}
+
+func newTC() *testcase {
+	ctx, cancel := context.WithCancel(context.Background())
+	return &testcase{
+		ctx:    ctx,
+		cancel: cancel,
+		recvCh: make(chan []byte, 10),
+		sendCh: make(chan []byte, 10),
+	}
+}
+
+type testcase struct {
+	ctx    context.Context
+	cancel context.CancelFunc
+	recvCh chan []byte
+	sendCh chan []byte
+}
+
+func createWServer(tc *testcase) *httptest.Server {
+	router := http.NewServeMux()
+	router.HandleFunc("/ws", tc.handler)
+	server := httptest.NewServer(router)
+	return server
+}
+
+var upgrader = websocket.Upgrader{
+	ReadBufferSize:  256,
+	WriteBufferSize: 256,
+	WriteBufferPool: &sync.Pool{},
+}
+
+func (tc *testcase) recvProcess(c *websocket.Conn) {
+	defer func() {
+		tc.cancel()
+		c.Close()
+	}()
+	for {
+		_, message, err := c.ReadMessage()
+		if err != nil {
+			return
+		}
+		tc.recvCh <- message
+	}
+}
+
+func (tc *testcase) sendProcess(c *websocket.Conn) {
+	defer func() {
+		tc.cancel()
+		c.Close()
+	}()
+	for {
+		select {
+		case <-tc.ctx.Done():
+			return
+		case x := <-tc.sendCh:
+			err := c.WriteMessage(websocket.TextMessage, x)
+			if err != nil {
+				return
+			}
+		}
+	}
+}
+
+func (tc *testcase) handler(w http.ResponseWriter, r *http.Request) {
+	c, err := upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		log.Print("upgrade:", err)
+		return
+	}
+	go tc.recvProcess(c)
+	go tc.sendProcess(c)
 }
diff --git a/internal/io/http/httpserver/websocket_client_test.go b/internal/io/http/httpserver/websocket_client_test.go
new file mode 100644
index 00000000..3be51159
--- /dev/null
+++ b/internal/io/http/httpserver/websocket_client_test.go
@@ -0,0 +1,53 @@
+// 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 httpserver
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/io/memory/pubsub"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+)
+
+func TestWebsocketClient(t *testing.T) {
+	tc := newTC()
+	s := createWServer(tc)
+	defer func() {
+		s.Close()
+	}()
+	ctx := mockContext.NewMockContext("1", "2")
+	wc := NewWebsocketClient(s.URL[len("http://"):], "/ws", nil)
+	require.NoError(t, wc.Connect())
+	rt, st := wc.Run(ctx)
+	pubsub.CreatePub(st)
+	defer func() {
+		pubsub.RemovePub(st)
+	}()
+	// wait process start
+	time.Sleep(100 * time.Millisecond)
+	data := []byte("123")
+	pubsub.ProduceAny(ctx, st, data)
+	require.Equal(t, data, <-tc.recvCh)
+	ch := pubsub.CreateSub(rt, nil, "", 1024)
+	defer func() {
+		pubsub.CloseSourceConsumerChannel(rt, "")
+	}()
+	tc.sendCh <- data
+	require.Equal(t, data, <-ch)
+	require.NoError(t, wc.Close(ctx))
+}
diff --git a/internal/io/http/httpserver/websocket_server_test.go b/internal/io/http/httpserver/websocket_server_test.go
index 02f47afd..02c9f97a 100644
--- a/internal/io/http/httpserver/websocket_server_test.go
+++ b/internal/io/http/httpserver/websocket_server_test.go
@@ -84,7 +84,7 @@ func TestWebsocketServerRecvDataOther(t *testing.T) {
 
 func TestWebsocketServerSendData(t *testing.T) {
 	endpoint := "/e1"
-	topic := sendTopic(endpoint)
+	topic := sendTopic(endpoint, true)
 	pubsub.CreatePub(topic)
 	ip := "127.0.0.1"
 	port := 10085
diff --git a/internal/io/simulator/simulator_test.go b/internal/io/simulator/simulator_test.go
new file mode 100644
index 00000000..4ced0962
--- /dev/null
+++ b/internal/io/simulator/simulator_test.go
@@ -0,0 +1,72 @@
+// 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 simulator
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+)
+
+func TestSourcePull(t *testing.T) {
+	data := make([]map[string]any, 0)
+	data = append(data, map[string]any{
+		"a": 1,
+	})
+	props1 := map[string]any{
+		"data": data,
+		"loop": true,
+	}
+	s1 := &SimulatorSource{}
+	ctx := mockContext.NewMockContext("1", "2")
+	require.NoError(t, s1.Provision(ctx, props1))
+	require.NoError(t, s1.Connect(ctx))
+	recvData := make(chan any, 10)
+	s1.Pull(ctx, time.Now(), func(ctx api.StreamContext, data any, meta map[string]any, ts time.Time) {
+		recvData <- data
+	}, func(ctx api.StreamContext, err error) {})
+	expData := map[string]any{
+		"a": 1,
+	}
+	require.Equal(t, expData, <-recvData)
+	s1.Pull(ctx, time.Now(), func(ctx api.StreamContext, data any, meta map[string]any, ts time.Time) {
+		recvData <- data
+	}, func(ctx api.StreamContext, err error) {})
+	require.Equal(t, expData, <-recvData)
+	require.NoError(t, s1.Close(ctx))
+
+	props2 := map[string]any{
+		"data": data,
+		"loop": false,
+	}
+	s2 := &SimulatorSource{}
+	require.NoError(t, s2.Provision(ctx, props2))
+	require.NoError(t, s2.Connect(ctx))
+	s2.Pull(ctx, time.Now(), func(ctx api.StreamContext, data any, meta map[string]any, ts time.Time) {
+		recvData <- data
+	}, func(ctx api.StreamContext, err error) {})
+	require.Equal(t, expData, <-recvData)
+	s2.Pull(ctx, time.Now(), func(ctx api.StreamContext, data any, meta map[string]any, ts time.Time) {}, func(ctx api.StreamContext, err error) {
+		recvData <- err
+	})
+	expErr := fmt.Errorf("simulator source message running out")
+	require.Equal(t, expErr, <-recvData)
+	require.NoError(t, s2.Close(ctx))
+}
diff --git a/internal/server/rest_test.go b/internal/server/rest_test.go
index 298c881b..8e21e258 100644
--- a/internal/server/rest_test.go
+++ b/internal/server/rest_test.go
@@ -207,6 +207,7 @@ func (suite *RestTestSuite) Test_sourcesManageHandler() {
 }
 
 func (suite *RestTestSuite) Test_rulesManageHandler() {
+	connection.InitConnectionManager4Test()
 	// Start rules
 	if rules, err := ruleProcessor.GetAllRules(); err != nil {
 		logger.Infof("Start rules error: %s", err)
@@ -419,7 +420,7 @@ func (suite *RestTestSuite) Test_rulesManageHandler() {
 	suite.r.ServeHTTP(w1, req1)
 
 	returnVal, _ = io.ReadAll(w1.Result().Body)
-	expect = `{"triggered":true,"id":"rule321","sql":"select * from alert","actions":[{"nop":{}}],"options":{"debug":false,"logFilename":"","isEventTime":false,"lateTolerance":"1s","concurrency":1,"bufferLength":1024,"sendMetaToSink":false,"sendError":true,"qos":0,"checkpointInterval":"5m0s","restartStrategy":{"attempts":0,"delay":"1s","multiplier":2,"maxDelay":"30s","jitter":0.1},"cron":"","duration":"","cronDatetimeRange":null}}`
+	expect = `{"triggered":true,"id":"rule321","sql":"select * from alert","actions":[{"nop":{}}],"options":{"lateTolerance":"1s","concurrency":1,"bufferLength":1024,"sendError":true,"checkpointInterval":"5m0s","restartStrategy":{"attempts":0,"delay":"1s","multiplier":2,"maxDelay":"30s","jitter":0.1}}}`
 	assert.Equal(suite.T(), expect, string(returnVal))
 
 	// delete rule
diff --git a/internal/server/yaml_import_export_test.go b/internal/server/yaml_import_export_test.go
new file mode 100644
index 00000000..dfe4125f
--- /dev/null
+++ b/internal/server/yaml_import_export_test.go
@@ -0,0 +1,69 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"testing"
+
+	"github.com/pingcap/failpoint"
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
+)
+
+func TestMetaConfiguration(t *testing.T) {
+	conf.InitConf()
+	conf.IsTesting = true
+	connection.InitConnectionManager4Test()
+	for _, v := range components {
+		v.register()
+	}
+	InitConfManagers()
+	prepare(t)
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/server/mockYamlExport", "return(true)")
+	defer failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/server/mockYamlExport")
+	m, err := GenMetaConfiguration()
+	require.NoError(t, err)
+	require.True(t, len(m.SourceConfig) > 0)
+	require.True(t, len(m.SinkConfig) > 0)
+	require.True(t, len(m.ConnectionConfig) > 0)
+	require.True(t, len(m.PortablePlugins) > 0)
+	require.True(t, len(m.Service) > 0)
+	require.True(t, len(m.Schema) > 0)
+	require.True(t, len(m.Uploads) > 0)
+	require.True(t, len(m.Streams) > 0)
+	require.True(t, len(m.Rules) > 0)
+}
+
+func prepare(t *testing.T) {
+	deleteRule("metaConf")
+	ruleProcessor.ExecDrop("metaConf")
+	streamProcessor.ExecStreamSql(`drop stream metaConfTest`)
+	require.NoError(t, conf.WriteCfgIntoKVStorage("sources", "mqtt", "demo1", map[string]any{
+		"a": 1,
+	}))
+	require.NoError(t, conf.WriteCfgIntoKVStorage("sinks", "mqtt", "demo1", map[string]any{
+		"a": 1,
+	}))
+	require.NoError(t, conf.WriteCfgIntoKVStorage("connections", "mqtt", "demo1", map[string]any{
+		"a": 1,
+	}))
+	_, err := streamProcessor.ExecStreamSql(`create stream metaConfTest() WITH (DATASOURCE="/API/DATA",CONF_KEY="demo1")`)
+	require.NoError(t, err)
+	rulejson := `{"trigger":false,"id":"metaConf","sql":"select * from metaConfTest","actions":[{"log":{}}]}`
+	_, err = createRule("metaConf", rulejson)
+	require.NoError(t, err)
+}