mirror of
https://github.com/acaloiaro/neoq
synced 2026-07-21 10:12:18 +00:00
429 lines
9.5 KiB
Go
429 lines
9.5 KiB
Go
package memory_test
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log/slog"
|
|
"os"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/acaloiaro/neoq"
|
|
"github.com/acaloiaro/neoq/backends/memory"
|
|
"github.com/acaloiaro/neoq/handler"
|
|
"github.com/acaloiaro/neoq/jobs"
|
|
"github.com/acaloiaro/neoq/logging"
|
|
"github.com/acaloiaro/neoq/testutils"
|
|
"github.com/pkg/errors"
|
|
"github.com/robfig/cron"
|
|
)
|
|
|
|
const (
|
|
queue = "testing"
|
|
)
|
|
|
|
var (
|
|
errPeriodicTimeout = errors.New("timed out waiting for periodic job")
|
|
errTimeout = errors.New("the test has timed out")
|
|
)
|
|
|
|
func ExampleNew() {
|
|
ctx := context.Background()
|
|
nq, err := neoq.New(ctx, neoq.WithBackend(memory.Backend))
|
|
if err != nil {
|
|
fmt.Println("initializing a new Neoq with no params should not return an error:", err)
|
|
return
|
|
}
|
|
defer nq.Shutdown(ctx)
|
|
|
|
fmt.Println("neoq initialized with default memory backend")
|
|
// Output: neoq initialized with default memory backend
|
|
}
|
|
|
|
// TestBasicJobProcessing tests that the memory backend is able to process the most basic jobs with the
|
|
// most basic configuration.
|
|
func TestBasicJobProcessing(t *testing.T) {
|
|
numJobs := 1000
|
|
doneCnt := 0
|
|
done := make(chan bool)
|
|
timeoutTimer := time.After(5 * time.Second)
|
|
|
|
ctx := context.TODO()
|
|
nq, err := neoq.New(ctx, neoq.WithBackend(memory.Backend))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
h := handler.New(queue, func(_ context.Context) (err error) {
|
|
done <- true
|
|
return
|
|
})
|
|
|
|
if err := nq.Start(ctx, h); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
go func() {
|
|
for i := 0; i < numJobs; i++ {
|
|
jid, e := nq.Enqueue(ctx, &jobs.Job{
|
|
Queue: queue,
|
|
Payload: map[string]interface{}{
|
|
"message": fmt.Sprintf("hello world: %d", i),
|
|
},
|
|
})
|
|
if e != nil || jid == jobs.DuplicateJobID {
|
|
slog.Error("job was not enqueued. either it was duplicate or this error caused it:", "error", e)
|
|
}
|
|
}
|
|
}()
|
|
|
|
for {
|
|
select {
|
|
case <-timeoutTimer:
|
|
err = jobs.ErrJobTimeout
|
|
case <-done:
|
|
doneCnt++
|
|
}
|
|
|
|
if doneCnt >= numJobs {
|
|
break
|
|
}
|
|
|
|
if err != nil {
|
|
break
|
|
}
|
|
}
|
|
|
|
if err != nil {
|
|
t.Error(err)
|
|
}
|
|
|
|
nq.Shutdown(ctx)
|
|
}
|
|
|
|
// TestBackendConfiguration tests that the memory backend receives and utilizes the MaxQueueCapacity handler
|
|
// configuration.
|
|
//
|
|
// This test works by enqueueing 3 jobs. Each job sleeps for longer than the test is willing to wait for the jobs to
|
|
// enqueue. The `done` channel is notified when all 3 jobs are enqueued.
|
|
//
|
|
// By serializing handler execution by initializing neoq with `handler.Concurrency(1)` and enqueueing jobs
|
|
// asynchronously, we can wait on `done` and a timeout channel to see which one completes first.
|
|
//
|
|
// Since the queue has a capacity of 1 and the handler is serialized, we know that `done` cannot be notified until job 1
|
|
// is complete, job 2 is processing, and job 3 can be added to the queue.
|
|
//
|
|
// If `done` is notified before the timeout channel, this test would fail, because that would mean Enqueue() is not
|
|
// blocking while the first Sleep()ing job is running. If the qeueue is blocking when it meets its capacity, we know
|
|
// that the max queue capacity configuration has taken effect.
|
|
func TestBackendConfiguration(t *testing.T) {
|
|
numJobs := 3
|
|
timeout := false
|
|
|
|
done := make(chan bool)
|
|
|
|
ctx := context.TODO()
|
|
nq, err := neoq.New(ctx, neoq.WithBackend(memory.Backend))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
h := handler.New(queue, func(_ context.Context) (err error) {
|
|
time.Sleep(100 * time.Millisecond)
|
|
return
|
|
}, handler.Concurrency(1), handler.MaxQueueCapacity(1))
|
|
|
|
if err := nq.Start(ctx, h); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
go func() {
|
|
for i := 0; i < numJobs; i++ {
|
|
jid, e := nq.Enqueue(ctx, &jobs.Job{
|
|
Queue: queue,
|
|
Payload: map[string]interface{}{
|
|
"message": fmt.Sprintf("hello world: %d", i),
|
|
},
|
|
})
|
|
if e != nil || jid == jobs.DuplicateJobID {
|
|
slog.Error("job was not enqueued. either it was duplicate or this error caused it:", "error", e)
|
|
}
|
|
}
|
|
|
|
done <- true
|
|
}()
|
|
|
|
select {
|
|
case <-time.After(time.Duration(50) * time.Millisecond):
|
|
timeout = true
|
|
case <-done:
|
|
err = errors.New("this job should have timed out")
|
|
}
|
|
|
|
if !timeout {
|
|
t.Error(err)
|
|
}
|
|
|
|
nq.Shutdown(ctx)
|
|
}
|
|
|
|
var testFutureJobs = &sync.Map{}
|
|
|
|
func TestFutureJobScheduling(t *testing.T) {
|
|
ctx := context.Background()
|
|
testBackend := memory.TestingBackend(
|
|
neoq.NewConfig(),
|
|
cron.New(),
|
|
&sync.Map{},
|
|
&sync.Map{},
|
|
testFutureJobs,
|
|
&sync.Map{},
|
|
slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: logging.LogLevelDebug})))
|
|
|
|
nq, err := neoq.New(ctx, neoq.WithBackend(testBackend))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer nq.Shutdown(ctx)
|
|
|
|
h := handler.New(queue, func(ctx context.Context) (err error) {
|
|
return
|
|
})
|
|
|
|
if err := nq.Start(ctx, h); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
jid, err := nq.Enqueue(ctx, &jobs.Job{
|
|
Queue: queue,
|
|
Payload: map[string]interface{}{
|
|
"message": "hello world",
|
|
},
|
|
RunAfter: time.Now().UTC().Add(5 * time.Second),
|
|
})
|
|
if err != nil || jid == jobs.DuplicateJobID {
|
|
err = fmt.Errorf("job was not enqueued. either it was duplicate or this error caused it: %w", err)
|
|
t.Error(err)
|
|
}
|
|
|
|
jobID, _ := strconv.ParseInt(jid, 0, 64)
|
|
var ok bool
|
|
if _, ok = testFutureJobs.Load(jobID); !ok {
|
|
t.Error(err)
|
|
}
|
|
}
|
|
|
|
// This test was added in response to the following issue: https://github.com/acaloiaro/neoq/issues/56
|
|
// nolint: gocognit, gocyclo
|
|
func TestFutureJobSchedulingMultipleQueues(t *testing.T) {
|
|
ctx := context.Background()
|
|
nq, err := neoq.New(ctx, neoq.WithBackend(memory.Backend))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer nq.Shutdown(ctx)
|
|
|
|
jobsPerQueueCount := 100
|
|
|
|
jobsProcessed1 := 0
|
|
jobsProcessed2 := 0
|
|
|
|
q1 := "queue1"
|
|
q2 := "queue2"
|
|
|
|
done1 := make(chan bool)
|
|
done2 := make(chan bool)
|
|
|
|
h1 := handler.New(q1, func(ctx context.Context) (err error) {
|
|
var j *jobs.Job
|
|
j, err = jobs.FromContext(ctx)
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
if j.Queue != q1 {
|
|
err = errors.New("handling job from queu2 with queue1 handler. this is a bug")
|
|
}
|
|
|
|
done1 <- true
|
|
return
|
|
})
|
|
|
|
h2 := handler.New(q2, func(ctx context.Context) (err error) {
|
|
var j *jobs.Job
|
|
j, err = jobs.FromContext(ctx)
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
if j.Queue != q2 {
|
|
err = errors.New("handling job from queue1 with queue2 handler. this is a bug")
|
|
}
|
|
|
|
done2 <- true
|
|
return
|
|
})
|
|
|
|
if err := nq.Start(ctx, h1); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
if err := nq.Start(ctx, h2); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
go func() {
|
|
for i := 1; i <= jobsPerQueueCount; i++ {
|
|
_, err = nq.Enqueue(ctx, &jobs.Job{
|
|
ID: int64(i),
|
|
Queue: q1,
|
|
Payload: map[string]interface{}{"ID": i},
|
|
RunAfter: time.Now().Add(1 * time.Millisecond),
|
|
})
|
|
if err != nil {
|
|
err = fmt.Errorf("job was not enqueued. either it was duplicate or this error caused it: %w", err)
|
|
t.Error(err)
|
|
}
|
|
}
|
|
|
|
for j := 1; j <= jobsPerQueueCount; j++ {
|
|
_, err = nq.Enqueue(ctx, &jobs.Job{
|
|
ID: int64(j),
|
|
Queue: q2,
|
|
Payload: map[string]interface{}{"ID": j},
|
|
RunAfter: time.Now().Add(1 * time.Millisecond),
|
|
})
|
|
if err != nil {
|
|
err = fmt.Errorf("job was not enqueued. either it was duplicate or this error caused it: %w", err)
|
|
t.Error(err)
|
|
}
|
|
}
|
|
}()
|
|
|
|
timeoutTimer := time.After(10 * time.Second)
|
|
results_loop:
|
|
for {
|
|
select {
|
|
case <-timeoutTimer:
|
|
err = jobs.ErrJobTimeout
|
|
break results_loop
|
|
case <-done1:
|
|
jobsProcessed1++
|
|
case <-done2:
|
|
jobsProcessed2++
|
|
default:
|
|
if jobsProcessed1 == jobsPerQueueCount && jobsProcessed2 == jobsPerQueueCount {
|
|
break results_loop
|
|
}
|
|
time.Sleep(10 * time.Millisecond)
|
|
}
|
|
}
|
|
if err != nil {
|
|
t.Error(err)
|
|
}
|
|
if jobsProcessed1 != jobsPerQueueCount {
|
|
// nolint: goerr113
|
|
t.Error(fmt.Errorf("handler1 should have handled %d jobs, but handled %d", jobsPerQueueCount, jobsProcessed1))
|
|
}
|
|
if jobsProcessed2 != jobsPerQueueCount {
|
|
// nolint: goerr113
|
|
t.Error(fmt.Errorf("handler2 should have handled %d jobs, but handled %d", jobsPerQueueCount, jobsProcessed2))
|
|
}
|
|
}
|
|
|
|
func TestCron(t *testing.T) {
|
|
const cronSpec = "* * * * * *"
|
|
ctx := context.TODO()
|
|
nq, err := neoq.New(ctx, neoq.WithBackend(memory.Backend))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer nq.Shutdown(ctx)
|
|
|
|
done := make(chan bool)
|
|
h := handler.New("foobar", func(ctx context.Context) (err error) {
|
|
done <- true
|
|
return
|
|
})
|
|
|
|
h.WithOptions(
|
|
handler.JobTimeout(500*time.Millisecond),
|
|
handler.Concurrency(1),
|
|
)
|
|
|
|
err = nq.StartCron(ctx, cronSpec, h)
|
|
if err != nil {
|
|
t.Error(err)
|
|
}
|
|
|
|
// allow time for listener to start
|
|
time.Sleep(5 * time.Millisecond)
|
|
|
|
select {
|
|
case <-time.After(1 * time.Second):
|
|
err = errPeriodicTimeout
|
|
case <-done:
|
|
}
|
|
|
|
if err != nil {
|
|
t.Error(err)
|
|
}
|
|
}
|
|
|
|
func TestMaxRetries(t *testing.T) {
|
|
const queue = "foobar"
|
|
maxRetries := 0
|
|
|
|
ctx := context.Background()
|
|
nq, err := neoq.New(ctx, neoq.WithBackend(memory.Backend), neoq.WithLogLevel(logging.LogLevelDebug))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer nq.Shutdown(ctx)
|
|
|
|
logChan := make(chan string, 1)
|
|
nq.SetLogger(testutils.NewTestLogger(logChan))
|
|
|
|
h := handler.New(queue, func(ctx context.Context) (err error) {
|
|
panic("i refuse success!")
|
|
})
|
|
|
|
err = nq.Start(ctx, h)
|
|
if err != nil {
|
|
t.Error(err)
|
|
}
|
|
|
|
// allow time for listener to start
|
|
time.Sleep(5 * time.Millisecond)
|
|
|
|
_, err = nq.Enqueue(ctx, &jobs.Job{
|
|
Queue: queue,
|
|
Payload: map[string]any{},
|
|
MaxRetries: &maxRetries,
|
|
})
|
|
if err != nil {
|
|
t.Error(err)
|
|
}
|
|
|
|
timeoutTimer := time.After(5 * time.Second)
|
|
|
|
result_loop:
|
|
for {
|
|
select {
|
|
case <-timeoutTimer:
|
|
err = errTimeout
|
|
break result_loop
|
|
case newLog := <-logChan:
|
|
if strings.Contains(newLog, "job exceeded max retries") {
|
|
break result_loop
|
|
}
|
|
}
|
|
}
|
|
|
|
if err != nil {
|
|
t.Error(err)
|
|
}
|
|
}
|