Add json path operator

pkg/elem/encoding_json_path.go

@@ -0,0 +1,81 @@
+package elem
+
+import (
+	"github.com/Bitspark/slang/pkg/core"
+	"github.com/google/uuid"
+	"github.com/tidwall/gjson"
+)
+
+var encodingJSONPathId = uuid.MustParse("89571f57-4aad-4bb8-9d03-573343ff1202")
+var encodingJSONPathCfg = &builtinConfig{
+	blueprint: core.Blueprint{
+		Id: encodingJSONPathId,
+		Meta: core.BlueprintMetaDef{
+			Name:             "JSON Path",
+			ShortDescription: "select values based json path expression from a JSON document",
+			Icon:             "brackets-curly",
+			Tags:             []string{"json", "encoding"},
+			DocURL:           "https://bitspark.de/slang/docs/operator/json-path",
+		},
+		ServiceDefs: map[string]*core.ServiceDef{
+			core.MAIN_SERVICE: {
+				In: core.TypeDef{
+					Type: "binary",
+				},
+				Out: core.TypeDef{
+					Type: "map",
+					Map: map[string]*core.TypeDef{
+						"valid": {
+							Type: "boolean",
+						},
+						"{paths}": {
+							Type: "primitive",
+						},
+					},
+				},
+			},
+		},
+		DelegateDefs: map[string]*core.DelegateDef{},
+		PropertyDefs: map[string]*core.TypeDef{
+			"paths": {
+				Type: "stream",
+				Stream: &core.TypeDef{
+					Type: "string",
+				},
+			},
+		},
+	},
+	opFunc: func(op *core.Operator) {
+		in := op.Main().In()
+		out := op.Main().Out()
+		paths := op.Property("paths").([]interface{})
+		for !op.CheckStop() {
+			in := in.Pull()
+			valid := true
+			if core.IsMarker(in) {
+				out.Push(in)
+				continue
+			}
+
+			jsonDoc := []byte(in.(core.Binary))
+			if !gjson.ValidBytes(jsonDoc) {
+				for _, v := range paths {
+					out.Map(v.(string)).Push(nil)
+				}
+				valid = false
+			} else {
+				for _, v := range paths {
+					res := gjson.GetBytes(jsonDoc, v.(string))
+					if !res.Exists() {
+						out.Map(v.(string)).Push(nil)
+					}
+					if res.IsArray() || res.IsObject() {
+						out.Map(v.(string)).Push(res.Raw)
+					}
+					out.Map(v.(string)).Push(res.Value())
+				}
+			}
+			out.Map("valid").Push(valid)
+		}
+	},
+}

pkg/elem/encoding_json_path_test.go

@@ -0,0 +1,112 @@
+package elem
+
+import (
+	"testing"
+
+	"github.com/Bitspark/slang/pkg/core"
+	"github.com/Bitspark/slang/tests/assertions"
+	"github.com/stretchr/testify/require"
+)
+
+const jsonDoc = `
+{
+	"age":37,
+	"children": ["Sara","Alex","Jack"],
+	"fav.movie": "Deer Hunter",
+	"friends": [
+	  {"age": 44, "first": "Dale", "last": "Murphy"},
+	  {"age": 68, "first": "Roger", "last": "Craig"},
+	  {"age": 47, "first": "Jane", "last": "Fonder"}
+	],
+	"name": {"first": "Tom", "last": "Anderson"}
+  }
+`
+
+func Test_JsonPath__IsRegistered(t *testing.T) {
+	a := assertions.New(t)
+
+	ocFork := getBuiltinCfg(encodingJSONPathId)
+	a.NotNil(ocFork)
+}
+
+func Test_JsonPath__String(t *testing.T) {
+	a := assertions.New(t)
+
+	o, err := buildOperator(
+		core.InstanceDef{
+			Operator: encodingJSONPathId,
+			Properties: map[string]interface{}{
+				"paths": []interface{}{"name.last"},
+			},
+		},
+	)
+	require.NoError(t, err)
+
+	o.Main().Out().Bufferize()
+	o.Start()
+
+	o.Main().In().Push(core.Binary(jsonDoc))
+	a.PortPushes("Anderson", o.Main().Out().Map("name.last"))
+	a.PortPushes(true, o.Main().Out().Map("valid"))
+}
+
+func Test_JsonPath__Invalid_Document(t *testing.T) {
+	a := assertions.New(t)
+
+	o, err := buildOperator(
+		core.InstanceDef{
+			Operator: encodingJSONPathId,
+			Properties: map[string]interface{}{
+				"paths": []interface{}{"name.last"},
+			},
+		},
+	)
+	require.NoError(t, err)
+
+	o.Main().Out().Bufferize()
+	o.Start()
+	o.Main().In().Push(core.Binary(`{"test"`))
+	a.PortPushes(nil, o.Main().Out().Map("name.last"))
+	a.PortPushes(false, o.Main().Out().Map("valid"))
+}
+
+func Test_JsonPath__NonExistent_Path(t *testing.T) {
+	a := assertions.New(t)
+
+	o, err := buildOperator(
+		core.InstanceDef{
+			Operator: encodingJSONPathId,
+			Properties: map[string]interface{}{
+				"paths": []interface{}{"name.missing", "name.last"},
+			},
+		},
+	)
+	require.NoError(t, err)
+
+	o.Main().Out().Bufferize()
+	o.Start()
+	o.Main().In().Push(core.Binary(jsonDoc))
+	a.PortPushes("Anderson", o.Main().Out().Map("name.last"))
+	a.PortPushes(nil, o.Main().Out().Map("name.missing"))
+	a.PortPushes(true, o.Main().Out().Map("valid"))
+}
+
+func Test_JsonPath__Non_Primitive_Return(t *testing.T) {
+	a := assertions.New(t)
+
+	o, err := buildOperator(
+		core.InstanceDef{
+			Operator: encodingJSONPathId,
+			Properties: map[string]interface{}{
+				"paths": []interface{}{"friends.#.last"},
+			},
+		},
+	)
+	require.NoError(t, err)
+
+	o.Main().Out().Bufferize()
+	o.Start()
+	o.Main().In().Push(core.Binary(jsonDoc))
+	a.PortPushes(`["Murphy","Craig","Fonder"]`, o.Main().Out().Map("friends.#.last"))
+	a.PortPushes(true, o.Main().Out().Map("valid"))
+}

pkg/elem/manager.go

@@ -121,6 +121,7 @@ func init() {
 	Register(encodingCSVWriteCfg)
 	Register(encodingJSONReadCfg)
 	Register(encodingJSONWriteCfg)
+	Register(encodingJSONPathCfg)
 	Register(encodingXLSXReadCfg)
 	Register(encodingURLWriteCfg)