added ability to configure queue size

config/config.go

@@ -76,6 +76,7 @@ type ServerConfig struct {
 	StunPort              int    `yaml:"stun_port"`
 	StunHost              string `yaml:"stun_host"`
 	Proxy                 string `yaml:"proxy"`
+	QueueSize             int    `yaml:"queue_size"`
 }
 
 // SQLConfig - Generic SQL Config

queue/queue.go

@@ -7,6 +7,7 @@ import (
 	"github.com/enriquebris/goconcurrentqueue"
 	"github.com/gravitl/netmaker/logger"
 	"github.com/gravitl/netmaker/models"
+	"github.com/gravitl/netmaker/servercfg"
 )
 
 // EventQueue - responsible for queueing and handling events sent to the server
@@ -14,8 +15,8 @@ var EventQueue goconcurrentqueue.Queue
 
 // StartQueue - starts the queue and listens for messages
 func StartQueue(ctx context.Context) {
+	initQueue()
 
-	EventQueue = goconcurrentqueue.NewFIFO()
 	go func(ctx context.Context) {
 		logger.Log(2, "initialized queue service!")
 		for {
@@ -36,3 +37,13 @@ func StartQueue(ctx context.Context) {
 		}
 	}(ctx)
 }
+
+// == private ==
+func initQueue() {
+	size := servercfg.GetQueueSize()
+	if size > 0 {
+		EventQueue = goconcurrentqueue.NewFixedFIFO(size)
+	} else {
+		EventQueue = goconcurrentqueue.NewFIFO()
+	}
+}

servercfg/serverconf.go

@@ -79,6 +79,7 @@ func GetServerConfig() config.ServerConfig {
 	if Is_EE {
 		cfg.IsEE = "yes"
 	}
+	cfg.QueueSize = 0
 
 	return cfg
 }
@@ -651,3 +652,17 @@ func IsProxyEnabled() bool {
 	}
 	return enabled
 }
+
+// GetQueueSize - retrieves the queue size if specified or 0
+func GetQueueSize() int {
+	size := 0
+	if os.Getenv("QUEUE_SIZE") != "" {
+		s, err := strconv.Atoi(os.Getenv("QUEUE_SIZE"))
+		if err == nil {
+			size = s
+		}
+	} else if config.Config.Server.QueueSize > 0 {
+		size = config.Config.Server.QueueSize
+	}
+	return size
+}