refactor(io): reconstruct the packages for io connectors

Move file sink into builtin

Signed-off-by: Jiyong Huang <huangjy@emqx.io>

.github/workflows/build_packages.yaml

@@ -177,7 +177,6 @@ jobs:
       fail-fast: false
       matrix:
         plugin:
-          - sinks/file
           - sinks/image
           - sinks/influx
           - sinks/tdengine

.github/workflows/run_test_case.yaml

@@ -43,10 +43,10 @@ jobs:
         set -e -u -x
         mkdir -p data
         mkdir -p log
+        mkdir -p plugins/sinks
         sed -i -r "s/debug: .*/debug: true/1" etc/kuiper.yaml
         go build --buildmode=plugin -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 -o plugins/sinks/File@v1.0.0.so extensions/sinks/file/file.go
         go build --buildmode=plugin -o plugins/functions/Echo.so extensions/functions/echo/echo.go
         go build --buildmode=plugin -o plugins/functions/CountPlusOne@v1.0.0.so extensions/functions/countPlusOne/countPlusOne.go
         go build --buildmode=plugin -o plugins/functions/AccumulateWordCount@v1.0.0.so extensions/functions/accumulateWordCount/accumulateWordCount.go

Makefile

@@ -89,8 +89,7 @@ docker:
 	docker buildx build --no-cache --platform=linux/amd64 -t $(TARGET):$(VERSION)-slim -f deploy/docker/Dockerfile-slim . --load
 	docker buildx build --no-cache --platform=linux/amd64 -t $(TARGET):$(VERSION)-dev -f deploy/docker/Dockerfile-dev . --load
 
-PLUGINS := sinks/file \
-	sinks/influx \
+PLUGINS := sinks/influx \
 	sinks/influx2 \
 	sinks/zmq \
 	sinks/image \

docs/directory.json

@@ -124,14 +124,14 @@
 									"path": "guide/sources/builtin/http_push"
 								},
 								{
-									"title": "内存源",
-									"path": "guide/sources/builtin/memory"
-								},
-								{
 									"title": "文件源",
 									"path": "guide/sources/builtin/file"
 								},
 								{
+									"title": "内存源",
+									"path": "guide/sources/builtin/memory"
+								},
+								{
 									"title": "Redis 源",
 									"path": "guide/sources/builtin/redis"
 								}
@@ -141,20 +141,20 @@
 							"title": "预定义插件源",
 							"children": [
 								{
-									"title": "Zero MQ 源",
-									"path": "guide/sources/plugin/zmq"
-								},
-								{
 									"title": "SQL 源",
 									"path": "guide/sources/plugin/sql"
 								},
 								{
-									"title": "随机数据产生器源",
-									"path": "guide/sources/plugin/random"
-								},
-								{
 									"title": "视频源",
 									"path": "guide/sources/plugin/video"
+								},
+								{
+									"title": "Zero MQ 源",
+									"path": "guide/sources/plugin/zmq"
+								},
+								{
+									"title": "随机数据产生器源",
+									"path": "guide/sources/plugin/random"
 								}
 							]
 						}
@@ -191,6 +191,14 @@
 									"path": "guide/sinks/builtin/rest"
 								},
 								{
+									"title": "Redis Sink",
+									"path": "guide/sinks/builtin/redis"
+								},
+								{
+									"title": "File Sink",
+									"path": "guide/sinks/builtin/file"
+								},
+								{
 									"title": "Memory Sink",
 									"path": "guide/sinks/builtin/memory"
 								},
@@ -201,10 +209,6 @@
 								{
 									"title": "Nop Sink",
 									"path": "guide/sinks/builtin/nop"
-								},
-								{
-									"title": "Redis Sink",
-									"path": "guide/sinks/builtin/redis"
 								}
 							]
 						},
@@ -212,14 +216,6 @@
 							"title": "预定义插件 Sink",
 							"children": [
 								{
-									"title": "ZeroMQ Sink",
-									"path": "guide/sinks/plugin/zmq"
-								},
-								{
-									"title": "File Sink",
-									"path": "guide/sinks/plugin/file"
-								},
-								{
 									"title": "SQL Sink",
 									"path": "guide/sinks/plugin/sql"
 								},
@@ -238,6 +234,10 @@
 								{
 									"title": "Image Sink",
 									"path": "guide/sinks/plugin/image"
+								},
+								{
+									"title": "ZeroMQ Sink",
+									"path": "guide/sinks/plugin/zmq"
 								}
 							]
 						}
@@ -798,14 +798,14 @@
 									"path": "guide/sources/builtin/http_push"
 								},
 								{
-									"title": "Memory Source",
-									"path": "guide/sources/builtin/memory"
-								},
-								{
 									"title": "File Source",
 									"path": "guide/sources/builtin/file"
 								},
 								{
+									"title": "Memory Source",
+									"path": "guide/sources/builtin/memory"
+								},
+								{
 									"title": "Redis Source",
 									"path": "guide/sources/builtin/redis"
 								}
