chore: Shift max queue size to general as it is global not receiver specific and remove redundant pending counts for non-ack connections

Author: driskell

lc-lib/config/general.go

@@ -30,10 +30,11 @@ import (
 var registeredGeneralCreators = make(map[string]SectionCreator)
 
 const (
-	defaultGeneralHost      string        = "localhost.localdomain"
-	defaultGeneralLogLevel  logging.Level = logging.INFO
-	defaultGeneralLogStdout bool          = true
-	defaultGeneralLogSyslog bool          = false
+	defaultGeneralHost         string        = "localhost.localdomain"
+	defaultGeneralLogLevel     logging.Level = logging.INFO
+	defaultGeneralLogStdout    bool          = true
+	defaultGeneralLogSyslog    bool          = false
+	defaultGeneralMaxQueueSize int64         = 128 * 1024 * 1024 // 128 MiB
 )
 
 // General holds the general configuration
@@ -44,6 +45,7 @@ type General struct {
 	LogLevel     logging.Level          `config:"log level"`
 	LogStdout    bool                   `config:"log stdout"`
 	LogSyslog    bool                   `config:"log syslog"`
+	MaxQueueSize int64                  `config:"max queue size"`
 
 	Custom map[string]interface{} `config:",embed_dynamic"`
 }
@@ -95,10 +97,11 @@ func RegisterGeneral(name string, creator SectionCreator) {
 func init() {
 	RegisterSection("general", func() interface{} {
 		c := &General{
-			LogLevel:  defaultGeneralLogLevel,
-			LogStdout: defaultGeneralLogStdout,
-			LogSyslog: defaultGeneralLogSyslog,
-			Custom:    make(map[string]interface{}),
+			LogLevel:     defaultGeneralLogLevel,
+			LogStdout:    defaultGeneralLogStdout,
+			LogSyslog:    defaultGeneralLogSyslog,
+			MaxQueueSize: defaultGeneralMaxQueueSize,
+			Custom:       make(map[string]interface{}),
 		}
 
 		for k, creator := range registeredGeneralCreators {

lc-lib/receiver/pool.go

@@ -61,6 +61,7 @@ type Pool struct {
 	spool             []*spoolEntry
 	spoolSize         int64
 
+	generalConfig  *config.General
 	apiConfig      *admin.Config
 	apiConnections api.Array
 	apiListeners   api.Array
@@ -72,6 +73,7 @@ type Pool struct {
 func NewPool(app *core.App) *Pool {
 	return &Pool{
 		apiConfig:        admin.FetchConfig(app.Config()),
+		generalConfig:    app.Config().General(),
 		ackChan:          make(chan []*event.Event),
 		eventChan:        make(chan transports.Event),
 		scheduler:        scheduler.NewScheduler(),
@@ -184,7 +186,7 @@ ReceiverLoop:
 				receiver := eventImpl.Context().Value(transports.ContextReceiver).(transports.Receiver)
 				r.startIdleTimeout(eventImpl.Context(), receiver, connection)
 				r.connectionLock.Lock()
-				connectionStatus := newPoolConnectionStatus(r, r.receivers[receiver].config.Name, r.receivers[receiver].listen, eventImpl.Remote(), eventImpl.Desc())
+				connectionStatus := newPoolConnectionStatus(r, r.receivers[receiver].config.Name, r.receivers[receiver].listen, eventImpl.Remote(), eventImpl.Desc(), receiver.SupportsAck())
 				r.connectionStatus[connection] = connectionStatus
 				r.connectionLock.Unlock()
 				// ReplaceEntry outside of connection lock, as entry updates that hold the entry lock will take the connection lock
@@ -196,7 +198,7 @@ ReceiverLoop:
 				connection := eventImpl.Context().Value(transports.ContextConnection)
 				receiver := eventImpl.Context().Value(transports.ContextReceiver).(transports.Receiver)
 				connectionStatus := r.connectionStatus[connection]
-				if r.spoolSize+int64(size) > r.receivers[receiver].config.MaxQueueSize {
+				if r.spoolSize+int64(size) > r.generalConfig.MaxQueueSize {
 					receiver.ShutdownConnectionRead(eventImpl.Context(), fmt.Errorf("max queue size exceeded"))
 					r.connectionLock.Unlock()
 					break
@@ -418,8 +420,10 @@ func (r *Pool) updateReceivers(newConfig *config.Config) {
 				newReceivers[newReceiversByListen[listen]] = &poolReceiverStatus{config: cfgEntry, listen: listen, active: true}
 				receiverApi := &api.KeyValue{}
 				receiverApi.SetEntry("listen", api.String(listen))
-				receiverApi.SetEntry("maxPendingPayloads", api.Number(cfgEntry.MaxPendingPayloads))
-				receiverApi.SetEntry("maxQueueSize", api.Number(cfgEntry.MaxQueueSize))
+				if newReceiversByListen[listen].SupportsAck() {
+					// Only applies if supports ack
+					receiverApi.SetEntry("maxPendingPayloads", api.Number(cfgEntry.MaxPendingPayloads))
+				}
 				r.apiListeners.AddEntry(listen, receiverApi)
 			}
 		}

lc-lib/receiver/status.go

@@ -41,11 +41,12 @@ type poolConnectionStatus struct {
 	progress         []*poolEventProgress
 	lines            int64
 	bytes            int
+	supportsAck      bool
 
 	api.KeyValue
 }
 
-func newPoolConnectionStatus(p *Pool, name string, listener string, remote string, desc string) *poolConnectionStatus {
+func newPoolConnectionStatus(p *Pool, name string, listener string, remote string, desc string, supportsAck bool) *poolConnectionStatus {
 	return &poolConnectionStatus{
 		p:        p,
 		name:     name,
@@ -58,7 +59,8 @@ func newPoolConnectionStatus(p *Pool, name string, listener string, remote strin
 			"remote":   remote,
 			"desc":     desc,
 		},
-		progress: make([]*poolEventProgress, 0),
+		progress:    make([]*poolEventProgress, 0),
+		supportsAck: supportsAck,
 	}
 }
 

lc-lib/transports/common.go

@@ -33,7 +33,6 @@ const (
 	defaultNetworkBackoff            time.Duration = 5 * time.Second
 	defaultNetworkBackoffMax         time.Duration = 300 * time.Second
 	defaultNetworkMaxPendingPayloads int64         = 10
-	defaultNetworkMaxQueueSize       int64         = 128 * 1024 * 1024 // 128 MiB
 	defaultNetworkMethod             string        = "random"
 	defaultNetworkRfc2782Service     string        = "courier"
 	defaultNetworkRfc2782Srv         bool          = true

lc-lib/transports/configreceiver.go

@@ -38,7 +38,6 @@ type ReceiverConfigEntry struct {
 	Transport          string   `config:"transport"`
 	Listen             []string `config:"listen"`
 	MaxPendingPayloads int64    `config:"max pending payloads"`
-	MaxQueueSize       int64    `config:"max queue size"`
 
 	Unused map[string]interface{} `json:",omitempty"`
 }
@@ -48,7 +47,6 @@ func (c *ReceiverConfigEntry) Defaults() {
 	c.Enabled = true
 	c.Transport = defaultReceiverTransport
 	c.MaxPendingPayloads = defaultNetworkMaxPendingPayloads
-	c.MaxQueueSize = defaultNetworkMaxQueueSize
 }
 
 // Init the receiver configuration