Change UpdateRepoIndex api to include watchers (#7012)

* Change UpdateRepoIndex api to include watchers

* Add timeout

integrations/repo_search_test.go

@@ -5,7 +5,6 @@
 package integrations
 
 import (
-	"log"
 	"net/http"
 	"testing"
 	"time"
@@ -34,22 +33,14 @@ func TestSearchRepo(t *testing.T) {
 	repo, err := models.GetRepositoryByOwnerAndName("user2", "repo1")
 	assert.NoError(t, err)
 
-	models.UpdateRepoIndexer(repo)
+	waiter := make(chan error, 1)
+	models.UpdateRepoIndexer(repo, waiter)
 
-	log.Printf("Waiting for indexing\n")
-
-	i := 0
-	for i < 60 {
-		if repo.IndexerStatus != nil && len(repo.IndexerStatus.CommitSha) != 0 {
-			break
-		}
-		time.Sleep(1 * time.Second)
-		i++
-	}
-	if i < 60 {
-		log.Printf("Indexing took: %ds\n", i)
-	} else {
-		log.Printf("Waited the limit: %ds for indexing, continuing\n", i)
+	select {
+	case err := <-waiter:
+		assert.NoError(t, err)
+	case <-time.After(1 * time.Minute):
+		assert.Fail(t, "UpdateRepoIndexer took too long")
 	}
 
 	req := NewRequestf(t, "GET", "/user2/repo1/search?q=Description&page=1")

models/repo_indexer.go

@@ -57,8 +57,9 @@ func (repo *Repository) updateIndexerStatus(sha string) error {
 }
 
 type repoIndexerOperation struct {
-	repo    *Repository
-	deleted bool
+	repo     *Repository
+	deleted  bool
+	watchers []chan<- error
 }
 
 var repoIndexerOperationQueue chan repoIndexerOperation
@@ -312,26 +313,30 @@ func nonGenesisChanges(repo *Repository, revision string) (*repoChanges, error)
 func processRepoIndexerOperationQueue() {
 	for {
 		op := <-repoIndexerOperationQueue
+		var err error
 		if op.deleted {
-			if err := indexer.DeleteRepoFromIndexer(op.repo.ID); err != nil {
+			if err = indexer.DeleteRepoFromIndexer(op.repo.ID); err != nil {
 				log.Error("DeleteRepoFromIndexer: %v", err)
 			}
 		} else {
-			if err := updateRepoIndexer(op.repo); err != nil {
+			if err = updateRepoIndexer(op.repo); err != nil {
 				log.Error("updateRepoIndexer: %v", err)
 			}
 		}
+		for _, watcher := range op.watchers {
+			watcher <- err
+		}
 	}
 }
 
 // DeleteRepoFromIndexer remove all of a repository's entries from the indexer
-func DeleteRepoFromIndexer(repo *Repository) {
-	addOperationToQueue(repoIndexerOperation{repo: repo, deleted: true})
+func DeleteRepoFromIndexer(repo *Repository, watchers ...chan<- error) {
+	addOperationToQueue(repoIndexerOperation{repo: repo, deleted: true, watchers: watchers})
 }
 
 // UpdateRepoIndexer update a repository's entries in the indexer
-func UpdateRepoIndexer(repo *Repository) {
-	addOperationToQueue(repoIndexerOperation{repo: repo, deleted: false})
+func UpdateRepoIndexer(repo *Repository, watchers ...chan<- error) {
+	addOperationToQueue(repoIndexerOperation{repo: repo, deleted: false, watchers: watchers})
 }
 
 func addOperationToQueue(op repoIndexerOperation) {