feature: support ParseJsonStreams

errors.go

@@ -58,7 +58,9 @@ func NewJSONError(str []byte, pos int, msg string) *JSONError {
 	if end > len(str) {
 		end = len(str)
 	}
-	substr := append(str[start:pos], '^')
-	substr = append(substr, str[pos:end]...)
+	substr := make([]byte, end-start+1)
+	copy(substr, str[start:pos])
+	substr[pos-start] = '^'
+	copy(substr[pos-start+1:], str[pos+1:end])
 	return &JSONError{pos: pos, substr: string(substr), msg: msg}
 }

jsonutils.go

@@ -22,6 +22,7 @@ import (
 	"time"
 	"unicode/utf8"
 
+	"yunion.io/x/log"
 	"yunion.io/x/pkg/errors"
 	"yunion.io/x/pkg/gotypes"
 	"yunion.io/x/pkg/sortedmap"
@@ -678,3 +679,32 @@ func ParseStream(str []byte, offset int) (JSONObject, int, error) {
 		return nil, i, NewJSONError(str, i, "Empty string")
 	}
 }
+
+func ParseJsonStreams(stream []byte) ([]JSONObject, error) {
+	ret := make([]JSONObject, 0)
+	errs := make([]error, 0)
+	offset := 0
+	for offset < len(stream) {
+		for offset < len(stream) && stream[offset] != '[' && stream[offset] != '{' {
+			offset++
+		}
+		if offset >= len(stream) {
+			break
+		}
+		json, noffset, err := ParseStream(stream, offset)
+		if err != nil {
+			errs = append(errs, errors.Wrapf(err, "jsonutils.ParseStream fail at %d", offset))
+			offset++
+		} else {
+			ret = append(ret, json)
+			offset = noffset
+		}
+	}
+	if len(errs) > 0 && len(ret) == 0 {
+		return nil, errors.NewAggregate(errs)
+	}
+	if len(errs) > 0 {
+		log.Warningf("jsonutils.ParseJsonStreams: %d errors, %s", len(errs), errors.NewAggregate(errs))
+	}
+	return ret, nil
+}

jsonutils_test.go

@@ -190,3 +190,56 @@ func TestParseParams(t *testing.T) {
 		t.Fatalf("%s!=%s", str, json.String())
 	}
 }
+
+func TestParseJsonStreams(t *testing.T) {
+	cases := []struct {
+		jsonStr []byte
+		jsonLen int
+	}{
+		{
+			jsonStr: []byte(`{"abc": 12, "def": [1,2,"123",4.43], "ghi": "hahahah"} {"abc": 12}`),
+			jsonLen: 2,
+		},
+		{
+			jsonStr: []byte(`    {"abc": 12, "def": [1,2,"123",4.43], "ghi": "hahahah"} {"abc": 12}         `),
+			jsonLen: 2,
+		},
+		{
+			jsonStr: []byte(`    {"abc": 12, "def": [1,2,"123",4.43], "ghi": "hahahah"} [] {"abc": 12}         `),
+			jsonLen: 3,
+		},
+		{
+			jsonStr: []byte(`    {"abc": 12, "def": [1,2,"123",4.43], "ghi": "hahahah"} [] {"abc": 12}     {}   [] `),
+			jsonLen: 5,
+		},
+		{
+			jsonStr: []byte(`  abc  {"abc": 12, "def": [1,2,"123",4.43], "ghi": "hahahah"} [] {"abc": 12}     {}   [] `),
+			jsonLen: 5,
+		},
+		{
+			jsonStr: []byte(`  abc ["abc"]  {"abc": 12,
+			 "def": [1,2,"123",4.43], "ghi": "hahahah"}
+			 [
+			 ] {"abc": 12}
+			   {}   [] `),
+			jsonLen: 6,
+		},
+		{
+			jsonStr: []byte(`  abc  [{"abc": 12, "def": [1,2,"123",4.43], "ghi": "hahahah"} ["abc"] {"abc": 12}     {}   [] `),
+			jsonLen: 5,
+		},
+		{
+			jsonStr: []byte(`  abc   {"abc": 12, "def": [1,2,"123",4.43], "ghi": "hahahah"} ["abc"] {"abc": 12}     {}   [] `),
+			jsonLen: 5,
+		},
+	}
+	for _, c := range cases {
+		results, err := ParseJsonStreams(c.jsonStr)
+		if err != nil {
+			t.Errorf("%s", err)
+		} else if len(results) != c.jsonLen {
+			t.Errorf("got %d != expect %d", len(results), c.jsonLen)
+		}
+		t.Logf("%s", Marshal(results))
+	}
+}