103 lines
3.2 KiB
Go
103 lines
3.2 KiB
Go
package task
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestCapacityDrainPanicAndClose(t *testing.T) {
|
|
r, err := New(Config{Workers: 1, QueueSize: 1, OnError: func(string, error) { panic("callback failure") }})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer r.Close()
|
|
started, release := make(chan struct{}), make(chan struct{})
|
|
first, _ := r.Submit(context.Background(), "first", func(context.Context) error { close(started); <-release; return nil })
|
|
<-started
|
|
second, _ := r.Submit(context.Background(), "panic", func(context.Context) error { panic("job failure") })
|
|
if _, err := r.Submit(context.Background(), "overflow", func(context.Context) error { return nil }); !errors.Is(err, ErrQueueFull) {
|
|
t.Fatal(err)
|
|
}
|
|
done := make(chan error, 1)
|
|
go func() { done <- r.Close() }()
|
|
close(release)
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
|
defer cancel()
|
|
if err := first.Wait(ctx); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := second.Wait(ctx); err == nil {
|
|
t.Fatal("panic lost")
|
|
}
|
|
if err := <-done; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if s := r.Stats(); s.Completed != 2 || s.Failed != 1 || s.Active != 0 || s.Queued != 0 {
|
|
t.Fatal(s)
|
|
}
|
|
if _, err := r.Submit(ctx, "late", func(context.Context) error { return nil }); !errors.Is(err, ErrClosed) {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func TestShutdownDeadlineAndQueuedCancellation(t *testing.T) {
|
|
r, _ := New(Config{Workers: 1, QueueSize: 1, ShutdownTimeout: 20 * time.Millisecond, OnError: func(string, error) {}})
|
|
started, release := make(chan struct{}), make(chan struct{})
|
|
var queued atomic.Bool
|
|
first, _ := r.Submit(context.Background(), "ignoresCancel", func(context.Context) error { close(started); <-release; return nil })
|
|
<-started
|
|
second, _ := r.Submit(context.Background(), "queued", func(context.Context) error { queued.Store(true); return nil })
|
|
if err := r.Close(); !errors.Is(err, context.DeadlineExceeded) {
|
|
t.Fatal(err)
|
|
}
|
|
close(release)
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
|
defer cancel()
|
|
if err := first.Wait(ctx); !errors.Is(err, context.Canceled) {
|
|
t.Fatal(err)
|
|
}
|
|
if err := second.Wait(ctx); !errors.Is(err, context.Canceled) {
|
|
t.Fatal(err)
|
|
}
|
|
if queued.Load() {
|
|
t.Fatal("ran queued work after forced shutdown")
|
|
}
|
|
if err := r.Shutdown(ctx); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func TestIntervalsDoNotOverlapAndTaskDeadline(t *testing.T) {
|
|
r, _ := New(Config{Workers: 2, TaskTimeout: 25 * time.Millisecond, OnError: func(string, error) {}})
|
|
defer r.Close()
|
|
var active, maxActive atomic.Int32
|
|
job := func(ctx context.Context) error {
|
|
n := active.Add(1)
|
|
if n > maxActive.Load() {
|
|
maxActive.Store(n)
|
|
}
|
|
defer active.Add(-1)
|
|
<-ctx.Done()
|
|
return ctx.Err()
|
|
}
|
|
if err := r.Every("periodic", 3*time.Millisecond, job); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := r.Every("periodic", time.Second, job); err == nil {
|
|
t.Fatal("duplicate schedule accepted")
|
|
}
|
|
deadline := time.Now().Add(time.Second)
|
|
for r.Stats().Completed < 2 && time.Now().Before(deadline) {
|
|
time.Sleep(2 * time.Millisecond)
|
|
}
|
|
if err := r.Close(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if maxActive.Load() != 1 || r.Stats().Skipped == 0 || r.Stats().Failed < 2 {
|
|
t.Fatal(maxActive.Load(), r.Stats())
|
|
}
|
|
}
|