130 lines
4.3 KiB
Go
130 lines
4.3 KiB
Go
package resultoutbox
|
|
|
|
import (
|
|
"context"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"cmpp-platform/gateway/internal/queue"
|
|
|
|
"github.com/alicebob/miniredis/v2"
|
|
"github.com/redis/go-redis/v9"
|
|
)
|
|
|
|
func TestPublishSubmitSegmentIsIdempotent(t *testing.T) {
|
|
mr := miniredis.RunT(t)
|
|
client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
|
|
outbox := New(client)
|
|
command := testCommand()
|
|
segment := queue.SubmitSegmentResult{SegmentTotal: 1, SegmentIndex: 1, SequenceID: 7, GatewayMessageID: "88", SubmitStatus: "accepted"}
|
|
|
|
if err := outbox.PublishSubmitSegment(context.Background(), command, segment); err != nil {
|
|
t.Fatalf("first publish: %v", err)
|
|
}
|
|
if err := outbox.PublishSubmitSegment(context.Background(), command, segment); err != nil {
|
|
t.Fatalf("duplicate publish: %v", err)
|
|
}
|
|
if got := client.XLen(context.Background(), outbox.StreamName()).Val(); got != 1 {
|
|
t.Fatalf("stream length = %d, want 1", got)
|
|
}
|
|
}
|
|
|
|
func TestPublishAggregateAndCommandAckAreAtomicAndIdempotent(t *testing.T) {
|
|
mr := miniredis.RunT(t)
|
|
client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
|
|
ctx := context.Background()
|
|
commandStream := "gateway.submit.commands"
|
|
commandGroup := "cmpp-gateway"
|
|
if err := client.XGroupCreateMkStream(ctx, commandStream, commandGroup, "0").Err(); err != nil {
|
|
t.Fatalf("create command group: %v", err)
|
|
}
|
|
commandID, err := client.XAdd(ctx, &redis.XAddArgs{Stream: commandStream, Values: map[string]interface{}{"data": "command"}}).Result()
|
|
if err != nil {
|
|
t.Fatalf("add command: %v", err)
|
|
}
|
|
if _, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{Group: commandGroup, Consumer: "gateway-1", Streams: []string{commandStream, ">"}, Count: 1}).Result(); err != nil {
|
|
t.Fatalf("claim command: %v", err)
|
|
}
|
|
outbox := New(client)
|
|
command := testCommand()
|
|
result := queue.SubmitResult{Envelope: command.Envelope, SubmitID: command.SubmitID, GatewayMessageID: "99", SubmitStatus: "accepted"}
|
|
|
|
for attempt := 0; attempt < 2; attempt++ {
|
|
if err := outbox.PublishSubmitResultAndAck(ctx, commandStream, commandGroup, commandID, command, result); err != nil {
|
|
t.Fatalf("publish attempt %d: %v", attempt+1, err)
|
|
}
|
|
}
|
|
pending, err := client.XPending(ctx, commandStream, commandGroup).Result()
|
|
if err != nil {
|
|
t.Fatalf("command pending: %v", err)
|
|
}
|
|
if pending.Count != 0 {
|
|
t.Fatalf("command pending = %d, want 0", pending.Count)
|
|
}
|
|
if got := client.XLen(ctx, outbox.StreamName()).Val(); got != 1 {
|
|
t.Fatalf("result stream length = %d, want 1", got)
|
|
}
|
|
}
|
|
|
|
func TestCallbackWorkerRetriesAndOnlyDeletesAfterSuccess(t *testing.T) {
|
|
mr := miniredis.RunT(t)
|
|
client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
|
|
var calls atomic.Int32
|
|
server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
|
|
if request.Header.Get("X-CMPP-Result-Event-ID") == "" {
|
|
t.Error("missing result event id header")
|
|
}
|
|
if calls.Add(1) == 1 {
|
|
http.Error(response, "temporary failure", http.StatusServiceUnavailable)
|
|
return
|
|
}
|
|
response.WriteHeader(http.StatusCreated)
|
|
}))
|
|
defer server.Close()
|
|
|
|
outbox := New(client)
|
|
outbox.APIBaseURL = server.URL
|
|
outbox.MinIdle = 10 * time.Millisecond
|
|
outbox.Concurrency = 1
|
|
command := testCommand()
|
|
if err := outbox.PublishSubmitSegment(context.Background(), command, queue.SubmitSegmentResult{
|
|
SegmentTotal: 1, SegmentIndex: 1, SequenceID: 7, GatewayMessageID: "88", SubmitStatus: "accepted",
|
|
}); err != nil {
|
|
t.Fatalf("publish segment: %v", err)
|
|
}
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan error, 1)
|
|
go func() { done <- outbox.Run(ctx) }()
|
|
deadline := time.Now().Add(5 * time.Second)
|
|
for calls.Load() < 2 || client.XLen(context.Background(), outbox.StreamName()).Val() != 0 {
|
|
if time.Now().After(deadline) {
|
|
t.Fatalf("calls=%d streamLength=%d", calls.Load(), client.XLen(context.Background(), outbox.StreamName()).Val())
|
|
}
|
|
time.Sleep(10 * time.Millisecond)
|
|
}
|
|
cancel()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("worker did not stop")
|
|
}
|
|
}
|
|
|
|
func testCommand() queue.SubmitCommand {
|
|
return queue.SubmitCommand{
|
|
Envelope: queue.Envelope{
|
|
SchemaVersion: queue.SchemaVersion,
|
|
MessageType: queue.MessageTypeSubmitCommand,
|
|
TraceID: "trace-1",
|
|
MessageID: "message-1",
|
|
ChannelID: "channel-1",
|
|
CreatedAt: time.Now().UTC(),
|
|
},
|
|
SubmitID: "submit-1",
|
|
}
|
|
}
|