Repository: lf-edge/ekuiper
Pull request #2995 (2024-07-08 08:44:37)

refactor: extension impl structure

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2995?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 `33.33333%` with `6 lines` in your changes missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@4884ab8`). [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/2995?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 |
|---|---|---|
| [extensions/sinks/sql/sql.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2995?src=pr&el=tree&filepath=extensions%2Fsinks%2Fsql%2Fsql.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-ZXh0ZW5zaW9ucy9zaW5rcy9zcWwvc3FsLmdv) | 0.00% | [2 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2995?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [extensions/sources/sql/sql.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2995?src=pr&el=tree&filepath=extensions%2Fsources%2Fsql%2Fsql.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-ZXh0ZW5zaW9ucy9zb3VyY2VzL3NxbC9zcWwuZ28=) | 0.00% | [2 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2995?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [extensions/sources/video/video.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2995?src=pr&el=tree&filepath=extensions%2Fsources%2Fvideo%2Fvideo.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-ZXh0ZW5zaW9ucy9zb3VyY2VzL3ZpZGVvL3ZpZGVvLmdv) | 0.00% | [2 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2995?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    #2995   +/-   ##
============================================
  Coverage             ?   68.21%           
============================================
  Files                ?      342           
  Lines                ?    33001           
  Branches             ?        0           
============================================
  Hits                 ?    22510           
  Misses               ?     8479           
  Partials             ?     2012           
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2995?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/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{}
 }


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))
+}