nfflow/internal/controller/pipe/loadash/loadash.go

package loadash

import (
	"errors"
	"fmt"
	"github.com/dop251/goja"
	"github.com/loveuer/nfflow/internal/interfaces"
	"github.com/loveuer/nfflow/internal/sqlType"
	"github.com/sirupsen/logrus"
)

type PipeMap struct {
	vm *goja.Runtime
	FN string `json:"fn"`
}

func NewPipeMap(cfg sqlType.JSONB) (interfaces.Pipe, error) {
	var (
		err    error
		pm     = &PipeMap{}
		prefix = "let result; let doc; let myFn;"
	)

	if err = cfg.Bind(pm); err != nil {
		return nil, err
	}

	if pm.FN == "" {
		return nil, errors.New("nil cfg")
	}

	pm.vm = goja.New()
	if _, err = pm.vm.RunString(prefix); err != nil {
		logrus.Debugf("NewPipeMap: run prepare=%s err=%v", prefix, err)
		return nil, err
	}

	fn := "myFn = " + pm.FN + ";"
	if _, err = pm.vm.RunString(fn); err != nil {
		err = fmt.Errorf("PipeMap: vm run string=%s err=%v", fn, err)
		logrus.Error(err)
		return nil, err
	}

	return pm, nil
}

func (m *PipeMap) Start(data any) (any, error) {
	var (
		err error
	)

	if err = m.vm.Set("doc", map[string]any{"data": data}); err != nil {
		return nil, err
	}

	if _, err = m.vm.RunString(`result = myFn(doc.data)`); err != nil {
		return nil, err
	}

	value := m.vm.Get("result").Export()

	return value, nil
}

func (m *PipeMap) Close() {
	m.vm = nil
}