refactoring http ammo

refactoring http ammo

components/providers/base/ammo.go

@@ -1,47 +1,110 @@
 package base
 
-import "github.com/yandex/pandora/core/aggregator/netsample"
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	urlpkg "net/url"
 
-type Ammo[R any] struct {
-	Req       *R
-	tag       string
-	id        uint64
-	isInvalid bool
+	"github.com/yandex/pandora/components/providers/http/util"
+	"github.com/yandex/pandora/core/aggregator/netsample"
+	"github.com/yandex/pandora/lib/netutil"
+)
+
+func NewAmmo(method string, url string, body []byte, header http.Header, tag string) (*Ammo, error) {
+	if ok := netutil.ValidHTTPMethod(method); !ok {
+		return nil, errors.New("invalid HTTP method " + method)
+	}
+	if _, err := urlpkg.Parse(url); err != nil {
+		return nil, fmt.Errorf("invalid URL %s; err %w ", url, err)
+	}
+	return &Ammo{
+		method:      method,
+		body:        body,
+		url:         url,
+		tag:         tag,
+		header:      header,
+		constructor: true,
+	}, nil
+}
+
+type Ammo struct {
+	Req         *http.Request
+	method      string
+	body        []byte
+	url         string
+	tag         string
+	header      http.Header
+	id          uint64
+	isInvalid   bool
+	constructor bool
 }
 
