Repository: lf-edge/ekuiper
Pull request #1533 (2022-11-24 12:21:20)

refactor(plugin): rename rtsp source to video source
Signed-off-by: Jiyong Huang <huangjy@emqx.io>


Patch:
diff --git a/.github/workflows/build_packages.yaml b/.github/workflows/build_packages.yaml
index 1c53c514..d8689351 100644
--- a/.github/workflows/build_packages.yaml
+++ b/.github/workflows/build_packages.yaml
@@ -186,7 +186,7 @@ jobs:
           - sources/random
           - sources/zmq
           - sources/sql
-          - sources/rtsp
+          - sources/video
           - functions/accumulateWordCount
           - functions/countPlusOne
           - functions/echo
diff --git a/Makefile b/Makefile
index 08414fd4..af08ffe5 100644
--- a/Makefile
+++ b/Makefile
@@ -90,7 +90,7 @@ PLUGINS := sinks/file \
 	sources/random \
 	sources/zmq \
 	sources/sql \
-	sources/rtsp \
+	sources/video \
 	sinks/tdengine \
 	functions/accumulateWordCount \
 	functions/countPlusOne \
diff --git a/docs/directory.json b/docs/directory.json
index 05e25e0a..907d9b4b 100644
--- a/docs/directory.json
+++ b/docs/directory.json
@@ -234,8 +234,8 @@
 									"path": "rules/sources/plugin/random"
 								},
 								{
-									"title": "RTSP 视频源",
-									"path": "rules/sources/plugin/rtsp"
+									"title": "视频源",
+									"path": "rules/sources/plugin/video"
 								}
 							]
 						}
@@ -819,8 +819,8 @@
 									"path": "rules/sources/plugin/random"
 								},
 								{
-									"title": "RTSP source",
-									"path": "rules/sources/plugin/rtsp"
+									"title": "Video source",
+									"path": "rules/sources/plugin/video"
 								}
 							]
 						}
diff --git a/docs/en_US/rules/sources/plugin/rtsp.md b/docs/en_US/rules/sources/plugin/video.md
similarity index 57%
rename from docs/en_US/rules/sources/plugin/rtsp.md
rename to docs/en_US/rules/sources/plugin/video.md
index b2524e0d..1f217eb2 100644
--- a/docs/en_US/rules/sources/plugin/rtsp.md
+++ b/docs/en_US/rules/sources/plugin/video.md
@@ -1,25 +1,25 @@
-# RTSP Source
+# Video Source
 
 <span style="background:green;color:white;">stream source</span>
 <span style="background:green;color:white">scan table source</span>
 
-The source will query rtsp video streams by `ffmpeg` command to get images.
+The source will query video streams such as RTSP encoded stream by `ffmpeg` command to get images.
 
 ## Compile & deploy plugin
 
 ```shell
 # cd $eKuiper_src
-# go build -trimpath -modfile extensions.mod --buildmode=plugin -o plugins/sources/Rtsp.so extensions/sources/rtsp/rtsp.go
-# cp plugins/sources/Rtsp.so $eKuiper_install/plugins/sources
-# cp plugins/sources/rtsp.json $eKuiper_install/etc/sources
-# cp plugins/sources/rtsp.yaml $eKuiper_install/etc/sources
+# go build -trimpath -modfile extensions.mod --buildmode=plugin -o plugins/sources/Video.so extensions/sources/video/video.go
+# cp plugins/sources/Video.so $eKuiper_install/plugins/sources
+# cp plugins/sources/video.json $eKuiper_install/etc/sources
+# cp plugins/sources/video.yaml $eKuiper_install/etc/sources
 ```
 
 Restart the eKuiper server to activate the plugin.
 
 ## Configuration
 
-The configuration for this source is `$ekuiper/etc/sources/rtsp.yaml`. The format is as below:
+The configuration for this source is `$ekuiper/etc/sources/video.yaml`. The format is as below:
 
 ```yaml
 default:
@@ -35,11 +35,11 @@ dedup:
 ```
 ### Global configurations
 
