Add javascript processor

go.mod

@@ -48,6 +48,8 @@ require (
 	github.com/dgraph-io/ristretto v0.1.0
 	github.com/docker/cli v20.10.12+incompatible // indirect
 	github.com/docker/docker v20.10.12+incompatible // indirect
+	github.com/dop251/goja v0.0.0-20220815083517-0c74f9139fd6
+	github.com/dop251/goja_nodejs v0.0.0-20220808115320-bac29516aae9
 	github.com/dustin/go-humanize v1.0.0
 	github.com/eclipse/paho.mqtt.golang v1.3.5
 	github.com/fatih/color v1.13.0

go.sum

@@ -362,6 +362,9 @@ github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2/go.mod h1:SqUrOPUn
 github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
 github.com/dimfeld/httptreemux v5.0.1+incompatible h1:Qj3gVcDNoOthBAqftuD596rm4wg/adLLz5xh5CmpiCA=
 github.com/dimfeld/httptreemux v5.0.1+incompatible/go.mod h1:rbUlSV+CCpv/SuqUTP/8Bk2O3LyUV436/yaRGkhP6Z0=
+github.com/dlclark/regexp2 v1.4.1-0.20201116162257-a2a8dda75c91/go.mod h1:2pZnwuY/m+8K6iRw6wQdMtk+rH5tNGR1i55kozfMjCc=
+github.com/dlclark/regexp2 v1.7.0 h1:7lJfhqlPssTb1WQx4yvTHN0uElPEv52sbaECrAQxjAo=
+github.com/dlclark/regexp2 v1.7.0/go.mod h1:DHkYz0B9wPfa6wondMfaivmHpzrQ3v9q8cnmRbL6yW8=
 github.com/dnaeon/go-vcr v1.1.0 h1:ReYa/UBrRyQdant9B4fNHGoCNKw6qh6P0fsdGmZpR7c=
 github.com/dnaeon/go-vcr v1.1.0/go.mod h1:M7tiix8f0r6mKKJ3Yq/kqU1OYf3MnfmBWVbPx/yU9ko=
 github.com/docker/cli v20.10.11+incompatible/go.mod h1:JLrzqnKDaYBop7H2jaqPtU4hHvMKP+vjCwu2uszcLI8=
@@ -374,6 +377,13 @@ github.com/docker/go-connections v0.4.0 h1:El9xVISelRB7BuFusrZozjnkIM5YnzCViNKoh
 github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5XhDvyHbTtUxmeec=
 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/dop251/goja v0.0.0-20211022113120-dc8c55024d06/go.mod h1:R9ET47fwRVRPZnOGvHxxhuZcbrMCuiqOz3Rlrh4KSnk=
+github.com/dop251/goja v0.0.0-20220815083517-0c74f9139fd6 h1:xHdUVG+c8SWJnct16Z3QJOVlaYo3OwoJyamo6kR6OL0=
+github.com/dop251/goja v0.0.0-20220815083517-0c74f9139fd6/go.mod h1:yRkwfj0CBpOGre+TwBsqPV0IH0Pk73e4PXJOeNDboGs=
+github.com/dop251/goja_nodejs v0.0.0-20210225215109-d91c329300e7/go.mod h1:hn7BA7c8pLvoGndExHudxTDKZ84Pyvv+90pbBjbTz0Y=
+github.com/dop251/goja_nodejs v0.0.0-20211022123610-8dd9abb0616d/go.mod h1:DngW8aVqWbuLRMHItjPUyqdj+HWPvnQe8V8y1nDpIbM=
+github.com/dop251/goja_nodejs v0.0.0-20220808115320-bac29516aae9 h1:7nszERfxMR5Gyw+M21EbrZZyTtVbRhNdRmtW/Vr3hzc=
+github.com/dop251/goja_nodejs v0.0.0-20220808115320-bac29516aae9/go.mod h1:DngW8aVqWbuLRMHItjPUyqdj+HWPvnQe8V8y1nDpIbM=
 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/dvsekhvalnov/jose2go v0.0.0-20200901110807-248326c1351b/go.mod h1:7BvyPhdbLxMXIYTFPLsyJRFMsKmOZnQmzh6Gb+uquuM=
@@ -448,6 +458,8 @@ github.com/go-ole/go-ole v1.2.4/go.mod h1:XCwSNxSkXRo4vlyPy93sltvi/qJq0jqQhjqQNI
 github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=
 github.com/go-redis/redis/v7 v7.4.1 h1:PASvf36gyUpr2zdOUS/9Zqc80GbM+9BDyiJSJDDOrTI=
 github.com/go-redis/redis/v7 v7.4.1/go.mod h1:JDNMw23GTyLNC4GZu9njt15ctBQVn7xjRfnwdHj/Dcg=
+github.com/go-sourcemap/sourcemap v2.1.3+incompatible h1:W1iEw64niKVGogNgBN3ePyLFfuisuzeidWPMPWmECqU=
+github.com/go-sourcemap/sourcemap v2.1.3+incompatible/go.mod h1:F8jJfvm2KbVjc5NqelyYJmf/v5J0dwNLS2mL4sNA1Jg=
 github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
 github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
 github.com/go-sql-driver/mysql v1.6.0 h1:BCTh4TKNUYmOmMUcQ3IipzF5prigylS7XXjEkfCHuOE=
@@ -715,8 +727,9 @@ github.com/kr/fs v0.1.0/go.mod h1:FFnZGqtBN9Gxj7eW1uZ42v5BccTP0vu6NEaFoC2HwRg=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
-github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI=
 github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0=
+github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
@@ -942,6 +955,7 @@ github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzG
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
+github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
 github.com/rogpeppe/go-internal v1.8.0 h1:FCbCCtXNOY3UtUuHUYaghJg4y7Fd14rXifAYUAtL9R8=
 github.com/rogpeppe/go-internal v1.8.0/go.mod h1:WmiCO8CzOY8rg0OYDC4/i/2WRWAB6poM+XZ2dLUbcbE=
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=

internal/impl/javascript/casts.go

@@ -0,0 +1,44 @@
+package javascript
+
+import (
+	"errors"
+
+	"github.com/dop251/goja"
+)
+
+func getMapFromValue(val goja.Value) (map[string]interface{}, error) {
+	outVal := val.Export()
+	v, ok := outVal.(map[string]interface{})
+	if !ok {
+		return nil, errors.New("value is not of type map")
+	}
+	return v, nil
+}
+
+func getSliceFromValue(val goja.Value) ([]interface{}, error) {
+	outVal := val.Export()
+	v, ok := outVal.([]interface{})
+	if !ok {
+		return nil, errors.New("value is not of type slice")
+	}
+	return v, nil
+}
+
+func getMapSliceFromValue(val goja.Value) ([]map[string]interface{}, error) {
+	outVal := val.Export()
+	if v, ok := outVal.([]map[string]interface{}); ok {
+		return v, nil
+	}
+	vSlice, ok := outVal.([]interface{})
+	if !ok {
+		return nil, errors.New("value is not of type map slice")
+	}
+	v := make([]map[string]interface{}, len(vSlice))
+	for i, e := range vSlice {
+		v[i], ok = e.(map[string]interface{})
+		if !ok {
+			return nil, errors.New("value is not of type map slice")
+		}
+	}
+	return v, nil
+}

internal/impl/javascript/logger.go

@@ -0,0 +1,19 @@
+package javascript
+
+import "github.com/benthosdev/benthos/v4/public/service"
+
+type Logger struct {
+	l *service.Logger
+}
+
+func (l *Logger) Log(message string) {
+	l.l.Info(message)
+}
+
+func (l *Logger) Warn(message string) {
+	l.l.Warn(message)
+}
+
+func (l *Logger) Error(message string) {
+	l.l.Error(message)
+}

internal/impl/javascript/processor_javascript.go

@@ -0,0 +1,101 @@
+package javascript
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+
+	"github.com/benthosdev/benthos/v4/public/service"
+	"github.com/dop251/goja"
+	"github.com/dop251/goja_nodejs/console"
+)
+
+func javascriptProcessorConfig() *service.ConfigSpec {
+	return service.NewConfigSpec().
+		Beta().
+		Categories("Mapping").
+		Summary("Executes the provided JavaScript code using the github.com/dop251/goja library. The `console` and `require` packages from /~https://github.com/dop251/goja_nodejs are also implementd.").
+		Field(service.NewInterpolatedStringField("code").
+			Description("The javascript code to use.").
+			Default("")).
+		Field(service.NewInterpolatedStringField("file").
+			Description("The javascript file to use.").
+			Default(""))
+}
+
+func init() {
+	err := service.RegisterProcessor(
+		"javascript", javascriptProcessorConfig(),
+		func(conf *service.ParsedConfig, mgr *service.Resources) (service.Processor, error) {
+			return newJavascriptProcessorFromConfig(conf, mgr)
+		})
+
+	if err != nil {
+		panic(err)
+	}
+}
+
+//------------------------------------------------------------------------------
+
+type javascriptProcessor struct {
+	program *goja.Program
+	logger  *service.Logger
+}
+
+func newJavascriptProcessorFromConfig(conf *service.ParsedConfig, mgr *service.Resources) (*javascriptProcessor, error) {
+	code, err := conf.FieldString("code")
+	if err != nil {
+		return nil, err
+	}
+	file, err := conf.FieldString("file")
+	if err != nil {
+		return nil, err
+	}
+
+	if code != "" && file != "" {
+		return nil, errors.New("both 'code' and 'file' fields are specified but only one is allowed")
+	}
+	if code == "" && file == "" {
+		return nil, errors.New("neither 'code' nor 'file' fields are specified but one of them is required")
+	}
+
+	filename := "main.js"
+	var program *goja.Program
+	if file != "" {
+		// Open file and read code
+		codeBytes, err := os.ReadFile(file)
+		if err != nil {
+			return nil, fmt.Errorf("failed to open the file specifed in 'file' field: %v", err)
+		}
+		filename = file
+		code = string(codeBytes)
+	}
+
+	program, err = goja.Compile(filename, code, false)
+	if err != nil {
+		return nil, fmt.Errorf("failed to compile javascript code: %v", err)
+	}
+
+	logger := mgr.Logger()
+
+	requireRegistry.RegisterNativeModule("console", console.RequireWithPrinter(&Logger{logger}))
+
+	return &javascriptProcessor{program: program, logger: logger}, nil
+}
+
+func (j *javascriptProcessor) Process(ctx context.Context, m *service.Message) (service.MessageBatch, error) {
+	// Create new JS VM
+	vm := getVM(m, j.logger)
+
+	// Run JS file
+	_, err := vm.RunProgram(j.program)
+	if err != nil {
+		return nil, err
+	}
+	return []*service.Message{m}, nil
+}
+
+func (j *javascriptProcessor) Close(ctx context.Context) error {
+	return nil
+}