101 lines
3.0 KiB
Go
101 lines
3.0 KiB
Go
package bus
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.sendico.io/tech/sxchnge/notifications/model"
|
|
"git.sendico.io/tech/sxchnge/shared/database/record"
|
|
merrors "git.sendico.io/tech/sxchnge/shared/errors"
|
|
"git.sendico.io/tech/sxchnge/shared/messaging/reliable"
|
|
"go.mongodb.org/mongo-driver/v2/bson"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
type outbox struct{ jobs []model.Notification }
|
|
|
|
func (o *outbox) Due(context.Context, time.Time) ([]model.Notification, error) { return o.jobs, nil }
|
|
func (o *outbox) Recover(context.Context, time.Time) error { return nil }
|
|
|
|
type publisher struct {
|
|
reliable.Transport
|
|
fail bool
|
|
records []reliable.Record
|
|
}
|
|
|
|
type consumerTransport struct {
|
|
reliable.Transport
|
|
reliable.Subscription
|
|
ctx context.Context
|
|
}
|
|
|
|
func (t *consumerTransport) OpenConsumer(ctx context.Context, _ reliable.DurableSettings, _ reliable.DurableConsumerSettings) (reliable.Subscription, error) {
|
|
t.ctx = ctx
|
|
return t.Subscription, nil
|
|
}
|
|
|
|
type consumerAdmin struct {
|
|
reliable.Admin
|
|
ctx context.Context
|
|
}
|
|
|
|
func (a *consumerAdmin) EnsureConsumerResources(ctx context.Context, _ reliable.DurableSettings, _ reliable.DurableConsumerSettings) error {
|
|
a.ctx = ctx
|
|
return nil
|
|
}
|
|
|
|
func TestSubscriptionOutlivesProvisioning(t *testing.T) {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
transport := &consumerTransport{}
|
|
admin := &consumerAdmin{}
|
|
if _, err := New(ctx, transport, admin, &outbox{}, nil, zap.NewNop(), DefaultConfig()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if admin.ctx.Err() != context.Canceled {
|
|
t.Fatal("provisioning context was not released")
|
|
}
|
|
if _, ok := admin.ctx.Deadline(); !ok {
|
|
t.Fatal("provisioning has no deadline")
|
|
}
|
|
if err := transport.ctx.Err(); err != nil {
|
|
t.Fatalf("subscription ended with provisioning: %v", err)
|
|
}
|
|
cancel()
|
|
if transport.ctx.Err() != context.Canceled {
|
|
t.Fatal("subscription does not follow service shutdown")
|
|
}
|
|
}
|
|
|
|
func (p *publisher) Publish(_ context.Context, r reliable.Record) error {
|
|
if p.fail {
|
|
return merrors.ServiceUnavailable("offline")
|
|
}
|
|
p.records = append(p.records, r)
|
|
return nil
|
|
}
|
|
func TestOutboxSurvivesPublishFailure(t *testing.T) {
|
|
id := bson.NewObjectID()
|
|
out := &outbox{jobs: []model.Notification{{Base: storable.Base{ID: id}, Email: model.Email{Address: "private@example.com", Text: "private"}}}}
|
|
transport := &publisher{fail: true}
|
|
b := &Bus{outbox: out, transport: transport}
|
|
if err := b.PublishDue(context.Background()); err == nil {
|
|
t.Fatal("publication failure hidden")
|
|
}
|
|
transport.fail = false
|
|
if err := b.PublishDue(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(transport.records) != 1 || string(transport.records[0].Payload) != id.Hex() || transport.records[0].MessageID != id.Hex()+"-0" {
|
|
t.Fatal(transport.records)
|
|
}
|
|
out.jobs[0].Attempts = 1
|
|
if err := b.PublishDue(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if transport.records[1].MessageID == transport.records[0].MessageID {
|
|
t.Fatal("retry shares transport deduplication identity")
|
|
}
|
|
}
|