Performance improvements

go.mod

@@ -6,7 +6,7 @@ require (
 	github.com/RackSec/srslog v0.0.0-20180709174129-a4725f04ec91
 	github.com/RoaringBitmap/roaring v1.5.0
 	github.com/asaskevich/EventBus v0.0.0-20200907212545-49d423059eef
-	github.com/g41797/go-syslog v1.0.6
+	github.com/g41797/go-syslog v1.0.7
 	github.com/g41797/kissngoqueue v0.1.5
 	github.com/g41797/sputnik v0.0.17
 )

go.sum

@@ -8,8 +8,8 @@ github.com/bits-and-blooms/bitset v1.2.0 h1:Kn4yilvwNtMACtf1eYDlG8H77R07mZSPbMjL
 github.com/bits-and-blooms/bitset v1.2.0/go.mod h1:gIdJ4wp64HaoK2YrL1Q5/N7Y16edYb8uY+O0FJTyyDA=
 github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/g41797/go-syslog v1.0.6 h1:OmTN5D/N7UX44mvScTbCNTpUCBtxr0gM9rs1rhbulGw=
-github.com/g41797/go-syslog v1.0.6/go.mod h1:UTVoy1HHCeTQrHszeHyQquF0hbqU8anfc2Lqr3ZgXkg=
+github.com/g41797/go-syslog v1.0.7 h1:ktpiFvBwrG5YFhlq5zG+0lZeEJYZAyZaoNeeXapzv7w=
+github.com/g41797/go-syslog v1.0.7/go.mod h1:UTVoy1HHCeTQrHszeHyQquF0hbqU8anfc2Lqr3ZgXkg=
 github.com/g41797/gonfig v1.0.1 h1:ywgkhF3SbNJuFN9Hlm9n7Lp+KxqeupkNmFObn2ZpWs8=
 github.com/g41797/gonfig v1.0.1/go.mod h1:5iL4oHFpYi+c7AE4D1dZnlBleJzM71dX6HZBhEELF90=
 github.com/g41797/kissngoqueue v0.1.5 h1:UrnpxbjOnTnKj/EqpbLA9LjhYkHjnLWMZQqsz3AOBG8=

server.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/g41797/go-syslog"
 	"github.com/g41797/go-syslog/format"
+	"github.com/g41797/kissngoqueue"
 	"github.com/g41797/sputnik"
 )
 
@@ -58,12 +59,14 @@ type server struct {
 	config  SyslogConfiguration
 	bc      atomic.Pointer[sputnik.BlockCommunicator]
 	syslogd *syslog.Server
+	q       *kissngoqueue.Queue[format.LogParts]
 }
 
 func newServer(conf SyslogConfiguration) *server {
 	srv := new(server)
 	srv.config = conf
 	srv.bc = atomic.Pointer[sputnik.BlockCommunicator]{}
+	srv.q = kissngoqueue.NewQueue[format.LogParts]()
 	return srv
 }
 
@@ -107,6 +110,8 @@ func (s *server) Init() error {
 		}
 	}
 
+	go s.processLogParts()
+
 	return nil
 }
 
@@ -115,8 +120,8 @@ func (s *server) Start() error {
 }
 
 func (s *server) Finish() error {
-	err := s.syslogd.Kill()
-	return err
+	s.q.Cancel()
+	return s.syslogd.Kill()
 }
 
 func (s *server) SetupHandling(bc sputnik.BlockCommunicator) {
@@ -132,9 +137,22 @@ func (s *server) Handle(logParts format.LogParts, msgLen int64, err error) {
 		return
 	}
 
-	msg := toMsg(logParts, msgLen, err)
+	s.q.PutMT(logParts)
+}
+
+func (s *server) startLogPartsProcessor() {
 
-	(*s.bc.Load()).Send(msg)
+}
+
+func (s *server) processLogParts() {
+	for {
+		lp, ok := s.q.Get()
+		if !ok {
+			break
+		}
+		(*s.bc.Load()).Send(toMsg(lp))
+	}
+	return
 }
 
 func (s *server) forHandle(logParts format.LogParts) bool {

syslogmsgparts.go

@@ -74,7 +74,7 @@ func newsyslogmsgparts() *syslogmsgparts {
 	return result
 }
 
-func (mp *syslogmsgparts) pack(logParts format.LogParts, err error) error {
+func (mp *syslogmsgparts) pack(logParts format.LogParts) error {
 
 	if mp == nil {
 		return fmt.Errorf("nil syslogmsgparts")
@@ -88,7 +88,7 @@ func (mp *syslogmsgparts) pack(logParts format.LogParts, err error) error {
 		return fmt.Errorf("empty logParts")
 	}
 
-	if err != nil {
+	if _, exists := logParts[Formermessage]; exists {
 		mp.packParts(formerMessage[:], logParts)
 		return nil
 	}
@@ -196,7 +196,7 @@ func toString(val any, typ string) string {
 	return result
 }
 
-func toMsg(logParts format.LogParts, msgLen int64, err error) sputnik.Msg {
+func toMsg(logParts format.LogParts) sputnik.Msg {
 
 	if logParts == nil {
 		return nil
@@ -210,7 +210,7 @@ func toMsg(logParts format.LogParts, msgLen int64, err error) sputnik.Msg {
 
 	slm, _ := msg[syslogmessage].(*syslogmsgparts)
 
-	perr := slm.pack(logParts, err)
+	perr := slm.pack(logParts)
 	if perr != nil {
 		return nil
 	}

syslogmsgparts_test.go

@@ -23,7 +23,7 @@ func Test_PackUnpackBadlyFormatted(t *testing.T) {
 
 	msgparts := new(syslogmsgparts)
 
-	err = msgparts.pack(logparts, fmt.Errorf("bad formatted message"))
+	err = msgparts.pack(logparts)
 	if err != nil {
 		t.Errorf("pack error %v", err)
 	}
@@ -57,7 +57,7 @@ func testPackUnpackRFCMsg(in map[string]string, descr []partType, t *testing.T)
 
 	msgparts := new(syslogmsgparts)
 
-	err = msgparts.pack(logparts, nil)
+	err = msgparts.pack(logparts)
 	if err != nil {
 		t.Errorf("pack error %v", err)
 	}