mirror of
https://github.com/go-gitea/gitea.git
synced 2024-08-24 21:34:36 -04:00
* Start adding mechanism to return unhandled data Signed-off-by: Andrew Thornton <art27@cantab.net> * Create pushback interface Signed-off-by: Andrew Thornton <art27@cantab.net> * Add Pausable interface to WorkerPool and Manager Signed-off-by: Andrew Thornton <art27@cantab.net> * Implement Pausable and PushBack for the bytefifos Signed-off-by: Andrew Thornton <art27@cantab.net> * Implement Pausable and Pushback for ChannelQueues and ChannelUniqueQueues Signed-off-by: Andrew Thornton <art27@cantab.net> * Wire in UI for pausing Signed-off-by: Andrew Thornton <art27@cantab.net> * add testcases and fix a few issues Signed-off-by: Andrew Thornton <art27@cantab.net> * fix build Signed-off-by: Andrew Thornton <art27@cantab.net> * prevent "race" in the test Signed-off-by: Andrew Thornton <art27@cantab.net> * fix jsoniter mismerge Signed-off-by: Andrew Thornton <art27@cantab.net> * fix conflicts Signed-off-by: Andrew Thornton <art27@cantab.net> * fix format Signed-off-by: Andrew Thornton <art27@cantab.net> * Add warnings for no worker configurations and prevent data-loss with redis/levelqueue Signed-off-by: Andrew Thornton <art27@cantab.net> * Use StopTimer Signed-off-by: Andrew Thornton <art27@cantab.net> Co-authored-by: Lauris BH <lauris@nix.lv> Co-authored-by: 6543 <6543@obermui.de> Co-authored-by: techknowlogick <techknowlogick@gitea.io> Co-authored-by: wxiaoguang <wxiaoguang@gmail.com>
154 lines
3.4 KiB
Go
154 lines
3.4 KiB
Go
// 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.
|
|
|
|
package queue
|
|
|
|
import (
|
|
"os"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"code.gitea.io/gitea/modules/util"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
)
|
|
|
|
func TestLevelQueue(t *testing.T) {
|
|
handleChan := make(chan *testData)
|
|
handle := func(data ...Data) []Data {
|
|
assert.True(t, len(data) == 2)
|
|
for _, datum := range data {
|
|
testDatum := datum.(*testData)
|
|
handleChan <- testDatum
|
|
}
|
|
return nil
|
|
}
|
|
|
|
var lock sync.Mutex
|
|
queueShutdown := []func(){}
|
|
queueTerminate := []func(){}
|
|
|
|
tmpDir, err := os.MkdirTemp("", "level-queue-test-data")
|
|
assert.NoError(t, err)
|
|
defer util.RemoveAll(tmpDir)
|
|
|
|
queue, err := NewLevelQueue(handle, LevelQueueConfiguration{
|
|
ByteFIFOQueueConfiguration: ByteFIFOQueueConfiguration{
|
|
WorkerPoolConfiguration: WorkerPoolConfiguration{
|
|
QueueLength: 20,
|
|
BatchLength: 2,
|
|
BlockTimeout: 1 * time.Second,
|
|
BoostTimeout: 5 * time.Minute,
|
|
BoostWorkers: 5,
|
|
MaxWorkers: 10,
|
|
},
|
|
Workers: 1,
|
|
},
|
|
DataDir: tmpDir,
|
|
}, &testData{})
|
|
assert.NoError(t, err)
|
|
|
|
go queue.Run(func(shutdown func()) {
|
|
lock.Lock()
|
|
queueShutdown = append(queueShutdown, shutdown)
|
|
lock.Unlock()
|
|
}, func(terminate func()) {
|
|
lock.Lock()
|
|
queueTerminate = append(queueTerminate, terminate)
|
|
lock.Unlock()
|
|
})
|
|
|
|
test1 := testData{"A", 1}
|
|
test2 := testData{"B", 2}
|
|
|
|
err = queue.Push(&test1)
|
|
assert.NoError(t, err)
|
|
go func() {
|
|
err := queue.Push(&test2)
|
|
assert.NoError(t, err)
|
|
}()
|
|
|
|
result1 := <-handleChan
|
|
assert.Equal(t, test1.TestString, result1.TestString)
|
|
assert.Equal(t, test1.TestInt, result1.TestInt)
|
|
|
|
result2 := <-handleChan
|
|
assert.Equal(t, test2.TestString, result2.TestString)
|
|
assert.Equal(t, test2.TestInt, result2.TestInt)
|
|
|
|
err = queue.Push(test1)
|
|
assert.Error(t, err)
|
|
|
|
lock.Lock()
|
|
for _, callback := range queueShutdown {
|
|
callback()
|
|
}
|
|
lock.Unlock()
|
|
|
|
time.Sleep(200 * time.Millisecond)
|
|
err = queue.Push(&test1)
|
|
assert.NoError(t, err)
|
|
err = queue.Push(&test2)
|
|
assert.NoError(t, err)
|
|
select {
|
|
case <-handleChan:
|
|
assert.Fail(t, "Handler processing should have stopped")
|
|
default:
|
|
}
|
|
lock.Lock()
|
|
for _, callback := range queueTerminate {
|
|
callback()
|
|
}
|
|
lock.Unlock()
|
|
|
|
// Reopen queue
|
|
queue, err = NewWrappedQueue(handle,
|
|
WrappedQueueConfiguration{
|
|
Underlying: LevelQueueType,
|
|
Config: LevelQueueConfiguration{
|
|
ByteFIFOQueueConfiguration: ByteFIFOQueueConfiguration{
|
|
WorkerPoolConfiguration: WorkerPoolConfiguration{
|
|
QueueLength: 20,
|
|
BatchLength: 2,
|
|
BlockTimeout: 1 * time.Second,
|
|
BoostTimeout: 5 * time.Minute,
|
|
BoostWorkers: 5,
|
|
MaxWorkers: 10,
|
|
},
|
|
Workers: 1,
|
|
},
|
|
DataDir: tmpDir,
|
|
},
|
|
}, &testData{})
|
|
assert.NoError(t, err)
|
|
|
|
go queue.Run(func(shutdown func()) {
|
|
lock.Lock()
|
|
queueShutdown = append(queueShutdown, shutdown)
|
|
lock.Unlock()
|
|
}, func(terminate func()) {
|
|
lock.Lock()
|
|
queueTerminate = append(queueTerminate, terminate)
|
|
lock.Unlock()
|
|
})
|
|
|
|
result3 := <-handleChan
|
|
assert.Equal(t, test1.TestString, result3.TestString)
|
|
assert.Equal(t, test1.TestInt, result3.TestInt)
|
|
|
|
result4 := <-handleChan
|
|
assert.Equal(t, test2.TestString, result4.TestString)
|
|
assert.Equal(t, test2.TestInt, result4.TestInt)
|
|
|
|
lock.Lock()
|
|
for _, callback := range queueShutdown {
|
|
callback()
|
|
}
|
|
for _, callback := range queueTerminate {
|
|
callback()
|
|
}
|
|
lock.Unlock()
|
|
}
|