fix: Cache fastjson.Value

Author: jachym-tousek-keboola

internal/pkg/service/stream/mapping/recordctx/fasthttp.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/keboola/go-utils/pkg/orderedmap"
 	"github.com/valyala/fasthttp"
+	"github.com/valyala/fastjson"
 
 	"github.com/keboola/keboola-as-code/internal/pkg/utils/errors"
 )
@@ -27,6 +28,7 @@ type fastHTTPContext struct {
 	bodyStringErr error
 	bodyMap       *orderedmap.OrderedMap
 	bodyMapErr    error
+	jsonValue     *fastjson.Value
 }
 
 func FromFastHTTP(ctx context.Context, timestamp time.Time, req *fasthttp.RequestCtx) Context {
@@ -122,6 +124,28 @@ func (c *fastHTTPContext) BodyMap() (*orderedmap.OrderedMap, error) {
 	return c.bodyMap, c.bodyMapErr
 }
 
+func (c *fastHTTPContext) JsonValue(parserPool *fastjson.ParserPool) (*fastjson.Value, error) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	if c.jsonValue != nil {
+		return c.jsonValue, nil
+	}
+
+	body, err := c.BodyBytes()
+	if err != nil {
+		return nil, err
+	}
+
+	parser := parserPool.Get()
+	defer parserPool.Put(parser)
+
+	jsonValue, err := parser.ParseBytes(body)
+	c.jsonValue = jsonValue
+
+	return jsonValue, err
+}
+
 func (c *fastHTTPContext) headersToMap() *orderedmap.OrderedMap {
 	out := orderedmap.New()
 	for _, k := range c.req.Request.Header.PeekKeys() {

internal/pkg/service/stream/mapping/recordctx/http.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/keboola/go-utils/pkg/orderedmap"
+	"github.com/valyala/fastjson"
 
 	"github.com/keboola/keboola-as-code/internal/pkg/utils/errors"
 	"github.com/keboola/keboola-as-code/internal/pkg/utils/ip"
@@ -29,6 +30,7 @@ type httpContext struct {
 	bodyBytesErr  error
 	bodyMap       *orderedmap.OrderedMap
 	bodyMapErr    error
+	jsonValue     *fastjson.Value
 }
 
 func FromHTTP(timestamp time.Time, req *http.Request) Context {
@@ -124,6 +126,28 @@ func (c *httpContext) BodyMap() (*orderedmap.OrderedMap, error) {
 	return c.bodyMap, c.bodyMapErr
 }
 
+func (c *httpContext) JsonValue(parserPool *fastjson.ParserPool) (*fastjson.Value, error) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	if c.jsonValue != nil {
+		return c.jsonValue, nil
+	}
+
+	body, err := c.BodyBytes()
+	if err != nil {
+		return nil, err
+	}
+
+	parser := parserPool.Get()
+	defer parserPool.Put(parser)
+
+	jsonValue, err := parser.ParseBytes(body)
+	c.jsonValue = jsonValue
+
+	return jsonValue, err
+}
+
 func (c *httpContext) bodyBytesWithoutLock() ([]byte, error) {
 	if c.bodyBytes == nil && c.bodyBytesErr == nil {
 		c.bodyBytes, c.bodyBytesErr = io.ReadAll(c.req.Body)

internal/pkg/service/stream/mapping/recordctx/recordctx.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/keboola/go-utils/pkg/orderedmap"
+	"github.com/valyala/fastjson"
 )
 
 type Context interface {
@@ -18,4 +19,5 @@ type Context interface {
 	BodyString() (string, error)
 	BodyBytes() ([]byte, error)
 	BodyMap() (*orderedmap.OrderedMap, error)
+	JsonValue(*fastjson.ParserPool) (*fastjson.Value, error)
 }

internal/pkg/service/stream/mapping/table/column/renderer.go

@@ -105,15 +105,7 @@ func (r *Renderer) mapPathCSVValue(c Path, ctx recordctx.Context) (string, error
 }
 
 func (r *Renderer) jsonPathCSVValue(c Path, ctx recordctx.Context) (string, error) {
-	body, err := ctx.BodyBytes()
-	if err != nil {
-		return "", err
-	}
-
-	parser := r.fastjsonPool.Get()
-	defer r.fastjsonPool.Put(parser)
-
-	value, err := parser.ParseBytes(body)
+	value, err := ctx.JsonValue(r.fastjsonPool)
 	if err != nil {
 		return "", err
 	}