From 54bd63cd5c5645202625368c185b645c7771f687 Mon Sep 17 00:00:00 2001
From: zeripath <art27@cantab.net>
Date: Thu, 23 May 2019 17:00:07 +0100
Subject: [PATCH] Change UpdateRepoIndex api to include watchers (#7012)
* Change UpdateRepoIndex api to include watchers
* Add timeout
---
integrations/repo_search_test.go | 23 +++++++----------------
models/repo_indexer.go | 21 +++++++++++++--------
2 files changed, 20 insertions(+), 24 deletions(-)
diff --git a/integrations/repo_search_test.go b/integrations/repo_search_test.go
index 3422aeaa12..282caa707e 100644
--- a/integrations/repo_search_test.go
+++ b/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")
diff --git a/models/repo_indexer.go b/models/repo_indexer.go
index 9a7daa0ff8..140ec66c03 100644
--- a/models/repo_indexer.go
+++ b/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) {