flow is working

cmd/enricher/main.go

@@ -22,6 +22,7 @@ import (
 	// import various transports
 	"github.com/netsampler/goflow2/v2/transport"
 	_ "github.com/netsampler/goflow2/v2/transport/file"
+	_ "github.com/netsampler/goflow2/v2/transport/http"
 	_ "github.com/netsampler/goflow2/v2/transport/kafka"
 
 	"github.com/oschwald/geoip2-golang"

cmd/goflow2/main.go

@@ -30,6 +30,7 @@ import (
 	"github.com/netsampler/goflow2/v2/transport"
 	_ "github.com/netsampler/goflow2/v2/transport/file"
 	_ "github.com/netsampler/goflow2/v2/transport/kafka"
+	_ "github.com/netsampler/goflow2/v2/transport/http"
 
 	// various producers
 	"github.com/netsampler/goflow2/v2/producer"

transport/http/http.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"encoding/json"
 	"flag"
+	"fmt"
 	"net/http"
 	"sync"
 
@@ -11,36 +12,71 @@ import (
 )
 
 type HTTPDriver struct {
-	httpDestination string
-	lock            *sync.RWMutex
-	q               chan bool
+	httpDestination     string
+	httpAuthHeader      string
+	httpAuthCredentials string
+	lock                *sync.RWMutex
+	q                   chan bool
+	batchSize           int
+	batchData           [][]byte
 }
 
 func (d *HTTPDriver) Prepare() error {
-	flag.StringVar(&d.httpDestination, "transport.http", "", "HTTP endpoint for output")
+	flag.StringVar(&d.httpDestination, "transport.http.destination", "", "HTTP endpoint for output")
+	flag.StringVar(&d.httpAuthHeader, "transport.http.auth.header", "", "HTTP header to set for credentials")
+	flag.StringVar(&d.httpAuthCredentials, "transport.http.auth.credentials", "", "credentials for the header")
+	flag.IntVar(&d.batchSize, "transport.http.batchSize", 1000, "Batch size for sending records")
+
+	if d.batchSize <= 0 {
+		d.batchSize = 1000 // default batch size
+	}
+
 	return nil
 }
 
 func (d *HTTPDriver) Init() error {
 	d.q = make(chan bool, 1)
+	d.batchData = make([][]byte, 0, d.batchSize)
 	return nil
 }
 
 func (d *HTTPDriver) Send(key, data []byte) error {
 	d.lock.RLock()
 	httpDestination := d.httpDestination
+	httpAuthHeader := d.httpAuthHeader
+	httpAuthCredentials := d.httpAuthCredentials
 	d.lock.RUnlock()
 
-	jsonData, err := json.Marshal(data)
-	if err != nil {
-		return err
-	}
+	d.batchData = append(d.batchData, data)
+	fmt.Println("batchData len:", len(d.batchData))
+	if len(d.batchData) >= d.batchSize {
+		jsonData, err := json.Marshal(d.batchData)
+		if err != nil {
+			return err
+		}
 
-	resp, err := http.Post(httpDestination, "application/json", bytes.NewBuffer(jsonData))
-	if err != nil {
-		return err
+		fmt.Println("jsonData:", string(jsonData))
+
+		req, err := http.NewRequest("POST", httpDestination, bytes.NewBuffer(jsonData))
+		if err != nil {
+			return err
+		}
+
+		req.Header.Set("Content-Type", "application/json")
+		fmt.Println("httpAuthHeader:", httpAuthHeader)
+		fmt.Println("httpAuthCredentials:", httpAuthCredentials)
+		req.Header.Set(httpAuthHeader, httpAuthCredentials)
+
+		client := &http.Client{}
+		resp, err := client.Do(req)
+		if err != nil {
+			return err
+		}
+		defer resp.Body.Close()
+
+		// reset batchData
+		d.batchData = d.batchData[:0]
 	}
-	defer resp.Body.Close()
 
 	return nil
 }