executor/options_test.go at main · xzc-coder/executor · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
package executor

import (
	"errors"
	"github.com/stretchr/testify/require"
	"strconv"
	"sync/atomic"
	"testing"
	"time"
)

// TestCorePoolSize 测试核心协程数设置
func TestCorePoolSize(t *testing.T) {
	executor := NewExecutor(WithCorePoolSize(1))
	require.Equal(t, 1, executor.CorePoolSize())
}

// TestMaxPoolSize 测试最大协程数设置
func TestMaxPoolSize(t *testing.T) {
	executor := NewExecutor(WithMaxPoolSize(1))
	require.Equal(t, 1, executor.MaxPoolSize())
}

// TestKeepAliveTime 测试非核心协程活跃时间
func TestKeepAliveTime(t *testing.T) {
	executor := NewExecutor(WithCorePoolSize(0), WithMaxPoolSize(1), WithTaskBlockingQueueDefault(0), WithKeepAliveTime(100*time.Millisecond))
	f, _ := executor.Execute(func() {
	})
	f.Get()
	require.Equal(t, 1, len(executor.WorkerInfos()))
	time.Sleep(150 * time.Millisecond)
	require.Equal(t, 0, len(executor.WorkerInfos()))
}

type executeExpandOption struct {
	number atomic.Int32
}

func (e *executeExpandOption) TaskBeforeExecute(workerName string, runnable Runnable) {
	e.number.Add(1)
}

func (e *executeExpandOption) TaskAfterExecute(workerName string, runnable Runnable) {
	e.number.Add(1)
}

// TestExecuteExpand 测试执行扩展
func TestExecuteExpand(t *testing.T) {
	executeExpand := &executeExpandOption{}
	executor := NewExecutor(WithCorePoolSize(1), WithExecuteExpand(executeExpand))
	f, _ := executor.Execute(func() {

	})
	f.Get()
	require.Equal(t, int32(2), executeExpand.number.Load())
}

// TestGoroutineNameFactory 测试协程命名工厂
func TestGoroutineNameFactory(t *testing.T) {
	gNameFactory := func(number int) string {
		return "test-" + strconv.Itoa(number)
	}
	executor := NewExecutor(WithCorePoolSize(2), WithGoroutineNameFactory(gNameFactory))
	f, _ := executor.Execute(func() {
		time.Sleep(50 * time.Millisecond)
	})
	f, _ = executor.Execute(func() {
		time.Sleep(100 * time.Millisecond)
	})
	f.Get()
	workerInfos := executor.WorkerInfos()
	require.Equal(t, 2, len(workerInfos))
	workerInfo := workerInfos[0]
	workerInfo1 := workerInfos[1]
	require.Equal(t, "test-1", workerInfo.WorkerName)
	require.Equal(t, "test-2", workerInfo1.WorkerName)
}

// TestRejectedHandler 测试拒绝处理器
func TestRejectedHandler(t *testing.T) {
	rHandler := func(Runnable, Executor) error {
		return errors.New("TestRejectedHandler")
	}
	executor := NewExecutor(WithCorePoolSize(0), WithTaskBlockingQueueDefault(0), WithRejectedHandler(rHandler))
	_, err := executor.Execute(func() {

	})
	require.Equal(t, "TestRejectedHandler", err.Error())
}

// TestPanicHandler 测试恐慌处理器
func TestPanicHandler(t *testing.T) {
	number := atomic.Int32{}
	pHandler := func(runnable Runnable, workerInfo *WorkerInfo, v any) {
		number.Add(v.(int32))
	}
	executor := NewExecutor(WithCorePoolSize(1), WithPanicHandler(pHandler))
	f, _ := executor.Execute(func() {
		panic(int32(100))
	})
	f.Get()
	time.Sleep(100 * time.Millisecond)
	require.Equal(t, int32(100), number.Load())
}