-func (a *Ammo[R]) Request() (*R, *netsample.Sample) {
+func (a *Ammo) Request() (*http.Request, *netsample.Sample) {
+	if a.Req == nil {
+		_ = a.BuildRequest() // TODO: what if error. There isn't a logger
+	}
 	sample := netsample.Acquire(a.Tag())
 	sample.SetID(a.ID())
 	return a.Req, sample
 }
 
-func (a *Ammo[R]) Reset(req *R, tag string) {
-	a.Req = req
-	a.tag = tag
-	a.id = 0
-	a.isInvalid = false
-}
-
-func (a *Ammo[_]) SetID(id uint64) {
+func (a *Ammo) SetID(id uint64) {
 	a.id = id
 }
 
-func (a *Ammo[_]) ID() uint64 {
+func (a *Ammo) ID() uint64 {
 	return a.id
 }
 
-func (a *Ammo[_]) Invalidate() {
+func (a *Ammo) Invalidate() {
 	a.isInvalid = true
 }
 
-func (a *Ammo[_]) IsInvalid() bool {
+func (a *Ammo) IsInvalid() bool {
 	return a.isInvalid
 }
 
-func (a *Ammo[_]) IsValid() bool {
+func (a *Ammo) IsValid() bool {
 	return !a.isInvalid
 }
 
-func (a *Ammo[_]) Tag() string {
+func (a *Ammo) SetTag(tag string) {
+	a.tag = tag
+}
+
+func (a *Ammo) Tag() string {
 	return a.tag
 }
+
+func (a *Ammo) FromConstructor() bool {
+	return a.constructor
+}
+
+// use NewAmmo() for skipping error here
+func (a *Ammo) BuildRequest() error {
+	var buff io.Reader
+	if a.body != nil {
+		buff = bytes.NewReader(a.body)
+	}
+	req, err := http.NewRequest(a.method, a.url, buff)
+	if err != nil {
+		return fmt.Errorf("cant create request: %w", err)
+	}
+	a.Req = req
+	util.EnrichRequestWithHeaders(req, a.header)
+	return nil
+}
+
+func (a *Ammo) Reset() {
+	a.Req = nil
+	a.method = ""
+	a.body = nil
+	a.url = ""
+	a.tag = ""
+	a.header = nil
+	a.id = 0
+	a.isInvalid = false
+}

components/providers/base/provider.go

@@ -8,7 +8,7 @@ import (
 )
 
 type ProviderBase struct {
-	core.ProviderDeps
+	Deps      core.ProviderDeps
 	FS        afero.Fs
 	idCounter atomic.Uint64
 }

components/providers/http/ammo.go

@@ -16,4 +16,4 @@ type Request interface {
 	http.Request
 }
 
-var _ phttp.Ammo = (*base.Ammo[http.Request])(nil)
+var _ phttp.Ammo = (*base.Ammo)(nil)

components/providers/http/decoders/decoder.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"net/http"
 
+	"github.com/yandex/pandora/components/providers/base"
 	"github.com/yandex/pandora/components/providers/http/config"
 	"github.com/yandex/pandora/components/providers/http/util"
 )
@@ -24,7 +25,7 @@ var (
 
 type Decoder interface {
 	// Decode(context.Context, chan<- *base.Ammo[http.Request], io.ReadSeeker) error
-	Scan(context.Context) (*http.Request, string, error)
+	Scan(context.Context) (*base.Ammo, error)
 }
 
 type protoDecoder struct {

components/providers/http/decoders/jsonline.go

@@ -7,9 +7,9 @@ import (
 	"net/http"
 	"strings"
 
+	"github.com/yandex/pandora/components/providers/base"
 	"github.com/yandex/pandora/components/providers/http/config"
 	"github.com/yandex/pandora/components/providers/http/decoders/jsonline"
-	"github.com/yandex/pandora/components/providers/http/util"
 	"golang.org/x/xerrors"
 )
 
@@ -35,53 +35,51 @@ type jsonlineDecoder struct {
 	line    uint
 }
 
-func (d *jsonlineDecoder) Scan(ctx context.Context) (*http.Request, string, error) {
+func (d *jsonlineDecoder) Scan(ctx context.Context) (*base.Ammo, error) {
 	if d.config.Limit != 0 && d.ammoNum >= d.config.Limit {
-		return nil, "", ErrAmmoLimit
+		return nil, ErrAmmoLimit
 	}
-	for ; ; d.line++ {
-		if ctx.Err() != nil {
-			return nil, "", ctx.Err()
+	for {
+		if d.config.Passes != 0 && d.passNum >= d.config.Passes {
+			return nil, ErrPassLimit
 		}
 
-		if !d.scanner.Scan() {
-			if d.scanner.Err() == nil { // assume as io.EOF; FIXME: check possible nil error with other reason
-				d.line = 0
-				d.passNum++
-				if d.config.Passes != 0 && d.passNum >= d.config.Passes {
-					return nil, "", ErrPassLimit
-				}
-				if d.ammoNum == 0 {
-					return nil, "", ErrNoAmmo
-				}
-				_, err := d.file.Seek(0, io.SeekStart)
-				if err != nil {
-					return nil, "", err
-				}
-				d.scanner = bufio.NewScanner(d.file)
-				if d.config.MaxAmmoSize != 0 {
-					var buffer []byte
-					d.scanner.Buffer(buffer, d.config.MaxAmmoSize)
-				}
+		for d.scanner.Scan() {
+			d.line++
+			data := d.scanner.Bytes()
+			if len(strings.TrimSpace(string(data))) == 0 {
 				continue
 			}
-			return nil, "", d.scanner.Err()
+			d.ammoNum++
+			ammo, err := jsonline.DecodeAmmo(data, d.decodedConfigHeaders)
+			if err != nil {
+				if !d.config.ContinueOnError {
+					return nil, xerrors.Errorf("failed to decode ammo at line: %v; data: %q, with err: %w", d.line+1, data, err)
+				}
+				// TODO: add log message about error
+				continue // skipping ammo
+			}
+			return ammo, err
 		}
-		data := d.scanner.Bytes()
-		if len(strings.TrimSpace(string(data))) == 0 {
-			continue
+
+		err := d.scanner.Err()
+		if err != nil {
+			return nil, err
+		}
+		if d.ammoNum == 0 {
+			return nil, ErrNoAmmo
 		}
-		d.ammoNum++
+		d.line = 0
+		d.passNum++
 
-		req, tag, err := jsonline.DecodeAmmo(data)
+		_, err = d.file.Seek(0, io.SeekStart)
 		if err != nil {
-			if !d.config.ContinueOnError {
-				return nil, "", xerrors.Errorf("failed to decode ammo at line: %v; data: %q, with err: %w", d.line+1, data, err)
-			}
-			// TODO: add log message about error
-			continue // skipping ammo
+			return nil, err
+		}
+		d.scanner = bufio.NewScanner(d.file)
+		if d.config.MaxAmmoSize != 0 {
+			var buffer []byte
+			d.scanner.Buffer(buffer, d.config.MaxAmmoSize)
 		}
-		util.EnrichRequestWithHeaders(req, d.decodedConfigHeaders)
-		return req, tag, err
 	}
 }

components/providers/http/decoders/jsonline/data.go

@@ -1,12 +1,12 @@
-//go:generate github.com/pquerna/ffjson data_ffjson.go
+//go:generate github.com/pquerna/ffjson@latest data_ffjson.go
 
 package jsonline
 
 import (
 	"net/http"
-	"strings"
 
 	"github.com/pkg/errors"
+	"github.com/yandex/pandora/components/providers/base"
 )
 
 // ffjson: noencoder
@@ -24,24 +24,21 @@ type data struct {
 	Body string `json:"body"`
 }
 
-func (d *data) ToRequest() (*http.Request, error) {
-	uri := "http://" + d.Host + d.URI
-	req, err := http.NewRequest(d.Method, uri, strings.NewReader(d.Body))
-	if err != nil {
-		return nil, errors.WithStack(err)
+func DecodeAmmo(jsonDoc []byte, baseHeader http.Header) (*base.Ammo, error) {
+	var d = new(data)
+	if err := d.UnmarshalJSON(jsonDoc); err != nil {
+		err = errors.WithStack(err)
+		return nil, err
 	}
+
+	header := baseHeader.Clone()
 	for k, v := range d.Headers {
-		req.Header.Set(k, v)
+		header.Set(k, v)
 	}
-	return req, err
-}
-
-func DecodeAmmo(jsonDoc []byte) (*http.Request, string, error) {
-	var data = new(data)
-	if err := data.UnmarshalJSON(jsonDoc); err != nil {
-		err = errors.WithStack(err)
-		return nil, data.Tag, err
+	url := "http://" + d.Host + d.URI
+	var body []byte
+	if d.Body != "" {
+		body = []byte(d.Body)
 	}
-	req, err := data.ToRequest()
-	return req, data.Tag, err
+	return base.NewAmmo(d.Method, url, body, header, d.Tag)
 }