-Use can specify the global rtsp source settings here. The configuration items specified in `default` section will be taken as default settings for the source when running this source.
+Use can specify the global video source settings here. The configuration items specified in `default` section will be taken as default settings for the source when running this source.
 
 ### url
 
-The url address for the video streaming
+The url address for the video streaming.
 
 ### interval
 
@@ -55,7 +55,7 @@ If you have a specific connection that need to overwrite the default settings, y
 ```
 demo (
 		...
-	) WITH (FORMAT="JSON", CONF_KEY="ext", TYPE="rtsp");
+	) WITH (FORMAT="JSON", CONF_KEY="ext", TYPE="video");
 ```
 
 The configuration keys "ext" will be used.
diff --git a/docs/zh_CN/rules/sources/plugin/rtsp.md b/docs/zh_CN/rules/sources/plugin/video.md
similarity index 68%
rename from docs/zh_CN/rules/sources/plugin/rtsp.md
rename to docs/zh_CN/rules/sources/plugin/video.md
index 882352e3..1e098f9a 100644
--- a/docs/zh_CN/rules/sources/plugin/rtsp.md
+++ b/docs/zh_CN/rules/sources/plugin/video.md
@@ -1,25 +1,25 @@
-# RTSP 源
+# 视频源
 
 <span style="background:green;color:white;">stream source</span>
 <span style="background:green;color:white">scan table source</span>
 
-RTSP 源会通过 `ffmpeg` 命令查询 RTSP 视频流获取图片
+视频源会通过 `ffmpeg` 命令查询视频流，例如 RTSP 视频流获取图片
 
 ## 编译和部署插件
 
 ```shell
 # cd $eKuiper_src
-# go build -trimpath -modfile extensions.mod --buildmode=plugin -o plugins/sources/Rtsp.so extensions/sources/rtsp/rtsp.go
-# cp plugins/sources/Rtsp.so $eKuiper_install/plugins/sources
-# cp plugins/sources/rtsp.json $eKuiper_install/etc/sources
-# cp plugins/sources/rtsp.yaml $eKuiper_install/etc/sources
+# go build -trimpath -modfile extensions.mod --buildmode=plugin -o plugins/sources/Video.so extensions/sources/video/video.go
+# cp plugins/sources/Video.so $eKuiper_install/plugins/sources
+# cp plugins/sources/video.json $eKuiper_install/etc/sources
+# cp plugins/sources/video.yaml $eKuiper_install/etc/sources
 ```
 
 重新启动 eKuiper 服务器以激活插件。
 
 ## 配置
 
