refactor: extension impl structure (#2995)

Signed-off-by: Jiyong Huang <[email protected]>

.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/[email protected] 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/[email protected] extension/function/countPlusOne/countPlusOne.go
-        go build --buildmode=plugin -trimpath --cover -covermode=atomic -coverpkg=./... -o plugins/functions/[email protected] extension/function/accumulateWordCount/accumulateWordCount.go
+        go build --buildmode=plugin -trimpath --cover -covermode=atomic -coverpkg=./... -o plugins/sources/[email protected] 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/[email protected] extensions/functions/countPlusOne/countPlusOne.go
+        go build --buildmode=plugin -trimpath --cover -covermode=atomic -coverpkg=./... -o plugins/functions/[email protected] extensions/functions/accumulateWordCount/accumulateWordCount.go
         
         mkdir -p plugins/portable/mirror
         cd sdk/go/example/mirror

extension/sql/client/client_test.go → 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"
 )
 

extension/sql/sink.go → 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"

extension/sql/sink_test.go → 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"
 )

extension/sql/source.go → 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
 }

extension/sql/source_test.go → 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"

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

extensions/sinks/sql/sql.json

@@ -0,0 +1,76 @@
+{
+  "about": {
+    "trial": true,
+    "author": {
+      "name": "EMQ",
+      "email": "[email protected]",
+      "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:[email protected]: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:[email protected]: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"
+    }
+  }
+}

extension/random/ext/export.go → 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 {

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