Repository: lf-edge/ekuiper
Pull request #1570 (2022-12-20 09:52:27)

doc(source): doc for file source


Patch:
diff --git a/docs/en_US/rules/sources/builtin/file.md b/docs/en_US/rules/sources/builtin/file.md
index f1e762bd..24c38afe 100644
--- a/docs/en_US/rules/sources/builtin/file.md
+++ b/docs/en_US/rules/sources/builtin/file.md
@@ -1,8 +1,12 @@
 ## File source
 
+<span style="background:green;color:white;">stream source</span>
 <span style="background:green;color:white">scan table source</span>
 
-eKuiper provides built-in support for reading file content into the eKuiper processing pipeline. The file source is usually used as a [table](../../../sqls/tables.md) and it is the default type for create table statement.
+eKuiper provides built-in support for reading file content into the eKuiper processing pipeline. The file source is
+usually used as a [table](../../../sqls/tables.md) and it is the default type for create table statement. File sources
+are also supported as streams, where it is usually necessary to set the `interval` parameter to pull updates at regular
+intervals.
 
 ```sql
 create table table1 (
@@ -18,12 +22,30 @@ The configure file for the file source is in */etc/sources/file.yaml* in which t
 
 ```yaml
 default:
+  # The type of the file, could be json, csv and lines
   fileType: json
-  # The directory of the file relative to eKuiper root or an absolute path.
+  # The directory of the file relative to kuiper root or an absolute path.
   # Do not include the file name here. The file name should be defined in the stream data source
   path: data
   # The interval between reading the files, time unit is ms. If only read once, set it to 0
   interval: 0
