131 lines
3.4 KiB
Go
131 lines
3.4 KiB
Go
package outbox
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
// fakePublisher records deliveries and can be toggled to fail (broker down).
|
|
type fakePublisher struct {
|
|
mu sync.Mutex
|
|
fail bool
|
|
gotKR []string
|
|
}
|
|
|
|
func (f *fakePublisher) setFail(v bool) {
|
|
f.mu.Lock()
|
|
f.fail = v
|
|
f.mu.Unlock()
|
|
}
|
|
|
|
func (f *fakePublisher) Publish(_, routingKey string, body []byte) error {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
if f.fail {
|
|
return errors.New("broker down")
|
|
}
|
|
f.gotKR = append(f.gotKR, routingKey+":"+string(body))
|
|
return nil
|
|
}
|
|
|
|
func (f *fakePublisher) delivered() []string {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
out := make([]string, len(f.gotKR))
|
|
copy(out, f.gotKR)
|
|
return out
|
|
}
|
|
|
|
func waitFor(t *testing.T, cond func() bool) {
|
|
t.Helper()
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for time.Now().Before(deadline) {
|
|
if cond() {
|
|
return
|
|
}
|
|
time.Sleep(5 * time.Millisecond)
|
|
}
|
|
t.Fatal("condition not met within deadline")
|
|
}
|
|
|
|
func TestRelayDelivers(t *testing.T) {
|
|
ob, err := Open(t.TempDir() + "/outbox.log")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
pub := &fakePublisher{}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go ob.Run(ctx, pub, 20*time.Millisecond, nil)
|
|
|
|
_ = ob.Publish("", "order-events", []byte("e1"))
|
|
_ = ob.Publish("", "order-events", []byte("e2"))
|
|
|
|
waitFor(t, func() bool { return len(pub.delivered()) == 2 })
|
|
waitFor(t, func() bool { return ob.PendingCount() == 0 })
|
|
}
|
|
|
|
func TestBrokerDownThenRecovers(t *testing.T) {
|
|
ob, _ := Open(t.TempDir() + "/outbox.log")
|
|
pub := &fakePublisher{}
|
|
pub.setFail(true)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go ob.Run(ctx, pub, 20*time.Millisecond, nil)
|
|
|
|
_ = ob.Publish("", "k", []byte("held"))
|
|
// While the broker is down the message stays pending and undelivered.
|
|
waitFor(t, func() bool { return ob.PendingCount() == 1 })
|
|
if len(pub.delivered()) != 0 {
|
|
t.Fatal("nothing should be delivered while broker is down")
|
|
}
|
|
// Broker recovers — the relay drains it.
|
|
pub.setFail(false)
|
|
waitFor(t, func() bool { return len(pub.delivered()) == 1 && ob.PendingCount() == 0 })
|
|
}
|
|
|
|
func TestDurableAcrossRestart(t *testing.T) {
|
|
path := t.TempDir() + "/outbox.log"
|
|
ob, _ := Open(path)
|
|
// Enqueue durably but never run a relay — simulates a crash after the state
|
|
// change recorded the event but before it was delivered.
|
|
_ = ob.Publish("", "order-events", []byte("survivor"))
|
|
_ = ob.Close()
|
|
|
|
// Restart: the pending event must still be there and then deliver.
|
|
ob2, err := Open(path)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if ob2.PendingCount() != 1 {
|
|
t.Fatalf("pending after restart = %d, want 1", ob2.PendingCount())
|
|
}
|
|
pub := &fakePublisher{}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go ob2.Run(ctx, pub, 20*time.Millisecond, nil)
|
|
waitFor(t, func() bool { return len(pub.delivered()) == 1 })
|
|
}
|
|
|
|
func TestAckedNotRedelivered(t *testing.T) {
|
|
path := t.TempDir() + "/outbox.log"
|
|
ob, _ := Open(path)
|
|
pub := &fakePublisher{}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
go ob.Run(ctx, pub, 20*time.Millisecond, nil)
|
|
_ = ob.Publish("", "k", []byte("once"))
|
|
waitFor(t, func() bool { return ob.PendingCount() == 0 })
|
|
cancel()
|
|
_ = ob.Close()
|
|
|
|
// Reopen (compaction drops the acked entry); nothing should redeliver.
|
|
ob2, _ := Open(path)
|
|
defer ob2.Close()
|
|
if ob2.PendingCount() != 0 {
|
|
t.Fatalf("acked message redelivered after restart: pending=%d", ob2.PendingCount())
|
|
}
|
|
}
|