cq/queue_priority_manager_test.go at master · gnikyt/cq · 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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
package cq

import (
	"context"
	"errors"
	"reflect"
	"testing"
	"time"
)

func TestPriorityQueueManager_RegisterAndNames(t *testing.T) {
	mgr := NewPriorityQueueManager()

	if err := mgr.Register("", newPriorityManagerTestQueue(t)); !errors.Is(err, ErrPriorityQueueManagerInvalidName) {
		t.Fatalf("Register(empty): got %v, want %v", err, ErrPriorityQueueManagerInvalidName)
	}
	if err := mgr.Register("high", nil); !errors.Is(err, ErrPriorityQueueManagerNilQueue) {
		t.Fatalf("Register(nil): got %v, want %v", err, ErrPriorityQueueManagerNilQueue)
	}

	if err := mgr.Register("low", newPriorityManagerTestQueue(t)); err != nil {
		t.Fatalf("Register(low): unexpected err: %v", err)
	}
	if err := mgr.Register("high", newPriorityManagerTestQueue(t)); err != nil {
		t.Fatalf("Register(high): unexpected err: %v", err)
	}
	if err := mgr.Register("high", newPriorityManagerTestQueue(t)); !errors.Is(err, ErrPriorityQueueManagerExists) {
		t.Fatalf("Register(duplicate): got %v, want %v", err, ErrPriorityQueueManagerExists)
	}

	names := mgr.Names()
	if !reflect.DeepEqual(names, []string{"high", "low"}) {
		t.Fatalf("Names(): got %v, want [high low]", names)
	}
}

func TestPriorityQueueManager_Submit(t *testing.T) {
	mgr := NewPriorityQueueManager()
	highQ := newPriorityManagerTestQueue(t)
	lowQ := newPriorityManagerTestQueue(t)

	if err := mgr.Register("high", highQ); err != nil {
		t.Fatalf("Register(high): %v", err)
	}
	if err := mgr.Register("low", lowQ); err != nil {
		t.Fatalf("Register(low): %v", err)
	}

	defer mgr.StopAll(true)

	doneHigh := make(chan struct{}, 1)
	doneLow := make(chan struct{}, 1)
	if _, err := mgr.Submit(context.Background(), "high", func(ctx context.Context) error {
		doneHigh <- struct{}{}
		return nil
	}, PriorityHighest); err != nil {
		t.Fatalf("Submit(high): unexpected err: %v", err)
	}
	if _, err := mgr.Submit(context.Background(), "low", func(ctx context.Context) error {
		doneLow <- struct{}{}
		return nil
	}, PriorityLow); err != nil {
		t.Fatalf("Submit(low): unexpected err: %v", err)
	}

	recvOrFail(t, doneHigh, 1*time.Second, "high priority queue job did not execute")
	recvOrFail(t, doneLow, 1*time.Second, "low priority queue job did not execute")

	handle, err := mgr.Submit(context.Background(), "high", func(ctx context.Context) error { return nil }, PriorityHigh, WithNonBlocking())
	if err != nil || handle == nil {
		t.Fatalf("Submit(non-blocking): got (%v, %v), want (handle, nil)", handle, err)
	}
}

func TestPriorityQueueManager_SubmitAfter(t *testing.T) {
	mgr := NewPriorityQueueManager()
	pq := newPriorityManagerTestQueue(t)

	if err := mgr.Register("delayed", pq); err != nil {
		t.Fatalf("Register(delayed): %v", err)
	}

	defer mgr.StopAll(true)

	done := make(chan struct{}, 1)
	if _, err := mgr.SubmitAfter(context.Background(), "delayed", func(ctx context.Context) error {
		done <- struct{}{}
		return nil
	}, PriorityHigh, 100*time.Millisecond); err != nil {
		t.Fatalf("SubmitAfter(delayed): unexpected err: %v", err)
	}

	mustNotRecv(t, done, 50*time.Millisecond, "delayed priority queue job executed too early")
	recvOrFail(t, done, 1*time.Second, "delayed priority queue job did not execute")
}

func TestPriorityQueueManager_StopAndNotFound(t *testing.T) {
	mgr := NewPriorityQueueManager()
	pq := newPriorityManagerTestQueue(t)

	if err := mgr.Register("high", pq); err != nil {
		t.Fatalf("Register(high): %v", err)
	}

	if err := mgr.Stop("high", true); err != nil {
		t.Fatalf("Stop(high): unexpected err: %v", err)
	}

	if !pq.queue.IsStopped() {
		t.Fatal("Stop(high, true): underlying queue should be stopped")
	}

	if err := mgr.Stop("missing", true); !errors.Is(err, ErrPriorityQueueManagerNotFound) {
		t.Fatalf("Stop(missing): got %v, want %v", err, ErrPriorityQueueManagerNotFound)
	}
	if _, err := mgr.Submit(context.Background(), "missing", func(ctx context.Context) error { return nil }, PriorityHigh); !errors.Is(err, ErrPriorityQueueManagerNotFound) {
		t.Fatalf("Submit(missing): got %v, want %v", err, ErrPriorityQueueManagerNotFound)
	}
	if _, err := mgr.SubmitAfter(context.Background(), "missing", func(ctx context.Context) error { return nil }, PriorityHigh, time.Millisecond); !errors.Is(err, ErrPriorityQueueManagerNotFound) {
		t.Fatalf("SubmitAfter(missing): got %v, want %v", err, ErrPriorityQueueManagerNotFound)
	}
}

func newPriorityManagerTestQueue(t *testing.T) *PriorityQueue {
	t.Helper()

	queue := NewQueue(1, 1, 10)
	queue.Start()
	return NewPriorityQueue(queue, 10, WithPriorityTick(time.Millisecond))
}