Notification: queue ui.go notification-service

Author: zeripath

modules/notification/ui/ui.go

@@ -7,14 +7,17 @@ package ui
 import (
 	"code.gitea.io/gitea/models"
 	"code.gitea.io/gitea/modules/git"
+	"code.gitea.io/gitea/modules/graceful"
 	"code.gitea.io/gitea/modules/log"
 	"code.gitea.io/gitea/modules/notification/base"
+	"code.gitea.io/gitea/modules/queue"
+	"code.gitea.io/gitea/modules/setting"
 )
 
 type (
 	notificationService struct {
 		base.NullNotifier
-		issueQueue chan issueNotificationOpts
+		issueQueue queue.Queue
 	}
 
 	issueNotificationOpts struct {
@@ -30,19 +33,24 @@ var (
 
 // NewNotifier create a new notificationService notifier
 func NewNotifier() base.Notifier {
-	return &notificationService{
-		issueQueue: make(chan issueNotificationOpts, 100),
-	}
+	ns := &notificationService{}
+	ns.issueQueue = setting.CreateQueue("notification-service", ns.handle, issueNotificationOpts{})
+	return ns
 }
 
-func (ns *notificationService) Run() {
-	for opts := range ns.issueQueue {
+func (ns *notificationService) handle(data ...queue.Data) {
+	for _, datum := range data {
+		opts := datum.(issueNotificationOpts)
 		if err := models.CreateOrUpdateIssueNotifications(opts.issueID, opts.commentID, opts.notificationAuthorID); err != nil {
 			log.Error("Was unable to create issue notification: %v", err)
 		}
 	}
 }
 
+func (ns *notificationService) Run() {
+	graceful.GetManager().RunWithShutdownFns(ns.issueQueue.Run)
+}
+
 func (ns *notificationService) NotifyCreateIssueComment(doer *models.User, repo *models.Repository,
 	issue *models.Issue, comment *models.Comment) {
 	var opts = issueNotificationOpts{
@@ -52,35 +60,39 @@ func (ns *notificationService) NotifyCreateIssueComment(doer *models.User, repo
 	if comment != nil {
 		opts.commentID = comment.ID
 	}
-	ns.issueQueue <- opts
+	_ = ns.issueQueue.Push(opts)
 }
 
 func (ns *notificationService) NotifyNewIssue(issue *models.Issue) {
-	ns.issueQueue <- issueNotificationOpts{
+	_ = ns.issueQueue.Push(issueNotificationOpts{
 		issueID:              issue.ID,
 		notificationAuthorID: issue.Poster.ID,
-	}
+	})
 }
 
 func (ns *notificationService) NotifyIssueChangeStatus(doer *models.User, issue *models.Issue, actionComment *models.Comment, isClosed bool) {
 	ns.issueQueue <- issueNotificationOpts{
 		issueID:              issue.ID,
 		notificationAuthorID: doer.ID,
-	}
+	})
 }
 
 func (ns *notificationService) NotifyMergePullRequest(pr *models.PullRequest, doer *models.User, gitRepo *git.Repository) {
-	ns.issueQueue <- issueNotificationOpts{
+	_ = ns.issueQueue.Push(issueNotificationOpts{
 		issueID:              pr.Issue.ID,
 		notificationAuthorID: doer.ID,
-	}
+	})
 }
 
 func (ns *notificationService) NotifyNewPullRequest(pr *models.PullRequest) {
-	ns.issueQueue <- issueNotificationOpts{
+	if err := pr.LoadIssue(); err != nil {
+		log.Error("Unable to load issue: %d for pr: %d: Error: %v", pr.IssueID, pr.ID, err)
+		return
+	}
+	_ = ns.issueQueue.Push(issueNotificationOpts{
 		issueID:              pr.Issue.ID,
 		notificationAuthorID: pr.Issue.PosterID,
-	}
+	})
 }
 
 func (ns *notificationService) NotifyPullRequestReview(pr *models.PullRequest, r *models.Review, c *models.Comment) {
@@ -91,5 +103,5 @@ func (ns *notificationService) NotifyPullRequestReview(pr *models.PullRequest, r
 	if c != nil {
 		opts.commentID = c.ID
 	}
-	ns.issueQueue <- opts
+	_ = ns.issueQueue.Push(opts)
 }