Enable periodic cleaning of mqueue

openldap-log-parser/cmd/root.go

@@ -125,7 +125,7 @@ var (
 	File   os.File
 	Writer *bufio.Writer
 
-	Version = "0.6.6"
+	Version = "0.6.8"
 
 	BuildInfo = promauto.NewGaugeVec(prometheus.GaugeOpts{
 		Name: "openldaplogparser_build_info",
@@ -321,35 +321,31 @@ func writeOut(msg string, filename string) error {
 	return nil
 }
 
-// Every 24H, remove sent, milter-rejected and deferred that entered queue more than 5 days ago
+// Every 24H, remove connections closed more than 24 hours ago
-/*
+func periodicallyCleanMQueue(mqueue map[string]*OpenLdapConnection, mqMtx *sync.Mutex) {
-func periodicallyCleanMQueue(mqueue map[int]*PostfixLogParser, mqMtx *sync.Mutex) {
+	var ok bool
-	var ok int
 
 	for range time.Tick(time.Hour * 24) {
 		// Do we need read lock?
-		for _, inmail := range mqueue {
+		for _, ldcon := range mqueue {
-			ok = 0
+			ok = false
-			// Check all mails were sent (multiple destinations mails)
+			// Check if a close operation exist
-			//  or rejected
+			for _, op := range ldcon.Operations {
-			for _, outmail := range inmail.Messages {
+				if op.OpType == "close" {
-				if outmail.Status == "sent" || outmail.Status == "milter-reject" {
+					if op.Time.Add(time.Hour * 1 * 24).Before(time.Now()) {
-					ok += 1
+						ok = true
-				} else if outmail.Status == "deferred" {
-					if inmail.Time.Add(time.Hour * 5 * 24).Before(time.Now()) {
-						ok += 1
 					}
 				}
 			}
-			if ok == len(inmail.Messages) {
+			if ok == true {
 				mqMtx.Lock()
-				delete(mqueue, inmail.MessageId)
+				//delete(mqueue, ldcon.ConnId)
+				delete(mqueue, fmt.Sprintf("%s:%d", ldcon.Hostname, ldcon.ConnId))
 				mqMtx.Unlock()
 			}
 		}
 	}
 }
-*/
 
 func initConfig() {}
 
@@ -975,7 +971,7 @@ func processLogs(cmd *cobra.Command, args []string) {
 	}
 
 	// Cleaner thread
-	//go periodicallyCleanMQueue(mQueue, &mqMtx)
+	go periodicallyCleanMQueue(mQueue, &mqMtx)
 
 	// Initialize Stdin input...
 	if true == strings.EqualFold(gSyslogListenAddress, "do-not-listen") {