pipeline adds new input and output implementations

.gitignore

@@ -9,3 +9,5 @@ dist/
 *.exe
 *.out
 *.test
+memprofile.ou
+cpuprofile.ou

pipeline/ptinput/funcs/fn_adjust_timezone_test.go

@@ -66,38 +66,40 @@ func TestAdjustTimezone(t *testing.T) {
 				return
 			}
 			{
-				pt := ptinput.NewPlPoint(
+				plpt := ptinput.NewPlPoint(
 					point.Logging, "test", nil, map[string]any{"message": tc.in}, time.Now())
-				errR := runScript(runner, pt)
+				errR := runScript(runner, plpt)
 				if errR != nil {
 					t.Fatal(errR)
 				}
 
-				pt.KeyTime2Time()
+				plpt.KeyTime2Time()
+				pt := plpt.Point()
 				var v interface{}
 				if tc.outkey != "time" {
-					v, _, _ = pt.Get(tc.outkey)
+					v = pt.Get(tc.outkey)
 				} else {
-					v = pt.PtTime()
+					v = pt.Time()
 				}
 				assert.Equal(t, tc.expect, v)
 				t.Logf("[%d] PASS", idx)
 			}
 
 			{
-				pt := ptinput.NewPlPoint(point.Logging,
+				plpt := ptinput.NewPlPoint(point.Logging,
 					"test", nil, map[string]any{"message": tc.in}, time.Now())
-				errR := runScript(runner, pt)
+				errR := runScript(runner, plpt)
 				if errR != nil {
 					t.Fatal(errR)
 				}
 
-				pt.KeyTime2Time()
+				plpt.KeyTime2Time()
+				pt := plpt.Point()
 				var v interface{}
 				if tc.outkey != "time" {
-					v, _, _ = pt.Get(tc.outkey)
+					v = pt.Get(tc.outkey)
 				} else {
-					v = pt.PtTime()
+					v = pt.Time()
 				}
 				assert.Equal(t, tc.expect, v)
 				t.Logf("[%d] PASS", idx)

pipeline/ptinput/funcs/fn_default_time_test.go

@@ -1082,20 +1082,22 @@ func TestDefaultTime(t *testing.T) {
 				return
 			}
 
-			pt := ptinput.NewPlPoint(
+			plpt := ptinput.NewPlPoint(
 				point.Logging, "test", nil, map[string]any{"message": tc.in}, time.Now())
-			errR := runScript(runner, pt)
+			errR := runScript(runner, plpt)
 			if errR != nil {
 				t.Fatal(errR)
 			}
 
-			pt.KeyTime2Time()
+			plpt.KeyTime2Time()
+
+			pt := plpt.Point()
 
 			var v interface{}
 			if tc.outkey != "time" && tc.outkey != "" {
-				v, _, _ = pt.Get(tc.outkey)
+				v = pt.Get(tc.outkey)
 			} else {
-				v = pt.PtTime().UnixNano()
+				v = pt.Time().UnixNano()
 			}
 			tu.Equals(t, tc.expect, v)
 			t.Logf("[%d] PASS", idx)

pipeline/ptinput/funcs/fn_default_time_with_fmt_test.go

@@ -101,20 +101,20 @@ func TestDefaultTimeWithFmt(t *testing.T) {
 				return
 			}
 			for idxIn := 0; idxIn < len(tc.in); idxIn++ {
-				pt := ptinput.NewPlPoint(
+				plpt := ptinput.NewPlPoint(
 					point.Logging, "test", nil, map[string]any{"message": tc.in[idxIn]}, time.Now())
-				errR := runScript(runner, pt)
+				errR := runScript(runner, plpt)
 				if errR != nil {
 					t.Fatal(errR)
 				}
 
-				pt.KeyTime2Time()
-
+				plpt.KeyTime2Time()
+				pt := plpt.Point()
 				var v interface{}
-				if tc.outkey != "time" && tc.outkey != "" {
-					v, _, _ = pt.Get(tc.outkey)
+				if tc.outkey != "time" {
+					v = pt.Get(tc.outkey)
 				} else {
-					v = pt.PtTime().UnixNano()
+					v = pt.Time().UnixNano()
 				}
 				tu.Equals(t, tc.expect[idxIn], v)
 				t.Logf("[%d] PASS", idx)

pipeline/ptinput/funcs/fn_http_request.go

@@ -2,14 +2,15 @@ package funcs
 
 import (
 	"bytes"
-	"encoding/json"
 	"io"
 	"net"
 	"net/http"
 	"strconv"
 	"strings"
 	"time"
 
+	"github.com/goccy/go-json"
+
 	"github.com/GuanceCloud/platypus/pkg/ast"
 	"github.com/GuanceCloud/platypus/pkg/engine/runtime"
 	"github.com/GuanceCloud/platypus/pkg/errchain"

pipeline/ptinput/funcs/fn_http_request_test.go

@@ -1,7 +1,6 @@
 package funcs
 
 import (
-	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -10,6 +9,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/goccy/go-json"
+
 	"github.com/GuanceCloud/cliutils/pipeline/ptinput"
 	"github.com/GuanceCloud/cliutils/point"
 	"github.com/stretchr/testify/assert"

pipeline/ptinput/funcs/fn_parse_date_test.go

@@ -252,9 +252,9 @@ parse_date(key="time", yy=year, MM=month, dd=day, hh=hour, mm=min, ss=sec, zone=
 				return
 			}
 
-			pt := ptinput.NewPlPoint(
+			plpt := ptinput.NewPlPoint(
 				point.Logging, "test", nil, map[string]any{"message": tc.in}, time.Now())
-			errR := runScript(runner, pt)
+			errR := runScript(runner, plpt)
 
 			if errR != nil {
 				if tc.fail {
@@ -263,12 +263,13 @@ parse_date(key="time", yy=year, MM=month, dd=day, hh=hour, mm=min, ss=sec, zone=
 					t.Error(errR.Error())
 				}
 			} else {
-				pt.KeyTime2Time()
+				plpt.KeyTime2Time()
+				pt := plpt.Point()
 				var v interface{}
-				if tc.outKey != "time" && tc.outKey != "" {
-					v, _, _ = pt.Get(tc.outKey)
+				if tc.outKey != "time" {
+					v = pt.Get(tc.outKey)
 				} else {
-					v = pt.PtTime().UnixNano()
+					v = pt.Time().UnixNano()
 				}
 				tu.Equals(t, tc.expected, v)
 				t.Logf("[%d] PASS", idx)

pipeline/ptinput/funcs/fn_valid_json.go

@@ -6,9 +6,10 @@
 package funcs
 
 import (
-	"encoding/json"
 	"fmt"
 
+	"github.com/goccy/go-json"
+
 	"github.com/GuanceCloud/platypus/pkg/ast"
 	"github.com/GuanceCloud/platypus/pkg/engine/runtime"
 	"github.com/GuanceCloud/platypus/pkg/errchain"