Queues: Move Notification to use a Queue

integrations/integration_test.go

@@ -213,6 +213,7 @@ func initIntegrationTest() {
 		defer db.Close()
 	}
 	routers.GlobalInit(graceful.GetManager().HammerContext())
+	NotifierListenerInit()
 }
 
 func prepareTestEnv(t testing.TB, skip ...int) func() {

integrations/mssql.ini.tmpl

@@ -82,3 +82,10 @@ LEVEL                = Debug
 INSTALL_LOCK   = true
 SECRET_KEY     = 9pCviYTWSb
 INTERNAL_TOKEN = eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJuYmYiOjE0OTU1NTE2MTh9.hhSVGOANkaKk3vfCd2jDOIww4pUk0xtg9JRde5UogyQ
+
+[queue]
+TYPE=channel
+
+[queue.test-notifier]
+BATCH_LENGTH=1
+LENGTH=20

integrations/mysql.ini.tmpl

@@ -84,3 +84,10 @@ LEVEL                = Debug
 INSTALL_LOCK   = true
 SECRET_KEY     = 9pCviYTWSb
 INTERNAL_TOKEN = eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJuYmYiOjE0OTU1NTE2MTh9.hhSVGOANkaKk3vfCd2jDOIww4pUk0xtg9JRde5UogyQ
+
+[queue]
+TYPE=channel
+
+[queue.test-notifier]
+BATCH_LENGTH=1
+LENGTH=20

integrations/mysql8.ini.tmpl

@@ -80,3 +80,10 @@ LEVEL                = Debug
 INSTALL_LOCK   = true
 SECRET_KEY     = 9pCviYTWSb
 INTERNAL_TOKEN = eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJuYmYiOjE0OTU1NTE2MTh9.hhSVGOANkaKk3vfCd2jDOIww4pUk0xtg9JRde5UogyQ
+
+[queue]
+TYPE=channel
+
+[queue.test-notifier]
+BATCH_LENGTH=1
+LENGTH=20

integrations/notification_helper_test.go

@@ -0,0 +1,117 @@
+// Copyright 2020 The Gitea Authors. All rights reserved.
+// Use of this source code is governed by a MIT-style
+// license that can be found in the LICENSE file.
+
+package integrations
+
+import (
+	"encoding/json"
+	"reflect"
+	"sync"
+	"testing"
+
+	"code.gitea.io/gitea/models"
+	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/notification"
+	"code.gitea.io/gitea/modules/notification/base"
+	"code.gitea.io/gitea/modules/queue"
+)
+
+var notifierListener *NotifierListener
+
+var once = sync.Once{}
+
+type NotifierListener struct {
+	lock      sync.RWMutex
+	callbacks map[string][]*func(string, [][]byte)
+	notifier  base.Notifier
+}
+
+func NotifierListenerInit() {
+	once.Do(func() {
+		notifierListener = &NotifierListener{
+			callbacks: map[string][]*func(string, [][]byte){},
+		}
+		notifierListener.notifier = base.NewQueueNotifierWithHandle("test-notifier", notifierListener.handle)
+		notification.RegisterNotifier(notifierListener.notifier)
+	})
+}
+
+// Register will register a callback with the provided notifier function
+func (n *NotifierListener) Register(functionName string, callback *func(string, [][]byte)) {
+	n.lock.Lock()
+	n.callbacks[functionName] = append(n.callbacks[functionName], callback)
+	n.lock.Unlock()
+}
+
+// Deregister will remove the provided callback from the provided notifier function
+func (n *NotifierListener) Deregister(functionName string, callback *func(string, [][]byte)) {
+	n.lock.Lock()
+	defer n.lock.Unlock()
+	for i, callbackPtr := range n.callbacks[functionName] {
+		if callbackPtr == callback {
+			n.callbacks[functionName] = append(n.callbacks[functionName][0:i], n.callbacks[functionName][i+1:]...)
+			return
+		}
+	}
+}
+
+// RegisterChannel will return a registered channel with function name and return a function to deregister it and close the channel at the end
+func (n *NotifierListener) RegisterChannel(name string, argNumber int, exemplar interface{}) (<-chan interface{}, func()) {
+	t := reflect.TypeOf(exemplar)
+	channel := make(chan interface{}, 10)
+	callback := func(_ string, args [][]byte) {
+		n := reflect.New(t).Elem()
+		err := json.Unmarshal(args[argNumber], n.Addr().Interface())
+		if err != nil {
+			log.Error("Wrong Argument passed to register channel: %v ", err)
+		}
+		channel <- n.Interface()
+	}
+	n.Register(name, &callback)
+
+	return channel, func() {
+		n.Deregister(name, &callback)
+		close(channel)
+	}
+}
+
+func (n *NotifierListener) handle(data ...queue.Data) {
+	n.lock.RLock()
+	defer n.lock.RUnlock()
+	for _, datum := range data {
+		call := datum.(*base.FunctionCall)
+		callbacks, ok := n.callbacks[call.Name]
+		if ok && len(callbacks) > 0 {
+			for _, callback := range callbacks {
+				(*callback)(call.Name, call.Args)
+			}
+		}
+	}
+}
+
+func TestNotifierListener(t *testing.T) {
+	defer prepareTestEnv(t)()
+
+	createPullNotified, deregister := notifierListener.RegisterChannel("NotifyNewPullRequest", 0, &models.PullRequest{})
+
+	bs, _ := json.Marshal(&models.PullRequest{})
+	notifierListener.handle(&base.FunctionCall{
+		Name: "NotifyNewPullRequest",
+		Args: [][]byte{
+			bs,
+		},
+	})
+	<-createPullNotified
+
+	notifierListener.notifier.NotifyNewPullRequest(&models.PullRequest{})
+	<-createPullNotified
+
+	notification.NotifyNewPullRequest(&models.PullRequest{})
+	<-createPullNotified
+
+	deregister()
+
+	notification.NotifyNewPullRequest(&models.PullRequest{})
+	// would panic if not deregistered
+}

integrations/pgsql.ini.tmpl

@@ -83,3 +83,10 @@ LEVEL                = Debug
 INSTALL_LOCK   = true
 SECRET_KEY     = 9pCviYTWSb
 INTERNAL_TOKEN = eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJuYmYiOjE0OTU1NTE2MTh9.hhSVGOANkaKk3vfCd2jDOIww4pUk0xtg9JRde5UogyQ
+
+[queue]
+TYPE=channel
+
+[queue.test-notifier]
+BATCH_LENGTH=1
+LENGTH=20

integrations/pull_merge_test.go

@@ -59,33 +59,67 @@ func testPullCleanUp(t *testing.T, session *TestSession, user, repo, pullnum str
 	return resp
 }
 
+func checkChannelWithTimeout(c <-chan interface{}, timeout time.Duration, callback func(interface{}) bool) bool {
+	timer := time.NewTimer(500 * time.Millisecond)
+	for {
+		select {
+		case received := <-c:
+			if callback(received) {
+				if !timer.Stop() {
+					select {
+					case <-timer.C:
+					default:
+					}
+				}
+				return true
+			}
+		case <-timer.C:
+			return false
+		}
+	}
+}
+
 func TestPullMerge(t *testing.T) {
 	onGiteaRun(t, func(t *testing.T, giteaURL *url.URL) {
-		hookTasks, err := models.HookTasks(1, 1) //Retrieve previous hook number
-		assert.NoError(t, err)
-		hookTasksLenBefore := len(hookTasks)
+		createPullNotified, unregisterNewPull := notifierListener.RegisterChannel("NotifyNewPullRequest", 0, &models.PullRequest{})
+		defer unregisterNewPull()
+
+		mergePullNotified, unregisterMergePull := notifierListener.RegisterChannel("NotifyMergePullRequest", 0, &models.PullRequest{})
+		defer unregisterMergePull()
 
 		session := loginUser(t, "user1")
-		testRepoFork(t, session, "user2", "repo1", "user1", "repo1")
-		testEditFile(t, session, "user1", "repo1", "master", "README.md", "Hello, World (Edited)\n")
+		testRepoFork(t, session, "user2", "repo1", "user1", "repo1-TestPullMerge")
+		testEditFile(t, session, "user1", "repo1-TestPullMerge", "master", "README.md", "Hello, World (Edited)\n")
 
-		resp := testPullCreate(t, session, "user1", "repo1", "master", "This is a pull title")
+		resp := testPullCreate(t, session, "user1", "repo1-TestPullMerge", "master", "This is a pull title")
+
+		isOurPR := func(received interface{}) bool {
+			pr := received.(*models.PullRequest)
+			pr.LoadBaseRepo()
+			pr.LoadHeadRepo()
+			return pr.BaseRepo.FullName() == "user2/repo1" &&
+				pr.BaseBranch == "master" &&
+				pr.HeadRepo.FullName() == "user1/repo1-TestPullMerge" &&
+				pr.HeadBranch == "master"
+		}
+
+		assert.True(t, checkChannelWithTimeout(createPullNotified, 500*time.Millisecond, isOurPR), "Failed to be notified pull created")
 
 		elem := strings.Split(test.RedirectURL(resp), "/")
 		assert.EqualValues(t, "pulls", elem[3])
+
 		testPullMerge(t, session, elem[1], elem[2], elem[4], models.MergeStyleMerge)
+		assert.True(t, checkChannelWithTimeout(mergePullNotified, 500*time.Millisecond, isOurPR), "Failed to be notified pull merged")
 
-		hookTasks, err = models.HookTasks(1, 1)
-		assert.NoError(t, err)
-		assert.Len(t, hookTasks, hookTasksLenBefore+1)
+		assert.False(t, checkChannelWithTimeout(createPullNotified, 100*time.Millisecond, isOurPR), "Duplicate notified pull created")
+		assert.False(t, checkChannelWithTimeout(mergePullNotified, 100*time.Millisecond, isOurPR), "Duplicate notified pull merged")
 	})
 }
 
 func TestPullRebase(t *testing.T) {
 	onGiteaRun(t, func(t *testing.T, giteaURL *url.URL) {
-		hookTasks, err := models.HookTasks(1, 1) //Retrieve previous hook number
-		assert.NoError(t, err)
-		hookTasksLenBefore := len(hookTasks)
+		mergePullNotified, deferable := notifierListener.RegisterChannel("NotifyMergePullRequest", 0, &models.PullRequest{})
+		defer deferable()
 
 		session := loginUser(t, "user1")
 		testRepoFork(t, session, "user2", "repo1", "user1", "repo1")
@@ -96,18 +130,18 @@ func TestPullRebase(t *testing.T) {
 		elem := strings.Split(test.RedirectURL(resp), "/")
 		assert.EqualValues(t, "pulls", elem[3])
 		testPullMerge(t, session, elem[1], elem[2], elem[4], models.MergeStyleRebase)
-
-		hookTasks, err = models.HookTasks(1, 1)
-		assert.NoError(t, err)
-		assert.Len(t, hookTasks, hookTasksLenBefore+1)
+		select {
+		case <-mergePullNotified:
+		case <-time.After(500 * time.Millisecond):
+			assert.Fail(t, "Took too long to notify!")
+		}
 	})
 }
 
 func TestPullRebaseMerge(t *testing.T) {
 	onGiteaRun(t, func(t *testing.T, giteaURL *url.URL) {
-		hookTasks, err := models.HookTasks(1, 1) //Retrieve previous hook number
-		assert.NoError(t, err)
-		hookTasksLenBefore := len(hookTasks)
+		mergePullNotified, deferable := notifierListener.RegisterChannel("NotifyMergePullRequest", 0, &models.PullRequest{})
+		defer deferable()
 
 		session := loginUser(t, "user1")
 		testRepoFork(t, session, "user2", "repo1", "user1", "repo1")
@@ -119,17 +153,18 @@ func TestPullRebaseMerge(t *testing.T) {
 		assert.EqualValues(t, "pulls", elem[3])
 		testPullMerge(t, session, elem[1], elem[2], elem[4], models.MergeStyleRebaseMerge)
 
-		hookTasks, err = models.HookTasks(1, 1)
-		assert.NoError(t, err)
-		assert.Len(t, hookTasks, hookTasksLenBefore+1)
+		select {
+		case <-mergePullNotified:
+		case <-time.After(500 * time.Millisecond):
+			assert.Fail(t, "Took too long to notify!")
+		}
 	})
 }
 
 func TestPullSquash(t *testing.T) {
 	onGiteaRun(t, func(t *testing.T, giteaURL *url.URL) {
-		hookTasks, err := models.HookTasks(1, 1) //Retrieve previous hook number
-		assert.NoError(t, err)
-		hookTasksLenBefore := len(hookTasks)
+		mergePullNotified, deferable := notifierListener.RegisterChannel("NotifyMergePullRequest", 0, &models.PullRequest{})
+		defer deferable()
 
 		session := loginUser(t, "user1")
 		testRepoFork(t, session, "user2", "repo1", "user1", "repo1")
@@ -142,9 +177,11 @@ func TestPullSquash(t *testing.T) {
 		assert.EqualValues(t, "pulls", elem[3])
 		testPullMerge(t, session, elem[1], elem[2], elem[4], models.MergeStyleSquash)
 
-		hookTasks, err = models.HookTasks(1, 1)
-		assert.NoError(t, err)
-		assert.Len(t, hookTasks, hookTasksLenBefore+1)
+		select {
+		case <-mergePullNotified:
+		case <-time.After(500 * time.Millisecond):
+			assert.Fail(t, "Took too long to notify!")
+		}
 	})
 }
 

integrations/sqlite.ini.tmpl

@@ -80,3 +80,11 @@ INTERNAL_TOKEN = eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJuYmYiOjE0OTI3OTU5ODN9.O
 
 [oauth2]
 JWT_SECRET = KZb_QLUd4fYVyxetjxC4eZkrBgWM2SndOOWDNtgUUko
+
+[queue]
+TYPE=channel
+
+[queue.test-notifier]
+BATCH_LENGTH=1
+LENGTH=20
+

models/issue.go

@@ -47,7 +47,7 @@ type Issue struct {
 	IsClosed         bool         `xorm:"INDEX"`
 	IsRead           bool         `xorm:"-"`
 	IsPull           bool         `xorm:"INDEX"` // Indicates whether is a pull request or not.
-	PullRequest      *PullRequest `xorm:"-"`
+	PullRequest      *PullRequest `xorm:"-" json:"-"`
 	NumComments      int
 	Ref              string