Move mailer to use a queue

modules/setting/queue.go

@@ -103,11 +103,11 @@ func NewQueueService() {
 
 	// Now handle the old issue_indexer configuration
 	section := Cfg.Section("queue.issue_indexer")
-	issueIndexerSectionMap := map[string]string{}
+	sectionMap := map[string]string{}
 	for _, key := range section.Keys() {
-		issueIndexerSectionMap[key.Name()] = key.Value()
+		sectionMap[key.Name()] = key.Value()
 	}
-	if _, ok := issueIndexerSectionMap["TYPE"]; !ok {
+	if _, ok := sectionMap["TYPE"]; !ok {
 		switch Indexer.IssueQueueType {
 		case LevelQueueType:
 			section.Key("TYPE").SetValue("level")
@@ -120,18 +120,28 @@ func NewQueueService() {
 				Indexer.IssueQueueType)
 		}
 	}
-	if _, ok := issueIndexerSectionMap["LENGTH"]; !ok {
+	if _, ok := sectionMap["LENGTH"]; !ok {
 		section.Key("LENGTH").SetValue(fmt.Sprintf("%d", Indexer.UpdateQueueLength))
 	}
-	if _, ok := issueIndexerSectionMap["BATCH_LENGTH"]; !ok {
+	if _, ok := sectionMap["BATCH_LENGTH"]; !ok {
 		section.Key("BATCH_LENGTH").SetValue(fmt.Sprintf("%d", Indexer.IssueQueueBatchNumber))
 	}
-	if _, ok := issueIndexerSectionMap["DATADIR"]; !ok {
+	if _, ok := sectionMap["DATADIR"]; !ok {
 		section.Key("DATADIR").SetValue(Indexer.IssueQueueDir)
 	}
-	if _, ok := issueIndexerSectionMap["CONN_STR"]; !ok {
+	if _, ok := sectionMap["CONN_STR"]; !ok {
 		section.Key("CONN_STR").SetValue(Indexer.IssueQueueConnStr)
 	}
+
+	// Handle the old mailer configuration
+	section = Cfg.Section("queue.mailer")
+	sectionMap = map[string]string{}
+	for _, key := range section.Keys() {
+		sectionMap[key.Name()] = key.Value()
+	}
+	if _, ok := sectionMap["LENGTH"]; !ok {
+		section.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,22 +282,33 @@ func NewContext() {
 		Sender = &dummySender{}
 	}
 
-	mailQueue = make(chan *Message, setting.MailService.QueueLength)
-	go processMailQueue()
+	mailQueue = queue.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{})
+
+	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)
 		}
 	}()
 }