feat(file): support decorator

internal/io/file/source.go

@@ -60,7 +60,9 @@
 	isDir  bool
 	config *SourceConfig
 	reader modules.FileStreamReader
-	eof    api.EOFIngest
+	// attach to a reader
+	decorator modules.FileStreamDecorator
+	eof       api.EOFIngest
 }
 
 func (fs *Source) Provision(ctx api.StreamContext, props map[string]any) error {
@@ -140,6 +142,14 @@
 		}
 	}
 	fs.config = cfg
+	decorator, ok := modules.GetFileStreamDecorator(ctx, cfg.FileType)
+	if ok {
+		err = decorator.Provision(ctx, props)
+		if err != nil {
+			return err
+		}
+		fs.decorator = decorator
+	}
 	return nil
 }
 
@@ -230,7 +240,7 @@
 		maxSize = int(info.Size())
 	}
 	if fs.config.IgnoreStartLines > 0 || fs.config.IgnoreEndLines > 0 {
-		r = ignoreLines(ctx, r, fs.config.IgnoreStartLines, fs.config.IgnoreEndLines)
+		r = ignoreLines(ctx, r, fs.decorator, fs.config.IgnoreStartLines, fs.config.IgnoreEndLines)
 	}
 	if closer, ok := r.(io.Closer); ok {
 		defer func() {
@@ -255,6 +265,9 @@
 				break
 			}
 			rcvTime := timex.GetNow()
+			if fs.decorator != nil {
+				line = fs.decorator.Decorate(ctx, line)
+			}
 			ingest(ctx, line, meta, rcvTime)
 			if fs.config.SendInterval > 0 {
 				time.Sleep(fs.config.SendInterval)
@@ -289,7 +302,7 @@
 	}
 }
 