+  # The sending interval between each event in millisecond
+  sendInterval: 0
+  # After read
+  # 0: keep the file
+  # 1: delete the file
+  # 2: move the file to moveTo
+  actionAfterRead: 0
+  # The path to move the file to after read, only valid when the actionAfterRead is 2
+  moveTo: /tmp/kuiper/moved
+  # If the first line is header
+  hasHeader: false
+  # Define the columns. If header is defined, this will be override
+  # columns: [id, name]
+  # How many lines to be ignored at the beginning. Notice that, empty line will be ignored and not be calculated.
+  ignoreStartLines: 0
+  # How many lines to be ignored in the end. Notice that, empty line will be ignored and not be calculated.
+  ignoreEndLines: 0
 ```
 
 With this yaml file, the table will refer to the file *${eKuiper}/data/lookup.json* and read it in json format.
\ No newline at end of file
diff --git a/docs/zh_CN/rules/sources/builtin/file.md b/docs/zh_CN/rules/sources/builtin/file.md
index 7209178e..d3eed542 100644
--- a/docs/zh_CN/rules/sources/builtin/file.md
+++ b/docs/zh_CN/rules/sources/builtin/file.md
@@ -1,8 +1,10 @@
 ## 文件源
 
+<span style="background:green;color:white;">stream source</span>
 <span style="background:green;color:white">scan table source</span>
 
-eKuiper 提供了内置支持，可将文件内容读入 eKuiper 处理管道。 文件源通常用作 [表格](../../../sqls/tables.md)， 并且采用 create table 语句的默认类型。
+eKuiper 提供了内置支持，可将文件内容读入 eKuiper 处理管道。 文件源通常用作[表格](../../../sqls/tables.md)， 并且采用 create
+table 语句的默认类型。文件源也支持作为用作流，此时通常需要设置 `interval` 参数以定时拉取更新。
 
 ```sql
 CREATE TABLE table1 (
@@ -18,12 +20,108 @@ CREATE TABLE table1 (
 
 ```yaml
 default:
+  # 文件的类型，支持 json， csv 和 lines
   fileType: json
   # 文件以 eKuiper 为根目录的目录或文件的绝对路径。
   # 请勿在此处包含文件名。文件名应在流数据源中定义
   path: data
   # 读取文件的时间间隔，单位为ms。 如果只读取一次，则将其设置为 0
   interval: 0
+  # 读取后，两条数据发送的间隔时间
+  sendInterval: 0
+  # 文件读取后的操作
+  # 0: 文件保持不变
+  # 1: 删除文件
+  # 2: 移动文件到 moveTo 定义的位置
+  actionAfterRead: 0
+  # 移动文件的位置, 仅用于 actionAfterRead 为 2 的情况
+  moveTo: /tmp/kuiper/moved
+  # 是否包含文件头，多用于 csv。若为 true，则第一行解析为文件头。
+  hasHeader: false
+  # 定义文件的列。如果定义了文件头，该选项将被覆盖。
+  # columns: [id, name]
+  # 忽略开头多少行的内容。
+  ignoreStartLines: 0
+  # 忽略结尾多少行的内容。最后的空行不计算在内。
+  ignoreEndLines: 0
 ```
 
-通过这个 yaml 文件，该表将引用文件 *${eKuiper}/data/lookup.json* 并以 json 格式读取它。
\ No newline at end of file
+### File Types
+
+The file source supports monitoring files or folders. If the monitored location is a folder, all files in the folder are
+required to be of the same type. When monitoring a folder, it will read in files order by file name alphabetically.
+
+The supported file types are
+
+- json: standard JSON array format files,
+  see [example](https://github.com/lf-edge/ekuiper/tree/master/internal/topo/source/test/test.json). If the file format
+  is a line-separated JSON string, it needs to be defined in lines format.
+- csv: comma-separated csv files are supported, as well as custom separators.
+- lines: line-separated file. The decoding method of each line can be defined by the format parameter in the stream
+  definition. For example, for a line-separated JSON string, the file type is set to lines and the format is set to
+  json.
+
+Some files may have most of the data in standard format, but have some metadata in the opening and closing lines of the
+file. The user can use the `ignoreStartLines` and `ignoreEndLines` arguments to remove the non-standard parts of the
+beginning and end so that the above file types can be parsed.
+
+### Example
+
+File sources involve the parsing of file contents and intersect with format-related definitions in data streams. We
+describe with some examples how to combine file types and formats for parsing file sources.
+
+#### Read a csv with a custom separator
+
+The standard csv separator is a comma, but there are a large number of files that use the csv-like format with custom
+separators. Some csv-like files have column names defined in the first line instead of data.
+
+```csv
+id name age
+1 John 56
+2 Jane 34
+```
+
+When the file is read, the configuration file is as follows, specifying that the file has a header.
+
+```yaml
+csv:
+  fileType: csv
+  hasHeader: true
+```
+
+In the stream definition, set the stream data to ``DELIMITED`` format, specifying the separator with the ``DELIMITER``
+parameter.
+
+```SQL
+create
+stream cscFileDemo () WITH (FORMAT="DELIMITED", DATASOURCE="abc.csv", TYPE="file", DELIMITER=" ", CONF_KEY="csv"
+```
+
+#### Read multi-line JSON data
+
+With a standard JSON file, the entire file should be a JSON object or an array. In practice, we often need to parse
+files that contain multiple JSON objects. These files are not actually JSON themselves, but are considered to be
+multiple lines of JSON data, assuming that each JSON object is a single line.
+
+```text
+{"id": 1, "name": "John Doe"}
+{"id": 2, "name": "Jane Doe"}
+{"id": 3, "name": "John Smith"}
+```
+
+When reading this file, the configuration file is as follows, specifying the file type as lines.
+
+```yaml
+jsonlines:
+  fileType: lines
+```
+
+In the stream definition, set the stream data to be in ``JSON`` format.
+
+```SQL
+create
+stream linesFileDemo () WITH (FORMAT="JSON", TYPE="file", CONF_KEY="jsonlines"
+```
+
+Moreover, the lines file type can be combined with any format. For example, if you set the format to protobuf and
+configure the schema, it can be used to parse data that contains multiple Protobuf encoded lines.
\ No newline at end of file
diff --git a/etc/sources/file.json b/etc/sources/file.json
index 087fd206..e83520f3 100644
--- a/etc/sources/file.json
+++ b/etc/sources/file.json
@@ -12,8 +12,8 @@
       "zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/rules/sources/builtin/file.md"
     },
     "description": {
-      "en_US": "Reading file content into the eKuiper processing pipeline. The file source is usually used as a table and it is the default type for create table statement.",
-      "zh_CN": "读取文件内容，并且将数据放入 eKuiper 数据处理流水线中。通常用于表格，且为使用 create table 语句时，默认的源类型。"
+      "en_US": "Monitor file system and load the content into the eKuiper processing pipeline.",
+      "zh_CN": "监控文件系统，读取文件内容，并且将数据放入 eKuiper 数据处理流水线中。"
     }
   },
   "libs": [],
@@ -24,8 +24,8 @@
       "zh_CN": "不含路径的文件名，例如 test.json"
     },
     "label": {
-      "en_US": "Data Source (File Name)",
-      "zh_CN": "数据源（文件名）"
+      "en_US": "Data Source (File or directory relative path)",
+      "zh_CN": "数据源（文件或者目录的相对地址）"
     }
   },
   "properties": {
@@ -34,8 +34,13 @@
         "name": "fileType",
         "default": "json",
         "optional": true,
-        "control": "text",
+        "control": "select",
         "type": "string",
+        "values": [
+          "json",
+          "csv",
+          "lines"
+        ],
         "hint": {
           "en_US": "The file format type.",
           "zh_CN": "文件格式类型"
@@ -61,17 +66,120 @@
       },{
         "name": "interval",
         "default": 0,
-        "optional": false,
+        "optional": true,
         "control": "text",
         "type": "int",
         "hint": {
           "en_US": "The interval between reading the files, time unit is ms. If only read once, set it to 0",
-          "zh_CN": "读取文件的间隔时间。如果只需读取一次，设置为0。"
+          "zh_CN": "读取文件的间隔时间，单位为毫秒。如果只需读取一次，设置为0。"
         },
         "label": {
           "en_US": "Interval",
           "zh_CN": "间隔时间"
         }
+      },{
+        "name": "sendInterval",
+        "default": 0,
+        "optional": true,
+        "control": "text",
+        "type": "int",
+        "hint": {
+          "en_US": "The sending interval between each event in millisecond.",
+          "zh_CN": "事件发送的间隔时间，单位为毫秒。"
+        },
+        "label": {
+          "en_US": "Send Interval",
+          "zh_CN": "发送间隔"
+        }
+      },{
+        "name": "actionAfterRead",
+        "default": 0,
+        "optional": true,
+        "control": "select",
+        "type": "int",
+        "values": [
+          0,
+          1,
+          2
+        ],
+        "hint": {
+          "en_US": "The action after read. 0 meas keep the file; 1 means delete the file; 2 means move the file to the path defined in the property 'moveTo'",
+          "zh_CN": "读取后的操作。0表示保留文件；1表示删除文件；2表示将文件移动到属性 'moveTo' 中定义的路径。"
+        },
+        "label": {
+          "en_US": "Action after read",
+          "zh_CN": "读取后动作"
+        }
+      },{
+        "name": "moveTo",
+        "default": "",
+        "optional": true,
+        "control": "text",
+        "type": "string",
+        "hint": {
+          "en_US": "The path to move the file to after read, only valid when the actionAfterRead is 1.",
+          "zh_CN": "移动文件的位置, 仅用于 actionAfterRead 为 2 的情况"
+        },
+        "label": {
+          "en_US": "Move to path",
+          "zh_CN": "移动位置"
+        }
+      },{
+        "name": "hasHeader",
+        "default": false,
+        "optional": true,
+        "control": "radio",
+        "type": "bool",
+        "hint": {
+          "en_US": "If the first line is header, usually used for csv file.",
+          "zh_CN": "是否包含文件头，多用于 csv。若为 true，则第一行解析为文件头。"
+        },
+        "label": {
+          "en_US": "Has header",
+          "zh_CN": "是否包含文件头"
+        }
+      },{
+        "name": "columns",
+        "default": [],
+        "optional": true,
+        "control": "list",
+        "type": "list_string",
+        "hint": {
+          "en_US": "Define the columns. If header is defined, this will be override.",
+          "zh_CN": "定义文件的列。如果定义了文件头，该选项将被覆盖。"
+        },
+        "label": {
+          "en_US": "Columns",
+          "zh_CN": "字段列表"
+        }
+      },{
+        "name": "ignoreStartLines",
+        "default": 0,
+        "optional": true,
+        "control": "text",
+        "type": "int",
+        "hint": {
+          "en_US": "How many lines to be ignored at the beginning. Notice that, empty line will be ignored and not be calculated.",
+          "zh_CN": "忽略开头多少行的内容。"
+        },
+        "label": {
+          "en_US": "Ignore start lines",
+          "zh_CN": "文件开头忽略的行数"
+        }
+      },{
+        "name": "ignoreEndLines",
+        "default": 0,
+        "optional": true,
+        "control": "text",
+        "type": "int",
+        "hint": {
+          "en_US": "How many lines to be ignored in the end. Notice that, empty line will be ignored and not be calculated.",
+          "zh_CN": "忽略结尾多少行的内容。最后的空行不计算在内。。"
+        },
+        "label": {
+          "en_US": "Ignore end lines",
+          "zh_CN": "文件结尾忽略的行数"
+        }
       }]
   },
   "outputs": [
diff --git a/etc/sources/file.yaml b/etc/sources/file.yaml
index 641c577a..4441ee6b 100644
--- a/etc/sources/file.yaml
+++ b/etc/sources/file.yaml
@@ -6,14 +6,14 @@ default:
   path: data
   # The interval between reading the files, time unit is ms. If only read once, set it to 0
   interval: 0
-  # When using as a stream, the sending interval between each event in millisecond
-  sendInterval: 10
+  # The sending interval between each event in millisecond
+  sendInterval: 0
   # After read
   # 0: keep the file
   # 1: delete the file
   # 2: move the file to moveTo
   actionAfterRead: 0
-  # The path to move the file to after read, only valid when the actionAfterRead is 1
+  # The path to move the file to after read, only valid when the actionAfterRead is 2
   moveTo: /tmp/kuiper/moved
   # If the first line is header
   hasHeader: false
diff --git a/extensions.mod b/extensions.mod
index ded8fcad..e578932f 100644
--- a/extensions.mod
+++ b/extensions.mod
@@ -16,7 +16,7 @@ require (
 	github.com/btnguyen2k/gocosmos v0.1.6
 	github.com/couchbase/go_n1ql v0.0.0-20220303011133-0ed4bf93e31d
 	github.com/denisenkom/go-mssqldb v0.12.2
-	github.com/eclipse/paho.mqtt.golang v1.4.2-0.20220810043731-079a117b4614
+	github.com/eclipse/paho.mqtt.golang v1.4.2
 	github.com/edgexfoundry/go-mod-core-contracts/v2 v2.3.0
 	github.com/edgexfoundry/go-mod-messaging/v2 v2.3.0
 	github.com/fxamacker/cbor/v2 v2.4.0
diff --git a/extensions.sum b/extensions.sum
index e89b672e..94532ee1 100644
--- a/extensions.sum
+++ b/extensions.sum
@@ -261,8 +261,8 @@ github.com/docker/go-units v0.4.0 h1:3uh0PgVws3nIA0Q+MwDC8yjEPf9zjRfZZWXZYDct3Tw
 github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
 github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
-github.com/eclipse/paho.mqtt.golang v1.4.2-0.20220810043731-079a117b4614 h1:36h5MuaQ0gVNV1clhTE4lF3Vram4HNGX3JcDlD3u6UA=
-github.com/eclipse/paho.mqtt.golang v1.4.2-0.20220810043731-079a117b4614/go.mod h1:JGt0RsEwEX+Xa/agj90YJ9d9DH2b7upDZMK9HRbFvCA=
+github.com/eclipse/paho.mqtt.golang v1.4.2 h1:66wOzfUHSSI1zamx7jR6yMEI5EuHnT1G6rNA5PM12m4=
+github.com/eclipse/paho.mqtt.golang v1.4.2/go.mod h1:JGt0RsEwEX+Xa/agj90YJ9d9DH2b7upDZMK9HRbFvCA=
 github.com/edgexfoundry/go-mod-core-contracts/v2 v2.3.0 h1:8Svk1HTehXEgwxgyA4muVhSkP3D9n1q+oSHI3B1Ac90=
 github.com/edgexfoundry/go-mod-core-contracts/v2 v2.3.0/go.mod h1:4/e61acxVkhQWCTjQ4XcHVJDnrMDloFsZZB1B6STCRw=
 github.com/edgexfoundry/go-mod-messaging/v2 v2.3.0 h1:aZfDYjj6n2Lo1P1t6NKuShhHjzFk9/mQV9nuJS3ioZ4=
diff --git a/go.mod b/go.mod
index 82a16035..69191bbb 100644
--- a/go.mod
+++ b/go.mod
@@ -6,7 +6,7 @@ require (
 	github.com/PaesslerAG/jsonpath v0.1.1
 	github.com/alicebob/miniredis/v2 v2.15.1
 	github.com/benbjohnson/clock v1.0.0
-	github.com/eclipse/paho.mqtt.golang v1.4.2-0.20220810043731-079a117b4614
+	github.com/eclipse/paho.mqtt.golang v1.4.2
 	github.com/edgexfoundry/go-mod-core-contracts/v2 v2.3.0
 	github.com/edgexfoundry/go-mod-messaging/v2 v2.3.0
 	github.com/fxamacker/cbor/v2 v2.4.0
diff --git a/go.sum b/go.sum
index 047d9ff2..0609a633 100644
--- a/go.sum
+++ b/go.sum
@@ -52,8 +52,8 @@ github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ3
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/eclipse/paho.mqtt.golang v1.4.2-0.20220810043731-079a117b4614 h1:36h5MuaQ0gVNV1clhTE4lF3Vram4HNGX3JcDlD3u6UA=
-github.com/eclipse/paho.mqtt.golang v1.4.2-0.20220810043731-079a117b4614/go.mod h1:JGt0RsEwEX+Xa/agj90YJ9d9DH2b7upDZMK9HRbFvCA=
+github.com/eclipse/paho.mqtt.golang v1.4.2 h1:66wOzfUHSSI1zamx7jR6yMEI5EuHnT1G6rNA5PM12m4=
+github.com/eclipse/paho.mqtt.golang v1.4.2/go.mod h1:JGt0RsEwEX+Xa/agj90YJ9d9DH2b7upDZMK9HRbFvCA=
 github.com/edgexfoundry/go-mod-core-contracts/v2 v2.3.0 h1:8Svk1HTehXEgwxgyA4muVhSkP3D9n1q+oSHI3B1Ac90=
 github.com/edgexfoundry/go-mod-core-contracts/v2 v2.3.0/go.mod h1:4/e61acxVkhQWCTjQ4XcHVJDnrMDloFsZZB1B6STCRw=
 github.com/edgexfoundry/go-mod-messaging/v2 v2.3.0 h1:aZfDYjj6n2Lo1P1t6NKuShhHjzFk9/mQV9nuJS3ioZ4=
diff --git a/internal/topo/connection/clients/mqtt/mqtt.go b/internal/topo/connection/clients/mqtt/mqtt.go
index ae816460..cb75deae 100644
--- a/internal/topo/connection/clients/mqtt/mqtt.go
+++ b/internal/topo/connection/clients/mqtt/mqtt.go
@@ -24,6 +24,7 @@ import (
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
 	"strings"
+	"time"
 )
 
 type MQTTConnectionConfig struct {
@@ -121,7 +122,7 @@ func (ms *MQTTClient) Connect(connHandler MQTT.OnConnectHandler, lostHandler MQT
 	}
 
 	c := MQTT.NewClient(opts)
-	if token := c.Connect(); token.Wait() && token.Error() != nil {
+	if token := c.Connect(); token.WaitTimeout(5*time.Second) && token.Error() != nil {
 		conf.Log.Errorf("The connection to mqtt broker %s failed : %s ", ms.srv, token.Error())
 		return fmt.Errorf("found error when connecting for %s: %s", ms.srv, token.Error())
 	}
@@ -132,14 +133,14 @@ func (ms *MQTTClient) Connect(connHandler MQTT.OnConnectHandler, lostHandler MQT
 }
 
 func (ms *MQTTClient) Subscribe(topic string, qos byte, handler MQTT.MessageHandler) error {
-	if token := ms.conn.Subscribe(topic, qos, handler); token.Wait() && token.Error() != nil {
+	if token := ms.conn.Subscribe(topic, qos, handler); token.WaitTimeout(5*time.Second) && token.Error() != nil {
 		return fmt.Errorf("%s: %s", errorx.IOErr, token.Error())
 	}
 	return nil
 }
 
 func (ms *MQTTClient) Publish(topic string, qos byte, retained bool, message []byte) error {
-	if token := ms.conn.Publish(topic, qos, retained, message); token.Wait() && token.Error() != nil {
+	if token := ms.conn.Publish(topic, qos, retained, message); token.WaitTimeout(5*time.Second) && token.Error() != nil {
 		return fmt.Errorf("%s: %s", errorx.IOErr, token.Error())
 	}
 	return nil
diff --git a/internal/topo/node/node.go b/internal/topo/node/node.go
index 0bc4d7a5..760886fe 100644
--- a/internal/topo/node/node.go
+++ b/internal/topo/node/node.go
@@ -111,7 +111,8 @@ func (o *defaultNode) doBroadcast(val interface{}) {
 		case <-o.ctx.Done():
 			// rule stop so stop waiting
 		default:
-			o.ctx.GetLogger().Errorf("drop message from %s to %s", o.name, name)
+			o.statManagers[0].IncTotalExceptions(fmt.Sprintf("buffer full, drop message from to %s", name))
+			o.ctx.GetLogger().Debugf("drop message from %s to %s", o.name, name)
 		}
 		switch vt := val.(type) {
 		case xsql.Collection:
diff --git a/internal/topo/source/file_source.go b/internal/topo/source/file_source.go
index 45ae4cbc..160f8ffe 100644
--- a/internal/topo/source/file_source.go
+++ b/internal/topo/source/file_source.go
@@ -322,6 +322,28 @@ func (fs *FileSource) publish(ctx api.StreamContext, file io.Reader, consumer ch
 				time.Sleep(time.Millisecond * time.Duration(fs.config.SendInterval))
 			}
 		}
+	case LINES_TYPE:
+		scanner := bufio.NewScanner(file)
+		scanner.Split(bufio.ScanLines)
+		for scanner.Scan() {
+			var tuple api.SourceTuple
+			m, err := ctx.Decode(scanner.Bytes())
+			if err != nil {
+				tuple = &xsql.ErrorSourceTuple{
+					Error: fmt.Errorf("Invalid data format, cannot decode %s with error %s", scanner.Text(), err),
+				}
+			} else {
+				tuple = api.NewDefaultSourceTuple(m, meta)
+			}
+			select {
+			case consumer <- tuple:
+			case <-ctx.Done():
+				return nil
+			}
+			if fs.config.SendInterval > 0 {
+				time.Sleep(time.Millisecond * time.Duration(fs.config.SendInterval))
+			}
+		}
 	default:
 		return fmt.Errorf("invalid file type %s", fs.config.FileType)
 	}


Test patch:
diff --git a/internal/topo/mock/test_source.go b/internal/topo/mock/test_source.go
index 798eeea9..776b9adb 100644
--- a/internal/topo/mock/test_source.go
+++ b/internal/topo/mock/test_source.go
@@ -16,7 +16,10 @@ package mock
 
 import (
 	"fmt"
+	"github.com/lf-edge/ekuiper/internal/converter"
+	"github.com/lf-edge/ekuiper/internal/topo/context"
 	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/ast"
 	"reflect"
 	"sync/atomic"
 	"testing"
@@ -32,6 +35,8 @@ func TestSourceOpen(r api.Source, exp []api.SourceTuple, t *testing.T) {
 		c = 0
 	}
 	ctx, cancel := NewMockContext(fmt.Sprintf("rule%d", c), "op1").WithCancel()
+	cv, _ := converter.GetOrCreateConverter(&ast.Options{FORMAT: "json"})
+	ctx = context.WithValue(ctx.(*context.DefaultContext), context.DecodeKey, cv)
 	count.Store(c.(int) + 1)
 	consumer := make(chan api.SourceTuple)
 	errCh := make(chan error)
diff --git a/internal/topo/source/file_source_test.go b/internal/topo/source/file_source_test.go
index d628123d..5d413a89 100644
--- a/internal/topo/source/file_source_test.go
+++ b/internal/topo/source/file_source_test.go
@@ -76,6 +76,8 @@ func TestJsonFolder(t *testing.T) {
 		return
 	}
 	mock.TestSourceOpen(r, exp, t)
+	// wait for the move to finish
+	time.Sleep(100 * time.Millisecond)
 	files, err := os.ReadDir(moveToFolder)
 	if err != nil {
 		t.Error(err)
@@ -194,3 +196,29 @@ func TestCSVFile(t *testing.T) {
 	}
 	mock.TestSourceOpen(r, exp, t)
 }
+
+func TestJsonLines(t *testing.T) {
+	path, err := os.Getwd()
+	if err != nil {
+		t.Fatal(err)
+	}
+	meta := map[string]interface{}{
+		"file": filepath.Join(path, "test", "test.lines"),
+	}
+	exp := []api.SourceTuple{
+		api.NewDefaultSourceTuple(map[string]interface{}{"id": float64(1), "name": "John Doe"}, meta),
+		api.NewDefaultSourceTuple(map[string]interface{}{"id": float64(2), "name": "Jane Doe"}, meta),
+		api.NewDefaultSourceTuple(map[string]interface{}{"id": float64(3), "name": "John Smith"}, meta),
+	}
+	p := map[string]interface{}{
+		"path":     filepath.Join(path, "test"),
+		"fileType": "lines",
+	}
+	r := &FileSource{}
+	err = r.Configure("test.lines", p)
+	if err != nil {
+		t.Errorf(err.Error())
+		return
+	}
+	mock.TestSourceOpen(r, exp, t)
+}
diff --git a/internal/topo/source/test/test.lines b/internal/topo/source/test/test.lines
new file mode 100644
index 00000000..d484285c
--- /dev/null
+++ b/internal/topo/source/test/test.lines
@@ -0,0 +1,3 @@
+{"id": 1,"name": "John Doe"}
+{"id": 2,"name": "Jane Doe"}
+{"id": 3,"name": "John Smith"}
\ No newline at end of file