@@ -815,20 +815,20 @@
 							"title": "Predefined Source Plugins",
 							"children": [
 								{
-									"title": "Zero MQ Source",
-									"path": "guide/sources/plugin/zmq"
-								},
-								{
 									"title": "SQL Source",
 									"path": "guide/sources/plugin/sql"
 								},
 								{
+									"title": "Video Source",
+									"path": "guide/sources/plugin/video"
+								},
+								{
 									"title": "Random Source",
 									"path": "guide/sources/plugin/random"
 								},
 								{
-									"title": "Video Source",
-									"path": "guide/sources/plugin/video"
+									"title": "Zero MQ Source",
+									"path": "guide/sources/plugin/zmq"
 								}
 							]
 						}
@@ -865,6 +865,14 @@
 									"path": "guide/sinks/builtin/rest"
 								},
 								{
+									"title": "Redis Sink",
+									"path": "guide/sinks/builtin/redis"
+								},
+								{
+									"title": "File Sink",
+									"path": "guide/sinks/builtin/file"
+								},
+								{
 									"title": "Memory Sink",
 									"path": "guide/sinks/builtin/memory"
 								},
@@ -875,10 +883,6 @@
 								{
 									"title": "Nop Sink",
 									"path": "guide/sinks/builtin/nop"
-								},
-								{
-									"title": "Redis Sink",
-									"path": "guide/sinks/builtin/redis"
 								}
 							]
 						},
@@ -886,14 +890,6 @@
 							"title": "Predefined Sink Plugins",
 							"children": [
 								{
-									"title": "ZeroMQ Sink",
-									"path": "guide/sinks/plugin/zmq"
-								},
-								{
-									"title": "File Sink",
-									"path": "guide/sinks/plugin/file"
-								},
-								{
 									"title": "SQL Sink",
 									"path": "guide/sinks/plugin/sql"
 								},
@@ -912,6 +908,10 @@
 								{
 									"title": "Image Sink",
 									"path": "guide/sinks/plugin/image"
+								},
+								{
+									"title": "ZeroMQ Sink",
+									"path": "guide/sinks/plugin/zmq"
 								}
 							]
 						}

docs/en_US/api/cli/data.md

@@ -19,7 +19,6 @@ The file format for importing and exporting Data is JSON, which can contain : `s
     "rule2": "{\"id\": \"rule2\",\"sql\": \"SELECT * FROM demo\",\"actions\": [{  \"log\": {}}]}"
   },
   "nativePlugins":{
-    "sinks_file":"fail to download file file:///root/ekuiper-jran/_plugins/ubuntu/sinks/file_amd64.zip: stat /root/ekuiper-jran/_plugins/ubuntu/sinks/file_amd64.zip: no such file or directory",
     "sinks_tdengine":"fail to download file file:///root/ekuiper-jran/_plugins/ubuntu/sinks/tdengine_amd64.zip: stat /root/ekuiper-jran/_plugins/ubuntu/sinks/tdengine_amd64.zip: no such file or directory",
     "sources_random":"fail to download file file:///root/ekuiper-jran/_plugins/ubuntu/sources/random_amd64.zip: stat /root/ekuiper-jran/_plugins/ubuntu/sources/random_amd64.zip: no such file or directory"
   },

docs/en_US/api/restapi/data.md

@@ -19,7 +19,6 @@ The file format for importing and exporting data is JSON, which can contain : `s
         "rule2": "{\"id\": \"rule2\",\"sql\": \"SELECT * FROM demo\",\"actions\": [{  \"log\": {}}]}"
     },
     "nativePlugins":{
-        "sinks_file":"fail to download file file:///root/ekuiper-jran/_plugins/ubuntu/sinks/file_amd64.zip: stat /root/ekuiper-jran/_plugins/ubuntu/sinks/file_amd64.zip: no such file or directory",
         "sinks_tdengine":"fail to download file file:///root/ekuiper-jran/_plugins/ubuntu/sinks/tdengine_amd64.zip: stat /root/ekuiper-jran/_plugins/ubuntu/sinks/tdengine_amd64.zip: no such file or directory",
         "sources_random":"fail to download file file:///root/ekuiper-jran/_plugins/ubuntu/sources/random_amd64.zip: stat /root/ekuiper-jran/_plugins/ubuntu/sources/random_amd64.zip: no such file or directory"
     },
@@ -117,8 +116,7 @@ Content-Type: application/json
   "streams":{},
   "tables":{},
   "rules":{},