-func ignoreLines(ctx api.StreamContext, reader io.Reader, ignoreStartLines int, ignoreEndLines int) io.Reader {
+func ignoreLines(ctx api.StreamContext, reader io.Reader, decorator modules.FileStreamDecorator, ignoreStartLines int, ignoreEndLines int) io.Reader {
 	r, w := io.Pipe()
 	go func() {
 		e := infra.SafeRun(func() error {
@@ -305,6 +318,10 @@
 			tempLines := make([][]byte, 0, ignoreEndLines)
 			for scanner.Scan() {
 				if ln >= ignoreStartLines {
+					if ln == ignoreStartLines && decorator != nil {
+						// Send EOF to decorator
+						decorator.ReadMeta(ctx, nil)
+					}
 					if ignoreEndLines > 0 { // the last n line are left in the tempLines
 						slot := (ln - ignoreStartLines) % ignoreEndLines
 						if len(tempLines) <= slot { // first round
@@ -334,6 +351,10 @@
 							break
 						}
 					}
+				} else {
+					if decorator != nil {
+						decorator.ReadMeta(ctx, scanner.Bytes())
+					}
 				}
 				ln++
 			}

internal/io/file/source_decorator_test.go

@@ -0,0 +1,87 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package file
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+
+	"github.com/lf-edge/ekuiper/v2/pkg/mock"
+	"github.com/lf-edge/ekuiper/v2/pkg/model"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+type MockDecoder struct {
+	meta string
+}
+
+func (m *MockDecoder) Provision(_ api.StreamContext, props map[string]any) error {
+	_, ok := props["wrong"]
+	if ok {
+		return fmt.Errorf("wrong")
+	}
+	return nil
+}
+
+func (m *MockDecoder) ReadMeta(_ api.StreamContext, line []byte) {
+	if line != nil {
+		m.meta = string(line)
+		fmt.Println(m.meta)
+	}
+}
+
+func (m *MockDecoder) Decorate(ctx api.StreamContext, data any) any {
+	data.(map[string]any)["meta"] = m.meta
+	return data
+}
+
+func TestCSVDec(t *testing.T) {
+	path, err := os.Getwd()
+	if err != nil {
+		t.Fatal(err)
+	}
+	path = filepath.Join(path, "test", "csv")
+
+	modules.RegisterFileStreamReaderAlias("mockmock", "csv")
+	modules.RegisterFileStreamDecorator("mockmock", func(ctx api.StreamContext) modules.FileStreamDecorator {
+		return &MockDecoder{}
+	})
+
+	meta := map[string]any{
+		"file": filepath.Join(path, "a.csv"),
+	}
+	mc := timex.Clock
+	exp := []api.MessageTuple{
+		model.NewDefaultSourceTuple(map[string]any{"@": "#", "id": "1", "ts": "1670170500", "value": "161.927872", "meta": "<special content>"}, meta, mc.Now()),
+		model.NewDefaultSourceTuple(map[string]any{"@": "#", "id": "2", "ts": "1670170900", "value": "176", "meta": "<special content>"}, meta, mc.Now()),
+	}
+	r := GetSource()
+	mock.TestSourceConnector(t, r, map[string]any{
+		"path":             path,
+		"fileType":         "mockmock",
+		"datasource":       "a.csv",
+		"hasHeader":        true,
+		"delimiter":        "\t",
+		"ignoreStartLines": 3,
+		"ignoreEndLines":   1,
+	}, exp, func() {
+		// do nothing
+	})
+}

pkg/cast/time.go

@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -138,7 +138,18 @@ func ParseTime(t string, f string) (_ time.Time, err error) {
 		TimeFormats:  now.TimeFormats,
 	}
 	if f != "" {
-		c.TimeFormats = []string{f}
+		c.TimeFormats = append([]string{f}, c.TimeFormats...)
+	}
+	return c.Parse(t)
+}
+
+func ParseTimeByFormats(t string, formats []string) (_ time.Time, err error) {
+	c := &now.Config{
+		TimeLocation: localTimeZone,
+		TimeFormats:  now.TimeFormats,
+	}
+	if len(formats) > 0 {
+		c.TimeFormats = append(formats, c.TimeFormats...)
 	}
 	return c.Parse(t)
 }

pkg/cast/time_test.go

@@ -301,3 +301,15 @@ func TestConvertFormat(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, int64(1718024079657497900), d.UnixNano())
 }
+
+func TestParseTimeFormats(t *testing.T) {
+	err := SetTimeZone("UTC")
+	require.NoError(t, err)
+	tsstr := "2023-12-16 00:15"
+	tts, err := ParseTimeByFormats(tsstr, []string{"2006-01-02 15:04", "2006-01-02 15-04-05"})
+	require.NoError(t, err)
+
+	timeString := "2023-12-16T00:15:00Z"
+	tt, _ := time.Parse(time.RFC3339, timeString)
+	require.Equal(t, tt, tts)
+}

pkg/modules/file.go

@@ -63,9 +63,38 @@ func RegisterFileStreamReader(name string, provider FileStreamReaderProvider) {
 	fileStreamReaders[name] = provider
 }
 
+func RegisterFileStreamReaderAlias(alias string, ref string) {
+	fileStreamReaders[alias] = fileStreamReaders[ref]
+}
+
 func GetFileStreamReader(ctx api.StreamContext, name string) (FileStreamReader, bool) {
 	if p, ok := fileStreamReaders[name]; ok {
 		return p(ctx), true
 	}
 	return nil, false
 }
+
+type FileStreamDecorator interface {
+	// Provision Set up the static properties
+	Provision(ctx api.StreamContext, props map[string]any) error
+	// ReadMeta Read the metadata from the file source, and save in the decorator itself
+	// It will receive lines, when receiving EOF, there will be no more lines.
+	ReadMeta(ctx api.StreamContext, line []byte)
+	// Decorate the file source
+	Decorate(ctx api.StreamContext, data any) any
+}
+
+type FileStreamDecoratorProvider func(ctx api.StreamContext) FileStreamDecorator
+
+var fileStreamDecorators = map[string]FileStreamDecoratorProvider{}
+
+func RegisterFileStreamDecorator(name string, provider FileStreamDecoratorProvider) {
+	fileStreamDecorators[name] = provider
+}
+
+func GetFileStreamDecorator(ctx api.StreamContext, name string) (FileStreamDecorator, bool) {
+	if p, ok := fileStreamDecorators[name]; ok {
+		return p(ctx), true
+	}
+	return nil, false
+}