Skip to content

Commit 78da4aa

Browse files
committed
add first draft custom next interval functions, unify tests
1 parent 0c036a1 commit 78da4aa

11 files changed

Lines changed: 220 additions & 38 deletions

database/dbWorker.go

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -70,6 +70,7 @@ func (r WorkerDBHandler) CreateTable() error {
7070
name VARCHAR(100) DEFAULT '',
7171
options JSONB DEFAULT '{}',
7272
available_tasks VARCHAR[] DEFAULT ARRAY[]::VARCHAR[],
73+
available_next_interval VARCHAR[] DEFAULT ARRAY[]::VARCHAR[],
7374
current_concurrency INT DEFAULT 0,
7475
max_concurrency INT DEFAULT 1,
7576
status VARCHAR(50) DEFAULT 'RUNNING',
@@ -144,24 +145,27 @@ func (r WorkerDBHandler) UpdateWorker(worker *model.Worker) (*model.Worker, erro
144145
name = $1,
145146
options = $2,
146147
available_tasks = $3,
147-
max_concurrency = $4,
148-
status = $5,
148+
available_next_interval = $4,
149+
max_concurrency = $5,
150+
status = $6,
149151
updated_at = CURRENT_TIMESTAMP
150152
WHERE
151-
rid = $6
153+
rid = $7
152154
RETURNING
153155
id,
154156
rid,
155157
name,
156158
options,
157159
available_tasks,
160+
available_next_interval,
158161
max_concurrency,
159162
status,
160163
created_at,
161164
updated_at;`,
162165
worker.Name,
163166
worker.Options,
164167
pq.Array(worker.AvailableTasks),
168+
pq.Array(worker.AvailableNextIntervalFuncs),
165169
worker.MaxConcurrency,
166170
worker.Status,
167171
worker.RID,
@@ -174,6 +178,7 @@ func (r WorkerDBHandler) UpdateWorker(worker *model.Worker) (*model.Worker, erro
174178
&updatedWorker.Name,
175179
&updatedWorker.Options,
176180
pq.Array(&updatedWorker.AvailableTasks),
181+
pq.Array(&updatedWorker.AvailableNextIntervalFuncs),
177182
&updatedWorker.MaxConcurrency,
178183
&updatedWorker.Status,
179184
&updatedWorker.CreatedAt,
@@ -211,6 +216,7 @@ func (r WorkerDBHandler) SelectWorker(rid uuid.UUID) (*model.Worker, error) {
211216
name,
212217
options,
213218
available_tasks,
219+
available_next_interval,
214220
max_concurrency,
215221
status,
216222
created_at,
@@ -227,6 +233,7 @@ func (r WorkerDBHandler) SelectWorker(rid uuid.UUID) (*model.Worker, error) {
227233
&worker.Name,
228234
&worker.Options,
229235
pq.Array(&worker.AvailableTasks),
236+
pq.Array(&worker.AvailableNextIntervalFuncs),
230237
&worker.MaxConcurrency,
231238
&worker.Status,
232239
&worker.CreatedAt,
@@ -250,6 +257,7 @@ func (r WorkerDBHandler) SelectAllWorkers(lastID int, entries int) ([]*model.Wor
250257
name,
251258
options,
252259
available_tasks,
260+
available_next_interval,
253261
max_concurrency,
254262
status,
255263
created_at,
@@ -284,6 +292,7 @@ func (r WorkerDBHandler) SelectAllWorkers(lastID int, entries int) ([]*model.Wor
284292
&worker.Name,
285293
&worker.Options,
286294
pq.Array(&worker.AvailableTasks),
295+
pq.Array(&worker.AvailableNextIntervalFuncs),
287296
&worker.MaxConcurrency,
288297
&worker.Status,
289298
&worker.CreatedAt,
@@ -314,6 +323,7 @@ func (r WorkerDBHandler) SelectAllWorkersBySearch(search string, lastID int, ent
314323
name,
315324
options,
316325
available_tasks,
326+
available_next_interval,
317327
max_concurrency,
318328
status,
319329
created_at,
@@ -351,6 +361,7 @@ func (r WorkerDBHandler) SelectAllWorkersBySearch(search string, lastID int, ent
351361
&worker.Name,
352362
&worker.Options,
353363
pq.Array(&worker.AvailableTasks),
364+
pq.Array(&worker.AvailableNextIntervalFuncs),
354365
&worker.MaxConcurrency,
355366
&worker.Status,
356367
&worker.CreatedAt,

database/dbWorker_test.go

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -98,11 +98,16 @@ func TestWorkerUpdateWorker(t *testing.T) {
9898
RetryDelay: 1,
9999
RetryBackoff: model.RETRY_BACKOFF_EXPONENTIAL,
100100
}
101+
insertedWorker.AvailableTasks = []string{"task1", "task2"}
102+
insertedWorker.AvailableNextIntervalFuncs = []string{"interval1", "interval2"}
101103

102104
updatedWorker, err := workerDBHandler.UpdateWorker(insertedWorker)
103105
assert.NoError(t, err, "Expected UpdateWorker to not return an error")
104106
assert.Equal(t, updatedWorker.Name, insertedWorker.Name, "Expected updated worker Name to match")
105107
assert.Equal(t, updatedWorker.Options, insertedWorker.Options, "Expected updated worker Options to match")
108+
assert.Equal(t, updatedWorker.AvailableTasks, insertedWorker.AvailableTasks, "Expected updated worker AvailableTasks to match")
109+
assert.Equal(t, updatedWorker.AvailableNextIntervalFuncs, insertedWorker.AvailableNextIntervalFuncs, "Expected updated worker AvailableNextInterval to match")
110+
assert.Equal(t, updatedWorker.MaxConcurrency, insertedWorker.MaxConcurrency, "Expected updated worker MaxConcurrency to match")
106111
}
107112

108113
func TestWorkerDeleteWorker(t *testing.T) {

main_test.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,7 @@ func newQueuerMock(name string, maxConcurrency int, options ...*model.OnError) *
9595
jobDeleteListener: jobDeleteListener,
9696
JobPollInterval: 1 * time.Minute,
9797
tasks: map[string]*model.Task{},
98+
nextIntervalFuncs: map[string]model.NextIntervalFunc{},
9899
log: logger,
99100
}
100101
}

model/option_test.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -145,7 +145,7 @@ func TestIsValid(t *testing.T) {
145145
Start: time.Now().Add(1 * time.Minute),
146146
Interval: 0,
147147
MaxCount: 3,
148-
NextInterval: nil,
148+
NextInterval: "",
149149
},
150150
},
151151
wantErr: true,

model/optionsSchedule.go

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -7,11 +7,15 @@ import (
77
"time"
88
)
99

10+
// NextIntervalFunc defines a function type that calculates the next interval
11+
// based on the start time and the current count of executions.
12+
type NextIntervalFunc func(start time.Time, currentCount int) time.Time
13+
1014
type Schedule struct {
11-
Start time.Time `json:"start"`
12-
Interval time.Duration `json:"interval"`
13-
MaxCount int `json:"max_count"`
14-
NextInterval func(start time.Time, currentCount int) time.Time `json:"-"`
15+
Start time.Time `json:"start"`
16+
Interval time.Duration `json:"interval"`
17+
MaxCount int `json:"max_count"`
18+
NextInterval string `json:"next_interval,omitempty"`
1519
}
1620

1721
func (c *Schedule) IsValid() error {
@@ -21,7 +25,7 @@ func (c *Schedule) IsValid() error {
2125
if c.MaxCount < 0 {
2226
return errors.New("maxCount must be greater than or equal to 0")
2327
}
24-
if c.Interval <= time.Duration(0) && c.NextInterval == nil {
28+
if c.Interval <= time.Duration(0) && c.NextInterval == "" {
2529
return errors.New("interval must be greater than zero or nextInterval must be provided")
2630
}
2731
return nil

model/worker.go

Lines changed: 10 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -13,15 +13,16 @@ const (
1313
)
1414

1515
type Worker struct {
16-
ID int `json:"id"`
17-
RID uuid.UUID `json:"rid"`
18-
Name string `json:"name"`
19-
Options *OnError `json:"options,omitempty"`
20-
MaxConcurrency int `json:"max_concurrency"`
21-
AvailableTasks []string `json:"available_tasks"`
22-
Status string `json:"status"`
23-
CreatedAt time.Time `json:"created_at"`
24-
UpdatedAt time.Time `json:"updated_at"`
16+
ID int `json:"id"`
17+
RID uuid.UUID `json:"rid"`
18+
Name string `json:"name"`
19+
Options *OnError `json:"options,omitempty"`
20+
MaxConcurrency int `json:"max_concurrency"`
21+
AvailableTasks []string `json:"available_tasks,omitempty"`
22+
AvailableNextIntervalFuncs []string `json:"available_next_interval,omitempty"`
23+
Status string `json:"status"`
24+
CreatedAt time.Time `json:"created_at"`
25+
UpdatedAt time.Time `json:"updated_at"`
2526
}
2627

2728
func NewWorker(name string, maxConcurrency int) (*Worker, error) {

queuer.go

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,8 +29,9 @@ type Queuer struct {
2929
jobUpdateListener *database.QueuerListener
3030
jobDeleteListener *database.QueuerListener
3131
JobPollInterval time.Duration
32-
// Tasks
33-
tasks map[string]*model.Task
32+
// Available functions
33+
tasks map[string]*model.Task
34+
nextIntervalFuncs map[string]model.NextIntervalFunc
3435
// Logger
3536
log *log.Logger
3637
}
@@ -109,6 +110,7 @@ func NewQueuer(name string, maxConcurrency int, options ...*model.OnError) *Queu
109110
jobDeleteListener: jobDeleteListener,
110111
JobPollInterval: 1 * time.Minute,
111112
tasks: map[string]*model.Task{},
113+
nextIntervalFuncs: map[string]model.NextIntervalFunc{},
112114
log: logger,
113115
}
114116
}

queuerNextInterval.go

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
package queuer
2+
3+
import (
4+
"queuer/helper"
5+
"queuer/model"
6+
"slices"
7+
)
8+
9+
func (q *Queuer) AddNextIntervalFunc(nif model.NextIntervalFunc) *model.Worker {
10+
if nif == nil {
11+
q.log.Panicf("error adding next interval: NextIntervalFunc cannot be nil")
12+
}
13+
14+
nifName, err := helper.GetTaskNameFromInterface(nif)
15+
if err != nil {
16+
q.log.Panicf("error getting function name: %v", err)
17+
}
18+
if slices.Contains(q.worker.AvailableNextIntervalFuncs, nifName) {
19+
q.log.Panicf("NextIntervalFunc with name %v already exists", nifName)
20+
}
21+
22+
q.nextIntervalFuncs[nifName] = nif
23+
q.worker.AvailableNextIntervalFuncs = append(q.worker.AvailableNextIntervalFuncs, nifName)
24+
25+
worker, err := q.dbWorker.UpdateWorker(q.worker)
26+
if err != nil {
27+
q.log.Panicf("error updating worker: %v", err)
28+
}
29+
30+
q.log.Printf("NextInterval function added with name %v", nifName)
31+
32+
return worker
33+
}
34+
35+
func (q *Queuer) AddNextIntervalFuncWithName(nif model.NextIntervalFunc, name string) *model.Worker {
36+
if nif == nil {
37+
q.log.Panicf("NextIntervalFunc cannot be nil")
38+
}
39+
if slices.Contains(q.worker.AvailableNextIntervalFuncs, name) {
40+
q.log.Panicf("NextIntervalFunc with name %v already exists", name)
41+
}
42+
43+
q.nextIntervalFuncs[name] = nif
44+
q.worker.AvailableNextIntervalFuncs = append(q.worker.AvailableNextIntervalFuncs, name)
45+
46+
worker, err := q.dbWorker.UpdateWorker(q.worker)
47+
if err != nil {
48+
q.log.Panicf("error updating worker: %v", err)
49+
}
50+
51+
q.log.Printf("NextInterval function added with name %v", name)
52+
53+
return worker
54+
}

queuerNextInterval_test.go

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,104 @@
1+
package queuer
2+
3+
import (
4+
"queuer/model"
5+
"testing"
6+
"time"
7+
8+
"github.com/stretchr/testify/assert"
9+
"github.com/stretchr/testify/require"
10+
)
11+
12+
func MockNextIntervalFunc1(start time.Time, currentCount int) time.Time {
13+
return start.Add(time.Hour * time.Duration(currentCount))
14+
}
15+
func MockNextIntervalFunc2(start time.Time, currentCount int) time.Time {
16+
return start.Add(time.Hour * time.Duration(currentCount))
17+
}
18+
19+
func TestAddNextIntervalFunc(t *testing.T) {
20+
t.Run("Successfully adds NextIntervalFunc", func(t *testing.T) {
21+
testQueuer := newQueuerMock("TestQueuer", 100)
22+
23+
worker := testQueuer.AddNextIntervalFunc(MockNextIntervalFunc1)
24+
require.NotNil(t, worker, "Expected worker to be returned after adding NextIntervalFunc")
25+
assert.Contains(t, testQueuer.worker.AvailableNextIntervalFuncs, "queuer.MockNextIntervalFunc1", "Expected NextIntervalFunc to be added to worker's AvailableNextIntervalFuncs")
26+
assert.Equal(t, 1, len(testQueuer.worker.AvailableNextIntervalFuncs), "Expected only one NextIntervalFunc to be added to worker's AvailableNextIntervalFuncs")
27+
})
28+
29+
t.Run("Successfully adds multiple NextIntervalFuncs", func(t *testing.T) {
30+
testQueuer := newQueuerMock("TestQueuer", 100)
31+
32+
testQueuer.AddNextIntervalFunc(MockNextIntervalFunc1)
33+
worker := testQueuer.AddNextIntervalFunc(MockNextIntervalFunc2)
34+
require.NotNil(t, worker, "Expected worker to be returned after adding NextIntervalFunc")
35+
assert.Contains(t, testQueuer.worker.AvailableNextIntervalFuncs, "queuer.MockNextIntervalFunc1", "Expected NextIntervalFunc to be added to worker's AvailableNextIntervalFuncs")
36+
assert.Contains(t, testQueuer.worker.AvailableNextIntervalFuncs, "queuer.MockNextIntervalFunc2", "Expected second NextIntervalFunc to be added to worker's AvailableNextIntervalFuncs")
37+
assert.Equal(t, 2, len(testQueuer.worker.AvailableNextIntervalFuncs), "Expected two NextIntervalFuncs to be added to worker's AvailableNextIntervalFuncs")
38+
})
39+
40+
t.Run("Fails to add nil NextIntervalFunc", func(t *testing.T) {
41+
testQueuer := newQueuerMock("TestQueuer", 100)
42+
43+
var worker *model.Worker
44+
defer func() {
45+
r := recover()
46+
require.NotNil(t, r, "Expected panic when adding nil NextIntervalFunc")
47+
require.Nil(t, worker, "Expected worker to be nil when adding nil NextIntervalFunc")
48+
}()
49+
50+
worker = testQueuer.AddNextIntervalFunc(nil)
51+
})
52+
53+
t.Run("Fails to add NextIntervalFunc with existing name", func(t *testing.T) {
54+
testQueuer := newQueuerMock("TestQueuer", 100)
55+
56+
var worker *model.Worker
57+
defer func() {
58+
r := recover()
59+
require.NotNil(t, r, "Expected panic when adding NextIntervalFunc with existing name")
60+
require.Nil(t, worker, "Expected worker to be nil when adding NextIntervalFunc with existing name")
61+
}()
62+
63+
testQueuer.AddNextIntervalFunc(MockNextIntervalFunc1)
64+
worker = testQueuer.AddNextIntervalFunc(MockNextIntervalFunc1)
65+
})
66+
}
67+
68+
func TestAddNextIntervalFuncWithName(t *testing.T) {
69+
t.Run("Successfully adds NextIntervalFunc with name", func(t *testing.T) {
70+
testQueuer := newQueuerMock("TestQueuer", 100)
71+
72+
worker := testQueuer.AddNextIntervalFuncWithName(MockNextIntervalFunc1, "CustomFuncName")
73+
require.NotNil(t, worker, "Expected worker to be returned after adding NextIntervalFunc with name")
74+
assert.Contains(t, testQueuer.worker.AvailableNextIntervalFuncs, "CustomFuncName", "Expected NextIntervalFunc to be added to worker's AvailableNextIntervalFuncs")
75+
assert.Equal(t, 1, len(testQueuer.worker.AvailableNextIntervalFuncs), "Expected only one NextIntervalFunc to be added to worker's AvailableNextIntervalFuncs")
76+
})
77+
78+
t.Run("Fails to add nil NextIntervalFunc with name", func(t *testing.T) {
79+
testQueuer := newQueuerMock("TestQueuer", 100)
80+
81+
var worker *model.Worker
82+
defer func() {
83+
r := recover()
84+
require.NotNil(t, r, "Expected panic when adding nil NextIntervalFunc with name")
85+
require.Nil(t, worker, "Expected worker to be nil when adding nil NextIntervalFunc with name")
86+
}()
87+
88+
worker = testQueuer.AddNextIntervalFuncWithName(nil, "CustomFuncName")
89+
})
90+
91+
t.Run("Fails to add NextIntervalFunc with existing name", func(t *testing.T) {
92+
testQueuer := newQueuerMock("TestQueuer", 100)
93+
94+
var worker *model.Worker
95+
defer func() {
96+
r := recover()
97+
require.NotNil(t, r, "Expected panic when adding NextIntervalFunc with existing name")
98+
require.Nil(t, worker, "Expected worker to be nil when adding NextIntervalFunc with existing name")
99+
}()
100+
101+
testQueuer.AddNextIntervalFuncWithName(MockNextIntervalFunc1, "CustomFuncName")
102+
worker = testQueuer.AddNextIntervalFuncWithName(MockNextIntervalFunc2, "CustomFuncName")
103+
})
104+
}

0 commit comments

Comments
 (0)