use levelqueue instead of ledis

Author: lunny

Gopkg.lock

@@ -112,4 +112,4 @@ ignored = ["google.golang.org/appengine*"]
 
 [[constraint]]
   name = "github.com/prometheus/client_golang"
-  version = "0.9.0"
+  version = "0.9.0"

Gopkg.toml

@@ -37,11 +37,10 @@ func InitIssueIndexer() error {
 
 	var err error
 	switch setting.Indexer.IssueIndexerQueueType {
-	case setting.LedisLocalQueueType:
-		issueIndexerUpdateQueue, err = issues.NewLedisLocalQueue(
+	case setting.LevelQueueType:
+		issueIndexerUpdateQueue, err = issues.NewLevelQueue(
 			issueIndexer,
 			setting.Indexer.IssueIndexerQueueDir,
-			setting.Indexer.IssueIndexerQueueDBIndex,
 			setting.Indexer.IssueIndexerQueueBatchNumber)
 		if err != nil {
 			return err

models/issue_indexer.go

@@ -1,4 +1,4 @@
-// Copyright 2018 The Gitea Authors. All rights reserved.
+// Copyright 2019 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.
 
@@ -9,52 +9,40 @@ import (
 	"time"
 
 	"code.gitea.io/gitea/modules/log"
-	"github.com/siddontang/ledisdb/config"
-	"github.com/siddontang/ledisdb/ledis"
+	"github.com/lunny/levelqueue"
 )
 
 var (
-	_ Queue = &LedisLocalQueue{}
-
-	ledisLocalKey = []byte("ledis_local_key")
+	_ Queue = &LevelQueue{}
 )
 
-// LedisLocalQueue implements a ledis as a disk library queue
-type LedisLocalQueue struct {
+// LevelQueue implements a disk library queue
+type LevelQueue struct {
 	indexer     Indexer
-	ledis       *ledis.Ledis
-	db          *ledis.DB
+	queue       *levelqueue.Queue
 	batchNumber int
 }
 
-// NewLedisLocalQueue creates a ledis local queue
-func NewLedisLocalQueue(indexer Indexer, dataDir string, dbIdx, batchNumber int) (*LedisLocalQueue, error) {
-	ledis, err := ledis.Open(&config.Config{
-		DataDir: dataDir,
-	})
-	if err != nil {
-		return nil, err
-	}
-
-	db, err := ledis.Select(dbIdx)
+// NewLevelQueue creates a ledis local queue
+func NewLevelQueue(indexer Indexer, dataDir string, batchNumber int) (*LevelQueue, error) {
+	queue, err := levelqueue.Open(dataDir)
 	if err != nil {
 		return nil, err
 	}
 
-	return &LedisLocalQueue{
+	return &LevelQueue{
 		indexer:     indexer,
-		ledis:       ledis,
-		db:          db,
+		queue:       queue,
 		batchNumber: batchNumber,
 	}, nil
 }
 
 // Run starts to run the queue
-func (l *LedisLocalQueue) Run() error {
+func (l *LevelQueue) Run() error {
 	var i int
 	var datas = make([]*IndexerData, 0, l.batchNumber)
 	for {
-		bs, err := l.db.RPop(ledisLocalKey)
+		bs, err := l.queue.RPop()
 		if err != nil {
 			log.Error(4, "RPop: %v", err)
 			time.Sleep(time.Millisecond * 100)
@@ -103,13 +91,13 @@ func (l *LedisLocalQueue) Run() error {
 }
 
 // Push will push the indexer data to queue
-func (l *LedisLocalQueue) Push(data *IndexerData) {
+func (l *LevelQueue) Push(data *IndexerData) {
 	bs, err := json.Marshal(data)
 	if err != nil {
 		log.Error(4, "Marshal: %v", err)
 		return
 	}
-	_, err = l.db.LPush(ledisLocalKey, bs)
+	err = l.queue.LPush(bs)
 	if err != nil {
 		log.Error(4, "LPush: %v", err)
 	}

modules/indexer/issues/queue_ledis_local.go renamed to modules/indexer/issues/queue_disk.go

@@ -85,8 +85,8 @@ const (
 
 // enumerates all the indexer queue types
 const (
-	LedisLocalQueueType = "ledis_local"
-	ChannelQueueType    = "channel"
+	LevelQueueType   = "levelqueue"
+	ChannelQueueType = "channel"
 )
 
 // settings
@@ -195,15 +195,13 @@ var (
 		MaxIndexerFileSize           int64
 		IssueIndexerQueueType        string
 		IssueIndexerQueueDir         string
-		IssueIndexerQueueDBIndex     int
 		IssueIndexerQueueBatchNumber int
 	}{
 		IssueType:                    "bleve",
 		IssuePath:                    "indexers/issues.bleve",
-		IssueIndexerQueueType:        LedisLocalQueueType,
+		IssueIndexerQueueType:        LevelQueueType,
 		IssueIndexerQueueDir:         "indexers/issues.queue",
 		IssueIndexerQueueBatchNumber: 20,
-		IssueIndexerQueueDBIndex:     0,
 	}
 
 	// Repository settings
@@ -1247,11 +1245,9 @@ func NewContext() {
 	}
 	Indexer.UpdateQueueLength = sec.Key("UPDATE_BUFFER_LEN").MustInt(20)
 	Indexer.MaxIndexerFileSize = sec.Key("MAX_FILE_SIZE").MustInt64(1024 * 1024)
-	Indexer.IssueIndexerQueueType = sec.Key("ISSUE_INDEXER_QUEUE_TYPE").MustString(LedisLocalQueueType)
+	Indexer.IssueIndexerQueueType = sec.Key("ISSUE_INDEXER_QUEUE_TYPE").MustString(LevelQueueType)
 	Indexer.IssueIndexerQueueDir = sec.Key("ISSUE_INDEXER_QUEUE_DIR").MustString(path.Join(AppDataPath, "indexers/issues.queue"))
 	Indexer.IssueIndexerQueueBatchNumber = sec.Key("ISSUE_INDEXER_QUEUE_BATCH_NUMBER").MustInt(20)
-	Indexer.IssueIndexerQueueDBIndex = sec.Key("ISSUE_INDEXER_QUEUE_DB_INDEX").MustInt(0)
-
 }
 
 // NewServices initializes the services