mirror of
https://gitee.com/dromara/mayfly-go
synced 2025-11-03 07:50:25 +08:00
* Merge branch 'dev' of gitee.com:dromara/mayfly-go into feat-db-bak * fix: 修复 BINLOG 同步任务加载问题
124 lines
2.4 KiB
Go
124 lines
2.4 KiB
Go
package runner
|
|
|
|
import (
|
|
"context"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
"mayfly-go/pkg/utils/timex"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
var _ Job = &testJob{}
|
|
|
|
func newTestJob(key string) *testJob {
|
|
return &testJob{
|
|
Key: key,
|
|
}
|
|
}
|
|
|
|
type testJob struct {
|
|
Key JobKey
|
|
status int
|
|
}
|
|
|
|
func (t *testJob) Update(_ Job) {}
|
|
|
|
func (t *testJob) GetKey() JobKey {
|
|
return t.Key
|
|
}
|
|
|
|
func TestRunner_Close(t *testing.T) {
|
|
signal := make(chan struct{}, 1)
|
|
waiting := sync.WaitGroup{}
|
|
waiting.Add(1)
|
|
runner := NewRunner[*testJob](1, func(ctx context.Context, job *testJob) {
|
|
waiting.Done()
|
|
timex.SleepWithContext(ctx, time.Hour)
|
|
signal <- struct{}{}
|
|
})
|
|
go func() {
|
|
job := &testJob{
|
|
Key: "close",
|
|
}
|
|
_ = runner.Add(context.Background(), job)
|
|
}()
|
|
waiting.Wait()
|
|
timer := time.NewTimer(time.Microsecond * 10)
|
|
defer timer.Stop()
|
|
runner.Close()
|
|
select {
|
|
case <-timer.C:
|
|
require.FailNow(t, "runner 未能及时退出")
|
|
case <-signal:
|
|
}
|
|
}
|
|
|
|
func TestRunner_AddJob(t *testing.T) {
|
|
type testCase struct {
|
|
name string
|
|
job *testJob
|
|
want error
|
|
}
|
|
testCases := []testCase{
|
|
{
|
|
name: "first job",
|
|
job: newTestJob("single"),
|
|
want: nil,
|
|
},
|
|
{
|
|
name: "second job",
|
|
job: newTestJob("dual"),
|
|
want: nil,
|
|
},
|
|
{
|
|
name: "repetitive job",
|
|
job: newTestJob("dual"),
|
|
want: ErrJobExist,
|
|
},
|
|
}
|
|
runner := NewRunner[*testJob](1, func(ctx context.Context, job *testJob) {
|
|
timex.SleepWithContext(ctx, time.Hour)
|
|
})
|
|
defer runner.Close()
|
|
for _, tc := range testCases {
|
|
err := runner.Add(context.Background(), tc.job)
|
|
if tc.want != nil {
|
|
require.ErrorIs(t, err, tc.want)
|
|
continue
|
|
}
|
|
require.NoError(t, err)
|
|
}
|
|
}
|
|
|
|
func TestJob_UpdateStatus(t *testing.T) {
|
|
const d = time.Millisecond * 20
|
|
const (
|
|
unknown = iota
|
|
running
|
|
finished
|
|
)
|
|
runner := NewRunner[*testJob](1, func(ctx context.Context, job *testJob) {
|
|
job.status = running
|
|
timex.SleepWithContext(ctx, d*2)
|
|
job.status = finished
|
|
})
|
|
first := newTestJob("first")
|
|
second := newTestJob("second")
|
|
_ = runner.Add(context.Background(), first)
|
|
_ = runner.Add(context.Background(), second)
|
|
|
|
time.Sleep(d)
|
|
assert.Equal(t, running, first.status)
|
|
assert.Equal(t, unknown, second.status)
|
|
|
|
time.Sleep(d * 2)
|
|
assert.Equal(t, finished, first.status)
|
|
assert.Equal(t, running, second.status)
|
|
|
|
time.Sleep(d * 2)
|
|
assert.Equal(t, finished, first.status)
|
|
assert.Equal(t, finished, second.status)
|
|
}
|