feat(taos):add taos sink plugin (#392)

* feat(taos):add taos sink plugin

* feat(taos):add taos sink plugin

* feat(taos):add taos sink plugin

* feat(taos):add taos sink plugin

* feat(taos):add taos sink plugin

* build comment must appear before package clause and be followed by a blank line

* docs(taos):add table

Co-authored-by: mayuedong <mayuedong@emx.io>

docs/en_US/plugins/sinks/taos.md

@@ -0,0 +1,60 @@
+## 编译插件
+
+### plugins/go.mod
+
+```go
+module plugins
+
+go 1.13
+
+replace github.com/emqx/kuiper => /$kuiper
+
+require (
+    github.com/emqx/kuiper v0.0.0-00010101000000-000000000000 // indirect
+    github.com/taosdata/driver-go v0.0.0-20200723061832-5be6460b0c20
+)
+```
+
+```shell
+go mod edit -replace github.com/emqx/kuiper=/$kuiper
+go build --buildmode=plugin -o /$kuiper/plugins/sinks/Taos@v1.0.0.so /$kuiper/plugins/sinks/taos.go
+```
+
+## 规则 Actions 说明
+
+| 名称     | 类型     | 是否必填                      | 释义       |
+| -------- | -------- | ----------------------------- | ---------- |
+| ip       | string   | 必填                          | 数据库ip   |
+| port     | int      | 必填                          | 数据库端口 |
+| user     | string   | 必填                          | 用户名     |
+| password | string   | 必填                          | 密码       |
+| database | string   | 必填                          | 数据库名   |
+| table    | string   | 必填                          | 表名       |
+| fields   | []string | 选填（不填时用数据的key替代） | 表字段集合 |
+
+## 操作示例
+
+### 创建数据库、表，参考以下文档：
+
+```http
+https://www.taosdata.com/cn/getting-started/
+```
+
+### 创建流
+
+```curl
+curl --location --request POST 'http://127.0.0.1:9081/streams' --header 'Content-Type:application/json' --data '{"sql":"create stream demoStream(time string, age BIGINT) WITH ( DATASOURCE = \"device/+/message\", FORMAT = \"json\");"}'
+```
+
+### 创建规则
+
+```curl
+curl --location --request POST 'http://127.0.0.1:9081/rules' --header 'Content-Type:application/json' --data '{"id":"demoRule","sql":"SELECT * FROM demoStream;","actions":[{"taos":{"port":0,"ip":"127.0.0.1","user":"root","password":"taosdata","database":"dbName","table":"tableName","fields":["time","age"]}}]}'
+```
+
+### 发送数据
+
+```curl
+mosquitto_pub -h broker.emqx.io -m '{"time":"2020-01-11 18:18:18", "age" : 18}' -t device/device_001/message
+```
+

docs/zh_CN/plugins/sinks/taos.md

@@ -0,0 +1,60 @@
+## 编译插件
+
+### plugins/go.mod
+
+```go
+module plugins
+
+go 1.13
+
+replace github.com/emqx/kuiper => /$kuiper
+
+require (
+    github.com/emqx/kuiper v0.0.0-00010101000000-000000000000 // indirect
+    github.com/taosdata/driver-go v0.0.0-20200723061832-5be6460b0c20
+)
+```
+
+```shell
+go mod edit -replace github.com/emqx/kuiper=/$kuiper
+go build --buildmode=plugin -o /$kuiper/plugins/sinks/Taos@v1.0.0.so /$kuiper/plugins/sinks/taos.go
+```
+
+## 规则 Actions 说明
+
+| 名称     | 类型     | 是否必填                      | 释义       |
+| -------- | -------- | ----------------------------- | ---------- |
+| ip       | string   | 必填                          | 数据库ip   |
+| port     | int      | 必填                          | 数据库端口 |
+| user     | string   | 必填                          | 用户名     |
+| password | string   | 必填                          | 密码       |
+| database | string   | 必填                          | 数据库名   |
+| table    | string   | 必填                          | 表名       |
+| fields   | []string | 选填（不填时用数据的key替代） | 表字段集合 |
+
+## 操作示例
+
+### 创建数据库、表，参考以下文档：
+
+```http
+https://www.taosdata.com/cn/getting-started/
+```
+
+### 创建流
+
+```curl
+curl --location --request POST 'http://127.0.0.1:9081/streams' --header 'Content-Type:application/json' --data '{"sql":"create stream demoStream(time string, age BIGINT) WITH ( DATASOURCE = \"device/+/message\", FORMAT = \"json\");"}'
+```
+
+### 创建规则
+
+```curl
+curl --location --request POST 'http://127.0.0.1:9081/rules' --header 'Content-Type:application/json' --data '{"id":"demoRule","sql":"SELECT * FROM demoStream;","actions":[{"taos":{"port":0,"ip":"127.0.0.1","user":"root","password":"taosdata","database":"dbName","table":"tableName","fields":["time","age"]}}]}'
+```
+
+### 发送数据
+
+```curl
+mosquitto_pub -h broker.emqx.io -m '{"time":"2020-01-11 18:18:18", "age" : 18}' -t device/device_001/message
+```
+

plugins/sinks/taos.go

@@ -0,0 +1,142 @@
+// +build plugins
+
+package main
+
+import (
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"github.com/emqx/kuiper/common"
+	"github.com/emqx/kuiper/xstream/api"
+	//	_ "github.com/taosdata/driver-go/taosSql"
+	"reflect"
+	"strings"
+)
+
+type (
+	taosConfig struct {
+		Port     int      `json:"port"`
+		Ip       string   `json:"ip"`
+		User     string   `json:"user"`
+		Password string   `json:"password"`
+		Database string   `json:"database"`
+		Table    string   `json:"table"`
+		Fields   []string `json:"fields"`
+	}
+	taosSink struct {
+		conf *taosConfig
+		db   *sql.DB
+	}
+)
+
+func (this *taosConfig) buildSql(ctx api.StreamContext, mapData map[string]interface{}) string {
+	if 0 == len(mapData) {
+		return ""
+	}
+	logger := ctx.GetLogger()
+	var keys, vals []string
+	for _, k := range this.Fields {
+		if v, ok := mapData[k]; ok {
+			keys = append(keys, k)
+			if reflect.String == reflect.TypeOf(v).Kind() {
+				vals = append(vals, fmt.Sprintf(`"%v"`, v))
+			} else {
+				vals = append(vals, fmt.Sprintf(`%v`, v))
+			}
+		} else {
+			logger.Debug("not found field:", k)
+		}
+	}
+	if 0 != len(keys) {
+		if len(this.Fields) < len(mapData) {
+			logger.Warnln("some of values will be ignored.")
+		}
+		return fmt.Sprintf(`INSERT INTO %s (%s)VALUES(%s);`, this.Table, strings.Join(keys, `,`), strings.Join(vals, `,`))
+	}
+
+	for k, v := range mapData {
+		keys = append(keys, k)
+		if reflect.String == reflect.TypeOf(v).Kind() {
+			vals = append(vals, fmt.Sprintf(`"%v"`, v))
+		} else {
+			vals = append(vals, fmt.Sprintf(`%v`, v))
+		}
+	}
+	if 0 != len(keys) {
+		return fmt.Sprintf(`INSERT INTO %s (%s)VALUES(%s);`, this.Table, strings.Join(keys, `,`), strings.Join(vals, `,`))
+	}
+	return ""
+}
+
+func (m *taosSink) Configure(props map[string]interface{}) error {
+	cfg := &taosConfig{}
+	err := common.MapToStruct(props, cfg)
+	if err != nil {
+		return fmt.Errorf("read properties %v fail with error: %v", props, err)
+	}
+	if cfg.Ip == "" {
+		return fmt.Errorf("property ip is required")
+	}
+	if cfg.User == "" {
+		return fmt.Errorf("property user is required")
+	}
+	if cfg.Password == "" {
+		return fmt.Errorf("property password is required")
+	}
+	if cfg.Database == "" {
+		return fmt.Errorf("property database is required")
+	}
+	if cfg.Table == "" {
+		return fmt.Errorf("property table is required")
+	}
+	m.conf = cfg
+	return nil
+}
+
+func (m *taosSink) Open(ctx api.StreamContext) (err error) {
+	logger := ctx.GetLogger()
+	logger.Debug("Opening taos sink")
+	url := fmt.Sprintf(`%s:%s@tcp(%s:%d)/%s`, m.conf.User, m.conf.Password, m.conf.Ip, m.conf.Port, m.conf.Database)
+	m.db, err = sql.Open("taosSql", url)
+	return err
+}
+
+func (m *taosSink) Collect(ctx api.StreamContext, item interface{}) error {
+	logger := ctx.GetLogger()
+	data, ok := item.([]byte)
+	if !ok {
+		logger.Debug("taos sink receive non string data")
+		return nil
+	}
+	logger.Debugf("taos sink receive %s", item)
+
+	var sliData []map[string]interface{}
+	err := json.Unmarshal(data, &sliData)
+	if nil != err {
+		return err
+	}
+	for _, mapData := range sliData {
+		sql := m.conf.buildSql(ctx, mapData)
+		if 0 == len(sql) {
+			continue
+		}
+		logger.Debugf(sql)
+		rows, err := m.db.Query(sql)
+		if err != nil {
+			return err
+		}
+		rows.Close()
+	}
+	return nil
+}
+
+func (m *taosSink) Close(ctx api.StreamContext) error {
+	if m.db != nil {
+		return m.db.Close()
+	}
+	return nil
+}
+
+func Taos() api.Sink {
+	return &taosSink{}
+}