[FEATURE] buffered channel drain

cshared.go

@@ -6,6 +6,7 @@ package plugin
 import "C"
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -16,6 +17,7 @@ import (
 	"runtime"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 	"unsafe"
 
@@ -31,6 +33,7 @@ var (
 	unregister func()
 	cmt        *cmetrics.Context
 	logger     Logger
+	buflock    sync.Mutex
 )
 
 // FLBPluginRegister registers a plugin in the context of the fluent-bit runtime, a name and description
@@ -133,47 +136,79 @@ func FLBPluginInputCallback(data *unsafe.Pointer, csize *C.size_t) int {
 	var err error
 	once.Do(func() {
 		runCtx, runCancel = context.WithCancel(context.Background())
-		theChannel = make(chan Message)
+		// we need to configure this part....
+		theChannel = make(chan Message, 300000)
+		// do we need to buffer this part???
+		cbuf := make(chan Message, 16)
+
 		go func() {
-			err = theInput.Collect(runCtx, theChannel)
+			err = theInput.Collect(runCtx, cbuf)
 		}()
+		go func(cbuf chan Message) {
+			t := time.NewTicker(1 * time.Second)
+			for {
+				buflock.Lock()
+				select {
+				case msg, ok := <-cbuf:
+					if !ok {
+						continue
+					}
+					buflock.Unlock()
+					theChannel <- msg
+					buflock.Lock()
+				case <-t.C:
+					buflock.Unlock()
+					buflock.Lock()
+				}
+				buflock.Unlock()
+			}
+		}(cbuf)
 	})
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "run: %s\n", err)
 		return input.FLB_ERROR
 	}
 
-	select {
-	case msg, ok := <-theChannel:
-		if !ok {
-			return input.FLB_OK
-		}
+	buf := bytes.NewBuffer([]byte{})
+	buflock.Lock()
 
-		t := input.FLBTime{Time: msg.Time}
-		b, err := input.NewEncoder().Encode([]any{t, msg.Record})
-		if err != nil {
-			fmt.Fprintf(os.Stderr, "encode: %s\n", err)
-			return input.FLB_ERROR
+	for loop := len(theChannel) > 0; loop; {
+		select {
+		case msg, ok := <-theChannel:
+			if !ok {
+				return input.FLB_ERROR
+			}
+
+			t := input.FLBTime{Time: msg.Time}
+			b, err := input.NewEncoder().Encode([]any{t, msg.Record})
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "encode: %s\n", err)
+				return input.FLB_ERROR
+			}
+			buf.Grow(len(b))
+			buf.Write(b)
+		default:
+			loop = false
+		case <-runCtx.Done():
+			err := runCtx.Err()
+			if err != nil && !errors.Is(err, context.Canceled) {
+				fmt.Fprintf(os.Stderr, "run: %s\n", err)
+				return input.FLB_ERROR
+			}
+			// enforce a runtime gc, to prevent the thread finalizer on
+			// fluent-bit to kick in before any remaining data has not been GC'ed
+			// causing a sigsegv.
+			defer runtime.GC()
+			loop = false
 		}
+	}
+	buflock.Unlock()
 
+	if buf.Len() > 0 {
+		b := buf.Bytes()
 		cdata := C.CBytes(b)
-
 		*data = cdata
 		*csize = C.size_t(len(b))
-
-		// C.free(unsafe.Pointer(cdata))
-	case <-runCtx.Done():
-		err := runCtx.Err()
-		if err != nil && !errors.Is(err, context.Canceled) {
-			fmt.Fprintf(os.Stderr, "run: %s\n", err)
-			return input.FLB_ERROR
-		}
-		// enforce a runtime gc, to prevent the thread finalizer on
-		// fluent-bit to kick in before any remaining data has not been GC'ed
-		// causing a sigsegv.
-		defer runtime.GC()
-	default:
-		break
 	}
 
 	return input.FLB_OK