-  "nativePlugins":{
-    "sinks_file":"fail to download file file:///root/ekuiper-jran/_plugins/ubuntu/sinks/file_amd64.zip: stat /root/ekuiper-jran/_plugins/ubuntu/sinks/file_amd64.zip: no such file or directory",
+  "nativePlugins":{    
     "sinks_tdengine":"fail to download file file:///root/ekuiper-jran/_plugins/ubuntu/sinks/tdengine_amd64.zip: stat /root/ekuiper-jran/_plugins/ubuntu/sinks/tdengine_amd64.zip: no such file or directory",
     "sources_random":"fail to download file file:///root/ekuiper-jran/_plugins/ubuntu/sources/random_amd64.zip: stat /root/ekuiper-jran/_plugins/ubuntu/sources/random_amd64.zip: no such file or directory"},
   "portablePlugins":{},

docs/en_US/extension/native/develop/overview.md

@@ -159,7 +159,6 @@ The following is a sample of metadata file.
 
 | Name                                                | Description                                                      | Remarks                                                   |
 |-----------------------------------------------------|------------------------------------------------------------------|-----------------------------------------------------------|
-| [file](../../../guide/sinks/plugin/file.md)         | The plugin saves the analysis results to a specified file system | Sample of plugin, not available in production environment |
 | [zmq](../../../guide/sinks/plugin/zmq.md)           | The plugin sends the analysis results to the topic of Zero Mq    | Sample of plugin, not available in production environment |
 | [Influxdb](../../../guide/sinks/plugin/influx.md)   | The plugin sends the analysis results to InfluxDB                | Provided by [@smart33690](https://github.com/smart33690)  |
 | [TDengine](../../../guide/sinks/plugin/tdengine.md) | The plugin sends the analysis results to TDengine                |                                                           |

docs/en_US/guide/sinks/plugin/file.md

@@ -2,16 +2,6 @@
 
 The sink is used for saving analysis result into a specified file.
 
-## Compile & deploy plugin
-
-```shell
-# cd $eKuiper_src
-# go build -trimpath --buildmode=plugin -o plugins/sinks/File.so extensions/sinks/file/file.go
-# cp plugins/sinks/File.so $eKuiper_install/plugins/sinks
-```
-
-Restart the eKuiper server to activate the plugin.
-
 ## Properties
 
 | Property name | Optional | Description                                                                                                                                     |

docs/en_US/guide/sinks/overview.md

@@ -10,10 +10,11 @@ Users can directly use the built-in sinks in the standard eKuiper instance. The
 - [Neuron sink](./builtin/neuron.md): sink to the local neuron instance.
 - [EdgeX sink](./builtin/edgex.md): sink to EdgeX Foundry. This sink only exist when enabling edgex build tag.
 - [Rest sink](./builtin/rest.md): sink to external http server.
+- [Redis sink](./builtin/redis.md): sink to redis.
+- [File sink](./builtin/file.md): sink to a file.
 - [Memory sink](./builtin/memory.md): sink to eKuiper memory topic to form rule pipelines.
 - [Log sink](./builtin/log.md): sink to log, usually for debug only.
 - [Nop sink](./builtin/nop.md): sink to nowhere. It is used for performance testing now.
-- [Redis sink](./builtin/redis.md): sink to redis.
 
 ## Predefined Sink Plugins
 
@@ -23,12 +24,11 @@ Additionally, these plugins have pre-built binaries for the mainstream cpu archi
 
 The list of predefined sink plugins:
 
-- [Zero MQ sink](./plugin/zmq.md): sink to zero mq.
-- [File sink](./plugin/file.md): sink to a file.
 - [InfluxDB sink](./plugin/influx.md): sink to influx db `v1.x`. 
 - [InfluxDBV2 sink](./plugin/influx2.md): sink to influx db `v2.x`.
 - [Tdengine sink](./plugin/tdengine.md): sink to tdengine.
 - [Image sink](./plugin/image.md): sink to an image file. Only used to handle binary result.
+- [Zero MQ sink](./plugin/zmq.md): sink to zero mq.
 
 ## Updatable Sink
 

docs/en_US/guide/sources/overview.md

@@ -20,9 +20,10 @@ Users can directly use the built-in sources in the standard eKuiper instance. Th
 - [EdgeX source](./builtin/edgex.md): read data from EdgeX foundry.
 - [Http pull source](./builtin/http_pull.md): source to pull data from http servers.
 - [Http push source](./builtin/http_push.md): push data to eKuiper through http.
-- [Memory source](./builtin/memory.md): source to read from eKuiper memory topic to form rule pipelines.
-- [File source](./builtin/file.md): source to read from file, usually used as tables.
 - [Redis source](./builtin/redis.md): source to lookup from redis as a lookup table.
+- [File source](./builtin/file.md): source to read from file, usually used as tables.
+- [Memory source](./builtin/memory.md): source to read from eKuiper memory topic to form rule pipelines.
+
 
 ## Predefined Source Plugins
 
@@ -32,9 +33,9 @@ Additionally, these plugins have pre-built binaries for the mainstream cpu archi
 
 The list of predefined source plugins:
 
-- [Zero MQ source](./plugin/zmq.md): read data from zero mq.
-- [Random source](./plugin/random.md): a source to generate random data for testing.
 - [SQL source](./plugin/sql.md): a source to periodically fetch data from SQL DB.
+- [Random source](./plugin/random.md): a source to generate random data for testing.
+- [Zero MQ source](./plugin/zmq.md): read data from zero mq.
 
 ## Use of sources
 

docs/en_US/operation/manager-ui/plugins_in_manager.md

@@ -116,7 +116,7 @@ In `etc/kuiper.yaml`, there is a `pluginHosts` configuration item, and users can
 https://$host/$folder/$version/$os/$type/$plugin.zip
 ```
 
-Example:  https://127.0.0.1:9090/kuiper-plugins/1.7.1/debian/sinks/file.zip
+Example:  https://127.0.0.1:9090/kuiper-plugins/1.7.1/debian/sinks/sql.zip
 
 Among them:
 

docs/zh_CN/api/cli/data.md

@@ -19,7 +19,6 @@ eKuiper 命令行工具允许您导入导出当前的所有数据。
     "rule2": "{\"id\": \"rule2\",\"sql\": \"SELECT * FROM demo\",\"actions\": [{  \"log\": {}}]}"
   },
   "nativePlugins":{
-    "sinks_file":"fail to download file file:///root/ekuiper-jran/_plugins/ubuntu/sinks/file_amd64.zip: stat /root/ekuiper-jran/_plugins/ubuntu/sinks/file_amd64.zip: no such file or directory",
     "sinks_tdengine":"fail to download file file:///root/ekuiper-jran/_plugins/ubuntu/sinks/tdengine_amd64.zip: stat /root/ekuiper-jran/_plugins/ubuntu/sinks/tdengine_amd64.zip: no such file or directory",
     "sources_random":"fail to download file file:///root/ekuiper-jran/_plugins/ubuntu/sources/random_amd64.zip: stat /root/ekuiper-jran/_plugins/ubuntu/sources/random_amd64.zip: no such file or directory"
   },

docs/zh_CN/api/restapi/data.md

@@ -19,7 +19,6 @@ eKuiper REST api 允许您导入导出当前的所有数据。
         "rule2": "{\"id\": \"rule2\",\"sql\": \"SELECT * FROM demo\",\"actions\": [{  \"log\": {}}]}"
     },
     "nativePlugins":{
-        "sinks_file":"fail to download file file:///root/ekuiper-jran/_plugins/ubuntu/sinks/file_amd64.zip: stat /root/ekuiper-jran/_plugins/ubuntu/sinks/file_amd64.zip: no such file or directory",
         "sinks_tdengine":"fail to download file file:///root/ekuiper-jran/_plugins/ubuntu/sinks/tdengine_amd64.zip: stat /root/ekuiper-jran/_plugins/ubuntu/sinks/tdengine_amd64.zip: no such file or directory",
         "sources_random":"fail to download file file:///root/ekuiper-jran/_plugins/ubuntu/sources/random_amd64.zip: stat /root/ekuiper-jran/_plugins/ubuntu/sources/random_amd64.zip: no such file or directory"
     },
@@ -116,7 +115,6 @@ Content-Type: application/json
   "tables":{},
   "rules":{},
   "nativePlugins":{
-    "sinks_file":"fail to download file file:///root/ekuiper-jran/_plugins/ubuntu/sinks/file_amd64.zip: stat /root/ekuiper-jran/_plugins/ubuntu/sinks/file_amd64.zip: no such file or directory",
     "sinks_tdengine":"fail to download file file:///root/ekuiper-jran/_plugins/ubuntu/sinks/tdengine_amd64.zip: stat /root/ekuiper-jran/_plugins/ubuntu/sinks/tdengine_amd64.zip: no such file or directory",
     "sources_random":"fail to download file file:///root/ekuiper-jran/_plugins/ubuntu/sources/random_amd64.zip: stat /root/ekuiper-jran/_plugins/ubuntu/sources/random_amd64.zip: no such file or directory"},
   "portablePlugins":{},

docs/zh_CN/extension/native/develop/overview.md

@@ -159,7 +159,6 @@ source 的大部分属性用户通过对应的配置文件指定，用户无法
 
 | 名称                                                  | 描述                       | 备注                                                |
 |-----------------------------------------------------|--------------------------|---------------------------------------------------|
-| [file](../../../guide/sinks/plugin/file.md)         | 该插件将分析结果保存到某个指定到文件系统中    | 插件样例，不可用于生产环境                                     |
 | [zmq](../../../guide/sinks/plugin/zmq.md)           | 该插件将分析结果发送到 Zero Mq 的主题中 | 插件样例，不能用于生产环境                                     |
 | [Influxdb](../../../guide/sinks/plugin/influx.md)   | 该插件将分析结果发送到 InfluxDB 中   | 由 [@smart33690](https://github.com/smart33690) 提供 |
 | [TDengine](../../../guide/sinks/plugin/tdengine.md) | 该插件将分析结果发送到 TDengine 中   |                                                   |

docs/zh_CN/guide/sinks/plugin/file.md

@@ -2,16 +2,6 @@
 
 目标（Sink）用于将分析结果保存到指定文件中。
 
-## 编译和部署插件
-
-```shell
-# cd $ekuiper_src
-# go build -trimpath --buildmode=plugin -o plugins/sinks/File.so extensions/sinks/file/file.go
-# cp plugins/sinks/File.so $ekuiper_install/plugins/sinks
-```
-
-重新启动 eKuiper 服务器以激活插件。
-
 ## 属性
 
 | 属性名称     | 是否可选 | 说明                                          |

docs/zh_CN/guide/sinks/overview.md

@@ -10,10 +10,11 @@
 - [Neuron sink](./builtin/neuron.md)：输出到本地的 Neuron 实例。
 - [EdgeX sink](./builtin/edgex.md)：输出到 EdgeX Foundry。此动作仅在启用 edgex 编译标签时存在。
 - [Rest sink](./builtin/rest.md)：输出到外部 http 服务器。
+- [Redis sink](./builtin/redis.md): 写入 Redis 。
+- [File sink](./builtin/file.md)： 写入文件。
 - [Memory sink](./builtin/memory.md)：输出到 eKuiper 内存主题以形成规则管道。
 - [Log sink](./builtin/log.md)：写入日志，通常只用于调试。
 - [Nop sink](./builtin/nop.md)：不输出，用于性能测试。
-- [Redis sink](./builtin/redis.md): 写入 Redis 。
 
 ## 预定义的动作插件
 
@@ -23,12 +24,11 @@
 
 预定义的动作插件列表。
 
-- [Zero MQ sink](./plugin/zmq.md)：输出到 Zero MQ 。
-- [File sink](./plugin/file.md)： 写入文件。
 - [InfluxDB sink](./plugin/influx.md)： 写入 Influx DB `v1.x`。
 - [InfluxDBV2 sink](./plugin/influx2.md)： 写入 Influx DB `v2.x`。
 - [Tdengine sink](./plugin/tdengine.md)： 写入 Tdengine 。
 - [Image sink](./plugin/image.md)： 写入一个图像文件。仅用于处理二进制结果。
+- [Zero MQ sink](./plugin/zmq.md)：输出到 Zero MQ 。
 
 ## 更新
 

docs/zh_CN/guide/sources/overview.md

@@ -20,9 +20,9 @@
 - [EdgeX source](./builtin/edgex.md): 从 EdgeX foundry 读取数据。
 - [Http pull source](./builtin/http_pull.md)：从 http 服务器中拉取数据。
 - [Http push source](./builtin/http_push.md)：通过 http 推送数据到 eKuiper。
-- [Memory source](./builtin/memory.md)：从 eKuiper 内存主题读取数据以形成规则管道。
-- [File source](./builtin/file.md)：从文件中读取数据，通常用作表格。
 - [Redis source](./builtin/redis.md): 从 Redis 中查询数据，用作查询表。
+- [File source](./builtin/file.md)：从文件中读取数据，通常用作表格。
+- [Memory source](./builtin/memory.md)：从 eKuiper 内存主题读取数据以形成规则管道。
 
 ## 预定义的源插件
 
@@ -32,9 +32,9 @@
 
 预定义的源插件列表：
 
-- [Zero MQ source](./plugin/zmq.md)：从Zero MQ读取数据。
-- [Random source](./plugin/random.md): 一个生成随机数据的源，用于测试。
 - [SQL source](./plugin/sql.md): 定期从关系数据库中拉取数据。
+- [Random source](./plugin/random.md): 一个生成随机数据的源，用于测试。
+- [Zero MQ source](./plugin/zmq.md)：从Zero MQ读取数据。
 
 ## 源的使用
 

docs/zh_CN/operation/manager-ui/plugins_in_manager.md

@@ -116,7 +116,7 @@ Function 元数据详细配置请参考 [sink 元数据配置](../../extension/n
 https://$host/$folder/$version/$os/$type/$plugin.zip
 ```
 
-例如： https://127.0.0.1:9090/kuiper-plugins/1.7.1/debian/sinks/file.zip
+例如： https://127.0.0.1:9090/kuiper-plugins/1.7.1/debian/sinks/sql.zip
 
 其中：
 

internal/binder/io/builtin.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -15,10 +15,12 @@
 package io
 
 import (
-	"github.com/lf-edge/ekuiper/internal/topo/memory"
-	"github.com/lf-edge/ekuiper/internal/topo/neuron"
-	"github.com/lf-edge/ekuiper/internal/topo/sink"
-	"github.com/lf-edge/ekuiper/internal/topo/source"
+	"github.com/lf-edge/ekuiper/internal/io/file"
+	"github.com/lf-edge/ekuiper/internal/io/http"
+	"github.com/lf-edge/ekuiper/internal/io/memory"
+	"github.com/lf-edge/ekuiper/internal/io/mqtt"
+	"github.com/lf-edge/ekuiper/internal/io/neuron"
+	"github.com/lf-edge/ekuiper/internal/io/sink"
 	"github.com/lf-edge/ekuiper/pkg/api"
 )
 
@@ -28,21 +30,22 @@ type NewSinkFunc func() api.Sink
 
 var (
 	sources = map[string]NewSourceFunc{
-		"mqtt":     func() api.Source { return &source.MQTTSource{} },
-		"httppull": func() api.Source { return &source.HTTPPullSource{} },
-		"httppush": func() api.Source { return &source.HTTPPushSource{} },
-		"file":     func() api.Source { return &source.FileSource{} },
+		"mqtt":     func() api.Source { return &mqtt.MQTTSource{} },
+		"httppull": func() api.Source { return &http.HTTPPullSource{} },
+		"httppush": func() api.Source { return &http.HTTPPushSource{} },
+		"file":     func() api.Source { return &file.FileSource{} },
 		"memory":   func() api.Source { return memory.GetSource() },
 		"neuron":   func() api.Source { return neuron.GetSource() },
 	}
 	sinks = map[string]NewSinkFunc{
 		"log":         sink.NewLogSink,
 		"logToMemory": sink.NewLogSinkToMemory,
-		"mqtt":        func() api.Sink { return &sink.MQTTSink{} },
-		"rest":        func() api.Sink { return &sink.RestSink{} },
+		"mqtt":        func() api.Sink { return &mqtt.MQTTSink{} },
+		"rest":        func() api.Sink { return &http.RestSink{} },
 		"nop":         func() api.Sink { return &sink.NopSink{} },
 		"memory":      func() api.Sink { return memory.GetSink() },
 		"neuron":      func() api.Sink { return neuron.GetSink() },
+		"file":        func() api.Sink { return file.File() },
 	}
 	lookupSources = map[string]NewLookupSourceFunc{
 		"memory": func() api.LookupSource { return memory.GetLookupSource() },

internal/binder/io/ext_edgex.go

@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -18,12 +18,11 @@
 package io
 
 import (
-	"github.com/lf-edge/ekuiper/internal/topo/sink"
-	"github.com/lf-edge/ekuiper/internal/topo/source"
+	"github.com/lf-edge/ekuiper/internal/io/edgex"
 	"github.com/lf-edge/ekuiper/pkg/api"
 )
 
 func init() {
-	sources["edgex"] = func() api.Source { return &source.EdgexSource{} }
-	sinks["edgex"] = func() api.Sink { return &sink.EdgexMsgBusSink{} }
+	sources["edgex"] = func() api.Source { return &edgex.EdgexSource{} }
+	sinks["edgex"] = func() api.Sink { return &edgex.EdgexMsgBusSink{} }
 }

internal/binder/io/ext_redis.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -17,7 +17,7 @@
 package io
 
 import (
-	"github.com/lf-edge/ekuiper/internal/topo/redis"
+	"github.com/lf-edge/ekuiper/internal/io/redis"
 	"github.com/lf-edge/ekuiper/pkg/api"
 )
 

internal/topo/sink/edgex_sink.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -15,7 +15,7 @@
 //go:build edgex
 // +build edgex
 
-package sink
+package edgex
 
 import (
 	"encoding/base64"
@@ -32,14 +32,7 @@ import (
 	"reflect"
 )
 
-type messageType string
-
-const (
-	MessageTypeEvent   messageType = "event"
-	MessageTypeRequest messageType = "request"
-)
-
-type EdgexConf struct {
+type SinkConf struct {
 	Topic        string      `json:"topic"`
 	TopicPrefix  string      `json:"topicPrefix"`
 	MessageType  messageType `json:"messageType"`
@@ -52,7 +45,7 @@ type EdgexConf struct {
 }
 
 type EdgexMsgBusSink struct {
-	c *EdgexConf
+	c *SinkConf
 
 	config map[string]interface{}
 	topic  string
@@ -62,7 +55,7 @@ type EdgexMsgBusSink struct {
 
 func (ems *EdgexMsgBusSink) Configure(ps map[string]interface{}) error {
 
-	c := &EdgexConf{
+	c := &SinkConf{
 		MessageType: MessageTypeEvent,
 		ContentType: "application/json",
 		DeviceName:  "ekuiper",

internal/topo/sink/edgex_sink_test.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -15,7 +15,7 @@
 //go:build edgex
 // +build edgex
 
-package sink
+package edgex
 
 import (
 	"encoding/json"
@@ -72,14 +72,14 @@ func compareReading(expected, actual dtos.BaseReading) bool {
 func TestConfigure(t *testing.T) {
 	var tests = []struct {
 		conf     map[string]interface{}
-		expected *EdgexConf
+		expected *SinkConf
 		error    string
 	}{
 		{ // 0
 			conf: map[string]interface{}{
 				"metadata": "meta",
 			},
-			expected: &EdgexConf{
+			expected: &SinkConf{
 				MessageType: MessageTypeEvent,
 				ContentType: "application/json",
 				DeviceName:  "ekuiper",
@@ -99,7 +99,7 @@ func TestConfigure(t *testing.T) {
 				"sourceName":  "ekuiper",
 				"contentType": "application/json",
 			},
-			expected: &EdgexConf{
+			expected: &SinkConf{
 				MessageType: MessageTypeEvent,
 				ContentType: "application/json",
 				DeviceName:  "ekuiper",
@@ -121,7 +121,7 @@ func TestConfigure(t *testing.T) {
 					"ClientId": "edgex_message_bus_001",
 				},
 			},
-			expected: &EdgexConf{
+			expected: &SinkConf{
 				MessageType: MessageTypeEvent,
 				ContentType: "application/json",
 				DeviceName:  "ekuiper",
@@ -141,7 +141,7 @@ func TestConfigure(t *testing.T) {
 				"messageType": "request",
 				"contentType": "application/json",
 			},
-			expected: &EdgexConf{
+			expected: &SinkConf{
 				MessageType: MessageTypeRequest,
 				ContentType: "application/json",
 				DeviceName:  "ekuiper",

internal/topo/source/edgex_source.go

@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -15,7 +15,7 @@
 //go:build edgex
 // +build edgex
 
-package source
+package edgex
 
 import (
 	"encoding/json"
@@ -41,7 +41,7 @@ type EdgexSource struct {
 	buflen      int
 }
 
-type EdgexConf struct {
+type SourceConf struct {
 	Topic       string      `json:"topic"`
 	MessageType messageType `json:"messageType"`
 	BufferLen   int         `json:"bufferLength"`
@@ -55,7 +55,7 @@ const (
 )
 
 func (es *EdgexSource) Configure(_ string, props map[string]interface{}) error {
-	c := &EdgexConf{
+	c := &SourceConf{
 		MessageType: MessageTypeEvent,
 	}
 	err := cast.MapToStruct(props, c)

internal/topo/source/edgex_source_test.go

@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -15,7 +15,7 @@
 //go:build edgex
 // +build edgex
 
-package source
+package edgex
 
 import (
 	"encoding/json"

extensions/sinks/file/file.go

@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package main
+package file
 
 import (
 	"bufio"

internal/topo/source/file_source.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package source
+package file
 
 import (
 	"bufio"

internal/topo/source/file_source_test.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.
@@ -12,11 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package source
+package file
 
 import (
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/topo/mock"
+	"github.com/lf-edge/ekuiper/internal/io/mock"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"io"
 	"os"

internal/topo/source/httppull_source.go

@@ -12,15 +12,15 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package source
+package http
 
 import (
 	"crypto/md5"
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"github.com/lf-edge/ekuiper/internal/io/mock"
 	"github.com/lf-edge/ekuiper/internal/pkg/cert"
-	"github.com/lf-edge/ekuiper/internal/topo/mock"
 	"github.com/lf-edge/ekuiper/pkg/infra"
 	"io"
 	"net/http"
@@ -39,9 +39,6 @@ const DefaultTimeout = 5000
 type HTTPPullConf struct {
 	Url                string            `json:"url"`
 	Method             string            `json:"method"`
-	Interval           int               `json:"interval"`
-	Timeout            int               `json:"timeout"`
-	Incremental        bool              `json:"incremental"`
 	Body               string            `json:"body"`
 	BodyType           string            `json:"bodyType"`
 	Headers            map[string]string `json:"headers"`
@@ -49,9 +46,13 @@ type HTTPPullConf struct {
 	CertificationPath  string            `json:"certificationPath"`
 	PrivateKeyPath     string            `json:"privateKeyPath"`
 	RootCaPath         string            `json:"rootCaPath"`
+	Timeout            int               `json:"timeout"`
 	// Could be code or body
 	ResponseType string                            `json:"responseType"`
 	OAuth        map[string]map[string]interface{} `json:"oauth"`
+	// Pull specific properties
+	Interval    int  `json:"interval"`
+	Incremental bool `json:"incremental"`
 }
 
 type AccessTokenConf struct {

internal/topo/source/httppull_source_test.go

@@ -12,14 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package source
+package http
 
 import (
 	"encoding/json"
 	"errors"
 	"fmt"
 	"github.com/gorilla/mux"
-	"github.com/lf-edge/ekuiper/internal/topo/mock"
+	"github.com/lf-edge/ekuiper/internal/io/mock"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"net"
 	"net/http"

internal/topo/source/httppush_source.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -12,13 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package source
+package http
 
 import (
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
-	"github.com/lf-edge/ekuiper/internal/topo/source/httpserver"
+	"github.com/lf-edge/ekuiper/internal/io/http/httpserver"
+	"github.com/lf-edge/ekuiper/internal/io/memory/pubsub"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"github.com/lf-edge/ekuiper/pkg/infra"

internal/topo/source/httpserver/data_server.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.
@@ -20,8 +20,8 @@ import (
 	"github.com/gorilla/handlers"
 	"github.com/gorilla/mux"
 	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/io/memory/pubsub"
 	kctx "github.com/lf-edge/ekuiper/internal/topo/context"
-	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
 	"github.com/lf-edge/ekuiper/internal/topo/state"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"net/http"

internal/topo/source/httpserver/data_server_test.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.

internal/topo/sink/rest_sink.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package sink
+package http
 
 import (
 	"encoding/json"

internal/topo/sink/rest_sink_test.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package sink
+package http
 
 import (
 	"fmt"

internal/topo/memory/export.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.

internal/topo/memory/lookupsource.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.
@@ -16,7 +16,7 @@ package memory
 
 import (
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/topo/memory/store"
+	"github.com/lf-edge/ekuiper/internal/io/memory/store"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"regexp"
 	"strings"

internal/topo/memory/lookupsource_test.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.
@@ -17,8 +17,8 @@ package memory
 import (
 	gocontext "context"
 	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/io/memory/pubsub"
 	"github.com/lf-edge/ekuiper/internal/topo/context"
-	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"reflect"
 	"testing"

internal/topo/memory/memory_test.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.
@@ -18,8 +18,8 @@ import (
 	"fmt"
 	"github.com/gdexlab/go-render/render"
 	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/io/memory/pubsub"
 	"github.com/lf-edge/ekuiper/internal/topo/context"
-	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
 	"github.com/lf-edge/ekuiper/internal/topo/state"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"reflect"

internal/topo/memory/pubsub/manager.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.

internal/topo/memory/pubsub/manager_test.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.

internal/topo/memory/pubsub/tuple.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.

internal/topo/memory/sink.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -17,7 +17,7 @@ package memory
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
+	"github.com/lf-edge/ekuiper/internal/io/memory/pubsub"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/ast"
 	"github.com/lf-edge/ekuiper/pkg/cast"

internal/topo/memory/sink_test.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.
@@ -17,8 +17,8 @@ package memory
 import (
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/io/memory/pubsub"
 	"github.com/lf-edge/ekuiper/internal/topo/context"
-	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"reflect"
 	"testing"

internal/topo/memory/source.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -16,7 +16,7 @@ package memory
 
 import (
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
+	"github.com/lf-edge/ekuiper/internal/io/memory/pubsub"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"regexp"

internal/topo/memory/source_test.go

@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.

internal/topo/memory/store/db.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.

internal/topo/memory/store/db_test.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.

internal/topo/memory/store/store.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.
@@ -18,7 +18,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
+	"github.com/lf-edge/ekuiper/internal/io/memory/pubsub"
 	"github.com/lf-edge/ekuiper/pkg/ast"
 	"regexp"
 )

internal/topo/memory/store/store_test.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.

internal/topo/mock/mock.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.

internal/topo/mock/test_sink.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.

internal/topo/sink/mqtt_sink.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package sink
+package mqtt
 
 import (
 	"fmt"

internal/topo/source/mqtt_source.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package source
+package mqtt
 
 import (
 	"fmt"

internal/topo/source/mqtt_type.go

@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package source
+package mqtt
 
 import (
 	"fmt"

internal/topo/source/mqtt_type_test.go

@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package source
+package mqtt
 
 import (
 	"encoding/json"

internal/topo/neuron/connection.go

@@ -18,8 +18,8 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/io/memory/pubsub"
 	kctx "github.com/lf-edge/ekuiper/internal/topo/context"
-	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
 	"github.com/lf-edge/ekuiper/internal/topo/state"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/errorx"

internal/topo/neuron/multiple_test.go

@@ -15,7 +15,7 @@
 package neuron
 
 import (
-	"github.com/lf-edge/ekuiper/internal/topo/mock"
+	"github.com/lf-edge/ekuiper/internal/io/mock"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"reflect"
 	"sync"

internal/topo/neuron/neuron_test.go

@@ -16,7 +16,7 @@ package neuron
 
 import (
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/topo/mock"
+	"github.com/lf-edge/ekuiper/internal/io/mock"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"go.nanomsg.org/mangos/v3"
 	"go.nanomsg.org/mangos/v3/protocol/pair"

internal/topo/neuron/sink_test.go

@@ -15,7 +15,7 @@
 package neuron
 
 import (
-	"github.com/lf-edge/ekuiper/internal/topo/mock"
+	"github.com/lf-edge/ekuiper/internal/io/mock"
 	"reflect"
 	"testing"
 	"time"

internal/topo/neuron/source.go

@@ -16,7 +16,7 @@ package neuron
 
 import (
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
+	"github.com/lf-edge/ekuiper/internal/io/memory/pubsub"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"github.com/lf-edge/ekuiper/pkg/infra"

internal/topo/neuron/source_test.go

@@ -16,7 +16,7 @@ package neuron
 
 import (
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/topo/mock"
+	"github.com/lf-edge/ekuiper/internal/io/mock"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	_ "go.nanomsg.org/mangos/v3/transport/ipc"

internal/topo/redis/lookup_test.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.

internal/topo/redis/sink.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.

internal/topo/redis/sink_test.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.

internal/topo/sink/log_sink.go

@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.

internal/topo/sink/nop_sink.go

@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.

internal/plugin/native/manager_test.go

@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -162,7 +162,7 @@ func TestManager_List(t *testing.T) {
 			r: []string{"random", "random2", "random3"},
 		}, {
 			t: plugin.SINK,
-			r: []string{"file", "file2"},
+			r: []string{"file2"},
 		}, {
 			t: plugin.FUNCTION,
 			r: []string{"accumulateWordCount", "comp", "countPlusOne", "echo", "echo2"},

internal/plugin/portable/test/portable_rule_test.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -20,11 +20,11 @@ import (
 	"github.com/lf-edge/ekuiper/internal/binder"
 	"github.com/lf-edge/ekuiper/internal/binder/function"
 	"github.com/lf-edge/ekuiper/internal/binder/io"
+	"github.com/lf-edge/ekuiper/internal/io/memory/pubsub"
 	"github.com/lf-edge/ekuiper/internal/plugin/portable"
 	"github.com/lf-edge/ekuiper/internal/plugin/portable/runtime"
 	"github.com/lf-edge/ekuiper/internal/processor"
 	"github.com/lf-edge/ekuiper/internal/topo"
-	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
 	"github.com/lf-edge/ekuiper/internal/topo/planner"
 	"github.com/lf-edge/ekuiper/internal/topo/topotest"
 	"github.com/lf-edge/ekuiper/pkg/api"

internal/server/rpc.go

@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-2023 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.
@@ -23,9 +23,9 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/io/sink"
 	"github.com/lf-edge/ekuiper/internal/pkg/model"
 	"github.com/lf-edge/ekuiper/internal/topo/rule"
-	"github.com/lf-edge/ekuiper/internal/topo/sink"
 	"github.com/lf-edge/ekuiper/pkg/infra"
 	"io"
 	"net/http"