-该源的配置为 `$ekuiper/etc/sources/rtsp.yaml`。格式如下：
+该源的配置为 `$ekuiper/etc/sources/video.yaml`。格式如下：
 
 ```yaml
 default:
@@ -38,7 +38,7 @@ dedup:
 
 ### url
 
-RTSP 视频源地址
+视频源地址
 
 ### interval
 
@@ -54,7 +54,7 @@ RTSP 视频源地址
 ```
 demo (
 		...
-	) WITH (FORMAT="JSON", CONF_KEY="ext", TYPE="rtsp");
+	) WITH (FORMAT="JSON", CONF_KEY="ext", TYPE="video");
 ```
 
 配置键 "ext" 将被使用。
diff --git a/extensions/sources/rtsp/install.sh b/extensions/sources/video/install.sh
similarity index 97%
rename from extensions/sources/rtsp/install.sh
rename to extensions/sources/video/install.sh
index b108b9a5..ff786eea 100644
--- a/extensions/sources/rtsp/install.sh
+++ b/extensions/sources/video/install.sh
@@ -1,6 +1,6 @@
 #!/bin/sh
 #
-# Copyright 2021 EMQ Technologies Co., Ltd.
+# Copyright 2021-2022 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.
diff --git a/extensions/sources/rtsp/rtsp.go b/extensions/sources/video/video.go
similarity index 79%
rename from extensions/sources/rtsp/rtsp.go
rename to extensions/sources/video/video.go
index e7cd43fa..e24847ea 100644
--- a/extensions/sources/rtsp/rtsp.go
+++ b/extensions/sources/video/video.go
@@ -27,12 +27,12 @@ import (
 const RTSP_DEFAULT_INTERVAL = 10000
 const FRAMENUMBER = 5
 
-type RTSPPullSource struct {
+type VideoPullSource struct {
 	url      string
 	interval int
 }
 
-func (rps *RTSPPullSource) Configure(_ string, props map[string]interface{}) error {
+func (rps *VideoPullSource) Configure(_ string, props map[string]interface{}) error {
 
 	if u, ok := props["url"]; ok {
 		if p, ok := u.(string); ok {
@@ -53,18 +53,18 @@ func (rps *RTSPPullSource) Configure(_ string, props map[string]interface{}) err
 	return nil
 }
 
-func (rps *RTSPPullSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error) {
+func (rps *VideoPullSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error) {
 	rps.initTimerPull(ctx, consumer, errCh)
 }
 
-func (rps *RTSPPullSource) Close(ctx api.StreamContext) error {
+func (rps *VideoPullSource) Close(ctx api.StreamContext) error {
 	logger := ctx.GetLogger()
-	logger.Infof("Closing rtsp pull source")
+	logger.Infof("Closing video pull source")
 
 	return nil
 }
 
-func (rps *RTSPPullSource) initTimerPull(ctx api.StreamContext, consumer chan<- api.SourceTuple, _ chan<- error) {
+func (rps *VideoPullSource) initTimerPull(ctx api.StreamContext, consumer chan<- api.SourceTuple, _ chan<- error) {
 	ticker := time.NewTicker(time.Millisecond * time.Duration(rps.interval))
 	logger := ctx.GetLogger()
 	defer ticker.Stop()
@@ -91,7 +91,7 @@ func (rps *RTSPPullSource) initTimerPull(ctx api.StreamContext, consumer chan<-
 	}
 }
 
-func (rps *RTSPPullSource) readFrameAsJpeg(ctx api.StreamContext) *bytes.Buffer {
+func (rps *VideoPullSource) readFrameAsJpeg(ctx api.StreamContext) *bytes.Buffer {
 	logger := ctx.GetLogger()
 	buf := bytes.NewBuffer(nil)
 	err := ffmpeg.Input(rps.url).
@@ -106,6 +106,6 @@ func (rps *RTSPPullSource) readFrameAsJpeg(ctx api.StreamContext) *bytes.Buffer
 	return buf
 }
 
-func Rtsp() api.Source {
-	return &RTSPPullSource{}
+func Video() api.Source {
+	return &VideoPullSource{}
 }
diff --git a/extensions/sources/rtsp/rtsp.json b/extensions/sources/video/video.json
similarity index 75%
rename from extensions/sources/rtsp/rtsp.json
rename to extensions/sources/video/video.json
index 4a8fdf64..567b5c5b 100644
--- a/extensions/sources/rtsp/rtsp.json
+++ b/extensions/sources/video/video.json
@@ -9,12 +9,12 @@
 			"website": "https://www.emqx.io"
 		},
 		"helpUrl": {
-			"en_US": "https://github.com/lf-edge/ekuiper/tree/master/docs/en_US/rules/sources/plugin/rtsp.md",
-			"zh_CN": "https://github.com/lf-edge/ekuiper/tree/master/docs/zh_CN/rules/sources/plugin/rtsp.md"
+			"en_US": "https://github.com/lf-edge/ekuiper/tree/master/docs/en_US/rules/sources/plugin/video.md",
+			"zh_CN": "https://github.com/lf-edge/ekuiper/tree/master/docs/zh_CN/rules/sources/plugin/video.md"
 		},
 		"description": {
-			"en_US": "eKuiper provides plug in support for RTSP source stream, which can pull images from RTSP server",
-			"zh_CN": "eKuiper 为提取 RTSP 源中的图片流提供了插件支持，该支持可从 RTSP 服务器提取图片消息并输入 eKuiper 处理管道。"
+			"en_US": "eKuiper provides plug in support for video stream, which can pull images from a video stream",
+			"zh_CN": "eKuiper 为提取视频源中的图片流提供了插件支持，该支持可从视频流提取图片消息并输入 eKuiper 处理管道。"
 		}
 	},
 	"properties": {
@@ -61,8 +61,8 @@
 		"category": "source",
 		"icon": "iconPath",
 		"label": {
-			"en_US": "RTSP PULL",
-			"zh_CN": "RTSP PULL"
+			"en_US": "Video",
+			"zh_CN": "视频"
 		}
 	}
 }
diff --git a/extensions/sources/rtsp/rtsp.yaml b/extensions/sources/video/video.yaml
similarity index 74%
rename from extensions/sources/rtsp/rtsp.yaml
rename to extensions/sources/video/video.yaml
index 359810dc..3485f860 100644
--- a/extensions/sources/rtsp/rtsp.yaml
+++ b/extensions/sources/video/video.yaml
@@ -1,4 +1,4 @@
-#Global RTSP configurations
+#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
diff --git a/internal/xsql/row.go b/internal/xsql/row.go
index 57ab8ad2..a3e5521f 100644
--- a/internal/xsql/row.go
+++ b/internal/xsql/row.go
@@ -154,7 +154,10 @@ func (d *AffiliateRow) MergeMap(cachedMap map[string]interface{}) {
 	d.lock.RLock()
 	defer d.lock.RUnlock()
 	for k, v := range d.CalCols {
-		cachedMap[k] = v
+		// Do not write out the internal fields
+		if !strings.HasPrefix(k, "$$") {
+			cachedMap[k] = v
+		}
 	}
 	for k, v := range d.AliasMap {
 		cachedMap[k] = v


Test patch:
diff --git a/internal/topo/operator/analyticfuncs_operator_test.go b/internal/topo/operator/analyticfuncs_operator_test.go
index 536db349..ecc969a4 100644
--- a/internal/topo/operator/analyticfuncs_operator_test.go
+++ b/internal/topo/operator/analyticfuncs_operator_test.go
@@ -31,7 +31,7 @@ func TestAnalyticFuncs(t *testing.T) {
 	var tests = []struct {
 		funcs  []*ast.Call
 		data   []interface{}
-		result [][]map[string]interface{}
+		result []map[string]interface{}
 	}{
 		{ // 0 Lag test
 			funcs: []*ast.Call{
@@ -85,17 +85,16 @@ func TestAnalyticFuncs(t *testing.T) {
 					},
 				},
 			},
-			result: [][]map[string]interface{}{{{
+			result: []map[string]interface{}{{
 				"$$a_lag_0": nil,
 				"$$a_lag_1": nil,
-				"a":         "a1", "b": "b1", "c": "c1",
-			}}, {{
-				"$$a_lag_0": "a1", "$$a_lag_1": "b1", "a": "a1", "b": "b2", "c": "c1",
-			}}, {{
-				"$$a_lag_0": "a1", "$$a_lag_1": "b2", "a": "a1", "c": "c1",
-			}}, {{
-				"$$a_lag_0": "a1", "$$a_lag_1": interface{}(nil), "a": "a1", "b": "b2", "c": "c2",
-			}}},
+			}, {
+				"$$a_lag_0": "a1", "$$a_lag_1": "b1",
+			}, {
+				"$$a_lag_0": "a1", "$$a_lag_1": "b2",
+			}, {
+				"$$a_lag_0": "a1", "$$a_lag_1": interface{}(nil),
+			}},
 		},
 		{ // 1 changed test
 			funcs: []*ast.Call{
@@ -157,16 +156,16 @@ func TestAnalyticFuncs(t *testing.T) {
 					},
 				},
 			},
-			result: [][]map[string]interface{}{
-				{{
-					"$$a_changed_col_0": "a1", "$$a_had_changed_0": false, "$$a_lag_1": nil, "a": "a1", "b": "b1",
-				}}, {{
-					"$$a_changed_col_0": nil, "$$a_had_changed_0": true, "$$a_lag_1": "b1", "a": "a1", "c": "c1",
-				}}, {{
-					"$$a_changed_col_0": nil, "$$a_had_changed_0": false, "$$a_lag_1": nil, "a": "a1", "c": "c1",
-				}}, {{
-					"$$a_changed_col_0": nil, "$$a_had_changed_0": true, "$$a_lag_1": nil, "a": "a1", "b": "b2", "c": "c2",
-				}},
+			result: []map[string]interface{}{
+				{
+					"$$a_changed_col_0": "a1", "$$a_had_changed_0": false, "$$a_lag_1": nil,
+				}, {
+					"$$a_changed_col_0": nil, "$$a_had_changed_0": true, "$$a_lag_1": "b1",
+				}, {
+					"$$a_changed_col_0": nil, "$$a_had_changed_0": false, "$$a_lag_1": nil,
+				}, {
+					"$$a_changed_col_0": nil, "$$a_had_changed_0": true, "$$a_lag_1": nil,
+				},
 			},
 		},
 	}
@@ -179,15 +178,10 @@ func TestAnalyticFuncs(t *testing.T) {
 
 		pp := &AnalyticFuncsOp{Funcs: tt.funcs}
 		fv, afv := xsql.NewFunctionValuersForOp(ctx)
-		r := make([][]map[string]interface{}, 0, len(tt.data))
+		r := make([]map[string]interface{}, 0, len(tt.data))
 		for _, d := range tt.data {
 			opResult := pp.Apply(ctx, d, fv, afv)
-			result, err := parseResult(opResult, false)
-			if err != nil {
-				t.Errorf("parse result error： %s", err)
-				continue
-			}
-			r = append(r, result)
+			r = append(r, opResult.(*xsql.Tuple).CalCols)
 		}
 
 		if !reflect.DeepEqual(tt.result, r) {
diff --git a/internal/xsql/row_test.go b/internal/xsql/row_test.go
index 0520b678..67dcb45a 100644
--- a/internal/xsql/row_test.go
+++ b/internal/xsql/row_test.go
@@ -75,12 +75,12 @@ func TestCollectionRow(t *testing.T) {
 			result:   []interface{}{4, "v1", 4, Message{"a": 4, "b": "v1", "d": 4}},
 		}, {
 			rowC: &JoinTuples{Content: []*JoinTuple{{Tuples: []TupleRow{
-				&Tuple{Emitter: "src1", Message: Message{"a": 1, "b": "v1"}},
+				&Tuple{Emitter: "src1", Message: Message{"a": 1, "b": "v1"}, AffiliateRow: AffiliateRow{CalCols: map[string]interface{}{"b": "v2", "$$lag_a": 1}}},
 				&Tuple{Emitter: "src2", Message: Message{"a": 2, "c": "w2"}},
 			}}}, AffiliateRow: AffiliateRow{CalCols: map[string]interface{}{"a": 4, "d": 3}, AliasMap: map[string]interface{}{"d": 4}}},
 			value:    []string{"a", "b", "d"},
 			wildcard: []string{""},
-			result:   []interface{}{4, "v1", 4, Message{"a": 4, "b": "v1", "c": "w2", "d": 4}},
+			result:   []interface{}{4, "v2", 4, Message{"a": 4, "b": "v2", "c": "w2", "d": 4}},
 		},
 	}
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))