Mailer: Make a queue

Author: zeripath

modules/setting/queue.go

@@ -182,6 +182,17 @@ func NewQueueService() {
 	if _, ok := issueIndexerSectionMap["CONN_STR"]; !ok {
 		section.Key("CONN_STR").SetValue(Indexer.IssueQueueConnStr)
 	}
+
+	hasLength := false
+	for _, key := range Cfg.Section("queue.mail").Keys() {
+		if key.Name() == "LENGTH" {
+			hasLength = true
+			break
+		}
+	}
+	if !hasLength {
+		Cfg.Section("queue.mail").Key("LENGTH").SetValue(fmt.Sprintf("%d", Cfg.Section("mailer").Key("SEND_BUFFER_LEN").MustInt(100)))
+	}
 }
 
 // ParseQueueConnStr parses a queue connection string

services/mailer/mailer.go

@@ -18,7 +18,9 @@ import (
 	"time"
 
 	"code.gitea.io/gitea/modules/base"
+	"code.gitea.io/gitea/modules/graceful"
 	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/queue"
 	"code.gitea.io/gitea/modules/setting"
 
 	"github.com/jaytaylor/html2text"
@@ -257,18 +259,7 @@ func (s *dummySender) Send(from string, to []string, msg io.WriterTo) error {
 	return nil
 }
 
-func processMailQueue() {
-	for msg := range mailQueue {
-		log.Trace("New e-mail sending request %s: %s", msg.GetHeader("To"), msg.Info)
-		if err := gomail.Send(Sender, msg.Message); err != nil {
-			log.Error("Failed to send emails %s: %s - %v", msg.GetHeader("To"), msg.Info, err)
-		} else {
-			log.Trace("E-mails sent %s: %s", msg.GetHeader("To"), msg.Info)
-		}
-	}
-}
-
-var mailQueue chan *Message
+var mailQueue queue.Queue
 
 // Sender sender for sending mail synchronously
 var Sender gomail.Sender
@@ -291,24 +282,33 @@ func NewContext() {
 		Sender = &dummySender{}
 	}
 
-	mailQueue = make(chan *Message, setting.MailService.QueueLength)
+	mailQueue = setting.CreateQueue("mail", func(data ...queue.Data) {
+		for _, datum := range data {
+			msg := datum.(*Message)
+			log.Trace("New e-mail sending request %s: %s", msg.GetHeader("To"), msg.Info)
+			if err := gomail.Send(Sender, msg.Message); err != nil {
+				log.Error("Failed to send emails %s: %s - %v", msg.GetHeader("To"), msg.Info, err)
+			} else {
+				log.Trace("E-mails sent %s: %s", msg.GetHeader("To"), msg.Info)
+			}
+		}
+	}, &Message{})
 
-	// FIXME: graceful: Needs to become a queue and graceful
-	go processMailQueue()
+	go graceful.GetManager().RunWithShutdownFns(mailQueue.Run)
 }
 
 // SendAsync send mail asynchronously
 func SendAsync(msg *Message) {
 	go func() {
-		mailQueue <- msg
+		_ = mailQueue.Push(msg)
 	}()
 }
 
 // SendAsyncs send mails asynchronously
 func SendAsyncs(msgs []*Message) {
 	go func() {
 		for _, msg := range msgs {
-			mailQueue <- msg
+			_ = mailQueue.Push(msg)
 		}
 	}()
 }