ae9a71034d
Add HTTP /v1 MVC, TCP send path, groups with server-side member-avatar collage, object-storage upload, and friends/favorites around the architecture-v3 layout. Co-authored-by: Cursor <cursoragent@cursor.com>
111 lines
2.7 KiB
Go
111 lines
2.7 KiB
Go
package logic
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"spark-server/internal/module/deliver/entity"
|
|
"spark-server/internal/module/deliver/model"
|
|
)
|
|
|
|
func TestMemoryQueueAckUsesOriginalBytes(t *testing.T) {
|
|
q := model.NewMemoryQueue()
|
|
ctx := t.Context()
|
|
job := &entity.Job{MsgID: 9, FromUserID: 1, ExcludeDevice: 7, ToUserIDs: []int64{2}}
|
|
if err := q.Push(ctx, job); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
got, err := q.Take(ctx, time.Second)
|
|
if err != nil || got == nil || got.MsgID != 9 {
|
|
t.Fatalf("take %+v %v", got, err)
|
|
}
|
|
got.FromUserID = 99
|
|
if err := q.Ack(ctx, got); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := q.RequeueStale(ctx); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
again, err := q.Take(ctx, 10*time.Millisecond)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if again != nil {
|
|
t.Fatalf("acked job still in processing %+v", again)
|
|
}
|
|
}
|
|
|
|
func TestDeduperSkipsSecond(t *testing.T) {
|
|
d := model.NewMemoryDone()
|
|
first, err := d.Mark(t.Context(), 11)
|
|
if err != nil || !first {
|
|
t.Fatalf("first %v %v", first, err)
|
|
}
|
|
second, err := d.Mark(t.Context(), 11)
|
|
if err != nil || second {
|
|
t.Fatalf("second %v %v", second, err)
|
|
}
|
|
}
|
|
|
|
func TestWorkerIdempotentAck(t *testing.T) {
|
|
q := model.NewMemoryQueue()
|
|
d := model.NewMemoryDone()
|
|
var n atomic.Int32
|
|
w := NewWorker(nil, q, d, WorkerOptions{
|
|
Workers: 1, TakeTimeout: 50 * time.Millisecond,
|
|
Handler: func(context.Context, *entity.Job) error { n.Add(1); return nil },
|
|
})
|
|
ctx := t.Context()
|
|
raw, _ := json.Marshal(&entity.Job{MsgID: 3, Cmd: 6})
|
|
if err := q.Push(ctx, &entity.Job{MsgID: 3}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_ = raw
|
|
w.Start(ctx)
|
|
defer w.Stop()
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for time.Now().Before(deadline) && n.Load() == 0 {
|
|
time.Sleep(10 * time.Millisecond)
|
|
}
|
|
if n.Load() != 1 {
|
|
t.Fatalf("handled %d", n.Load())
|
|
}
|
|
if err := q.Push(ctx, &entity.Job{MsgID: 3}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
time.Sleep(150 * time.Millisecond)
|
|
if n.Load() != 1 {
|
|
t.Fatalf("idempotent %d", n.Load())
|
|
}
|
|
}
|
|
|
|
func TestWorkerRecoversPanic(t *testing.T) {
|
|
q := model.NewMemoryQueue()
|
|
var n atomic.Int32
|
|
w := NewWorker(nil, q, model.NewMemoryDone(), WorkerOptions{
|
|
Workers: 1, TakeTimeout: 50 * time.Millisecond,
|
|
Handler: func(_ context.Context, job *entity.Job) error {
|
|
n.Add(1)
|
|
if job.MsgID == 1 {
|
|
panic("boom")
|
|
}
|
|
return nil
|
|
},
|
|
})
|
|
ctx := t.Context()
|
|
_ = q.Push(ctx, &entity.Job{MsgID: 1})
|
|
_ = q.Push(ctx, &entity.Job{MsgID: 2})
|
|
w.Start(ctx)
|
|
defer w.Stop()
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for time.Now().Before(deadline) && n.Load() < 2 {
|
|
time.Sleep(10 * time.Millisecond)
|
|
}
|
|
if n.Load() < 2 {
|
|
t.Fatalf("worker died after panic, handled %d", n.Load())
|
|
}
|
|
}
|