lib/protoparser/datadog: extract stream parsing code into a separate stream package

This is a follow-up for 057698f7fb

app/vmagent/datadog/request_handler.go

@@ -9,6 +9,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadog"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadog/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -28,7 +29,7 @@ func InsertHandlerForHTTP(at *auth.Token, req *http.Request) error {
 		return err
 	}
 	ce := req.Header.Get("Content-Encoding")
-	return parser.ParseStream(req.Body, ce, func(series []parser.Series) error {
+	return stream.Parse(req.Body, ce, func(series []parser.Series) error {
 		return insertRows(at, series, extraLabels)
 	})
 }

app/vminsert/datadog/request_handler.go

@@ -9,6 +9,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadog"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadog/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
 	"github.com/VictoriaMetrics/metrics"
@@ -29,7 +30,7 @@ func InsertHandlerForHTTP(at *auth.Token, req *http.Request) error {
 		return err
 	}
 	ce := req.Header.Get("Content-Encoding")
-	return parser.ParseStream(req.Body, ce, func(series []parser.Series) error {
+	return stream.Parse(req.Body, ce, func(series []parser.Series) error {
 		return insertRows(at, series, extraLabels)
 	})
 }

lib/protoparser/datadog/stream/streamparser.go

@@ -1,4 +1,4 @@
-package datadog
+package stream
 
 import (
 	"bufio"
@@ -13,6 +13,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadog"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -33,7 +34,7 @@ var (
 // ParseStream parses DataDog POST request for /api/v1/series from reader and calls callback for the parsed request.
 //
 // callback shouldn't hold series after returning.
-func ParseStream(r io.Reader, contentEncoding string, callback func(series []Series) error) error {
+func Parse(r io.Reader, contentEncoding string, callback func(series []datadog.Series) error) error {
 	wcr := writeconcurrencylimiter.GetReader(r)
 	defer writeconcurrencylimiter.PutReader(wcr)
 	r = wcr
@@ -143,15 +144,15 @@ func putPushCtx(ctx *pushCtx) {
 var pushCtxPool sync.Pool
 var pushCtxPoolCh = make(chan *pushCtx, cgroup.AvailableCPUs())
 
-func getRequest() *Request {
+func getRequest() *datadog.Request {
 	v := requestPool.Get()
 	if v == nil {
-		return &Request{}
+		return &datadog.Request{}
 	}
-	return v.(*Request)
+	return v.(*datadog.Request)
 }
 
-func putRequest(req *Request) {
+func putRequest(req *datadog.Request) {
 	requestPool.Put(req)
 }
 

lib/protoparser/datadog/stream/streamparser_test.go

@@ -1,4 +1,4 @@
-package datadog
+package stream
 
 import (
 	"testing"