279 lines
10 KiB
Go
279 lines
10 KiB
Go
package mq
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"log/slog"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"kra/internal/integration/runtimeconfig"
|
|
platformmq "kra/pkg/mq"
|
|
)
|
|
|
|
type fakeClient struct {
|
|
subscribed []string
|
|
unsubscribed []string
|
|
handlers map[string]platformmq.Handler
|
|
subscribeErrors map[string]error
|
|
unsubscribeErrors map[string]error
|
|
}
|
|
|
|
func (*fakeClient) Publish(context.Context, string, []byte, byte, bool) error { return nil }
|
|
func (f *fakeClient) Subscribe(_ context.Context, topic string, _ byte, handler platformmq.Handler) error {
|
|
f.subscribed = append(f.subscribed, topic)
|
|
if err := f.subscribeErrors[topic]; err != nil {
|
|
return err
|
|
}
|
|
if f.handlers == nil {
|
|
f.handlers = make(map[string]platformmq.Handler)
|
|
}
|
|
f.handlers[topic] = handler
|
|
return nil
|
|
}
|
|
func (f *fakeClient) Unsubscribe(_ context.Context, topics ...string) error {
|
|
f.unsubscribed = append(f.unsubscribed, topics...)
|
|
for _, topic := range topics {
|
|
if err := f.unsubscribeErrors[topic]; err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
func (*fakeClient) Connected() bool { return true }
|
|
func (*fakeClient) Close() error { return nil }
|
|
|
|
func TestReloadableTracksSubscriptions(t *testing.T) {
|
|
client := &fakeClient{}
|
|
r := &Reloadable{
|
|
clients: map[string]platformmq.Client{ProviderEMQX: client},
|
|
configs: map[string]runtimeconfig.Config{ProviderEMQX: {Kind: "mq", Provider: ProviderEMQX, Enabled: true}},
|
|
subscriptions: make(map[string]map[string]map[string]subscription),
|
|
bindings: make(map[string]map[string]byte),
|
|
pending: make(map[string]bool),
|
|
nextRetry: make(map[string]time.Time),
|
|
}
|
|
handler := func(context.Context, platformmq.Message) {}
|
|
if err := r.Subscribe(context.Background(), "orders/+/paid", platformmq.AtLeastOnce, handler); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, ok := r.subscriptions[ProviderEMQX]["orders/+/paid"]; !ok {
|
|
t.Fatal("subscription was not retained for configuration reload")
|
|
}
|
|
if err := r.Unsubscribe(context.Background(), "orders/+/paid"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, ok := r.subscriptions[ProviderEMQX]["orders/+/paid"]; ok {
|
|
t.Fatal("unsubscribed topic remained in the reload registry")
|
|
}
|
|
}
|
|
|
|
func TestReloadableRestoresSubscriptions(t *testing.T) {
|
|
client := &fakeClient{}
|
|
r := &Reloadable{subscriptions: map[string]map[string]map[string]subscription{
|
|
ProviderEMQX: {
|
|
"orders/+/paid": {
|
|
"orders": {qos: platformmq.AtLeastOnce, handler: func(context.Context, platformmq.Message) {}},
|
|
},
|
|
},
|
|
}}
|
|
bindings, err := r.restoreSubscriptionsLocked(ProviderEMQX, client)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(client.subscribed) != 1 || client.subscribed[0] != "orders/+/paid" {
|
|
t.Fatalf("restored subscriptions = %v", client.subscribed)
|
|
}
|
|
if bindings["orders/+/paid"] != platformmq.AtLeastOnce {
|
|
t.Fatalf("restored binding qos = %d", bindings["orders/+/paid"])
|
|
}
|
|
}
|
|
|
|
func TestReloadableRegistersWhileOfflineAndRestoresLater(t *testing.T) {
|
|
client := &fakeClient{}
|
|
r := &Reloadable{
|
|
clients: make(map[string]platformmq.Client),
|
|
configs: map[string]runtimeconfig.Config{ProviderEMQX: {Kind: "mq", Provider: ProviderEMQX, Enabled: true}},
|
|
subscriptions: make(map[string]map[string]map[string]subscription),
|
|
bindings: make(map[string]map[string]byte),
|
|
pending: make(map[string]bool),
|
|
nextRetry: make(map[string]time.Time),
|
|
logger: slog.Default(),
|
|
}
|
|
called := make(chan struct{}, 1)
|
|
err := r.Register(platformmq.SubscriptionSet{
|
|
Owner: "orders",
|
|
Provider: ProviderEMQX,
|
|
Topics: []platformmq.TopicSubscription{{Topic: "orders.created", QoS: platformmq.AtLeastOnce, Handler: func(context.Context, platformmq.Message) { called <- struct{}{} }}},
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !r.pending[ProviderEMQX] {
|
|
t.Fatal("offline subscription was not marked pending")
|
|
}
|
|
r.clients[ProviderEMQX] = client
|
|
if err = r.reconcileProviderLocked(ProviderEMQX); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(client.subscribed) != 1 || client.subscribed[0] != "orders.created" {
|
|
t.Fatalf("restored subscriptions = %v", client.subscribed)
|
|
}
|
|
}
|
|
|
|
func TestReloadableDispatchesSameTopicToMultipleOwners(t *testing.T) {
|
|
client := &fakeClient{}
|
|
r := &Reloadable{
|
|
clients: map[string]platformmq.Client{ProviderEMQX: client},
|
|
configs: map[string]runtimeconfig.Config{ProviderEMQX: {Kind: "mq", Provider: ProviderEMQX, Enabled: true}},
|
|
subscriptions: make(map[string]map[string]map[string]subscription),
|
|
bindings: make(map[string]map[string]byte),
|
|
pending: make(map[string]bool),
|
|
nextRetry: make(map[string]time.Time),
|
|
logger: slog.Default(),
|
|
}
|
|
first, second := make(chan struct{}, 1), make(chan struct{}, 1)
|
|
for _, set := range []platformmq.SubscriptionSet{
|
|
{Owner: "orders", Provider: ProviderEMQX, Topics: []platformmq.TopicSubscription{{Topic: "events.created", Handler: func(context.Context, platformmq.Message) { first <- struct{}{} }}}},
|
|
{Owner: "audit", Provider: ProviderEMQX, Topics: []platformmq.TopicSubscription{{Topic: "events.created", Handler: func(context.Context, platformmq.Message) { second <- struct{}{} }}}},
|
|
} {
|
|
if err := r.Register(set); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
if err := r.reconcileProviderLocked(ProviderEMQX); err != nil {
|
|
t.Fatalf("explicit reconcile failed: %v", err)
|
|
}
|
|
if len(client.subscribed) != 1 {
|
|
t.Fatalf("broker subscriptions = %v, want one shared topic; desired=%#v bindings=%#v clients=%#v", client.subscribed, r.subscriptions, r.bindings, r.clients)
|
|
}
|
|
client.handlers["events.created"](context.Background(), platformmq.Message{Topic: "events.created"})
|
|
select {
|
|
case <-first:
|
|
default:
|
|
t.Fatal("orders handler was not called")
|
|
}
|
|
select {
|
|
case <-second:
|
|
default:
|
|
t.Fatal("audit handler was not called")
|
|
}
|
|
}
|
|
|
|
func TestReloadablePersistsPartialBindingChangesWhenReconcileFails(t *testing.T) {
|
|
client := &fakeClient{subscribeErrors: map[string]error{"events.new": errors.New("subscribe failed")}}
|
|
r := &Reloadable{
|
|
clients: map[string]platformmq.Client{ProviderEMQX: client},
|
|
subscriptions: map[string]map[string]map[string]subscription{
|
|
ProviderEMQX: {
|
|
"events.keep": {"owner": {qos: platformmq.AtMostOnce, handler: func(context.Context, platformmq.Message) {}}},
|
|
"events.new": {"owner": {qos: platformmq.AtMostOnce, handler: func(context.Context, platformmq.Message) {}}},
|
|
},
|
|
},
|
|
bindings: map[string]map[string]byte{ProviderEMQX: {
|
|
"events.old": platformmq.AtMostOnce,
|
|
"events.keep": platformmq.AtMostOnce,
|
|
}},
|
|
pending: make(map[string]bool),
|
|
nextRetry: make(map[string]time.Time),
|
|
}
|
|
|
|
if err := r.reconcileProviderLocked(ProviderEMQX); err == nil {
|
|
t.Fatal("reconcileProviderLocked() error = nil, want subscribe failure")
|
|
}
|
|
if _, exists := r.bindings[ProviderEMQX]["events.old"]; exists {
|
|
t.Fatal("successfully removed binding remained after failed reconcile")
|
|
}
|
|
if _, exists := r.bindings[ProviderEMQX]["events.keep"]; !exists {
|
|
t.Fatal("unchanged binding was lost after failed reconcile")
|
|
}
|
|
if _, exists := r.bindings[ProviderEMQX]["events.new"]; exists {
|
|
t.Fatal("failed subscription was recorded as bound")
|
|
}
|
|
}
|
|
|
|
func TestReloadableLegacySubscribeDoesNotLeaveOfflineDeclaration(t *testing.T) {
|
|
r := &Reloadable{
|
|
clients: make(map[string]platformmq.Client),
|
|
configs: map[string]runtimeconfig.Config{ProviderEMQX: {Kind: "mq", Provider: ProviderEMQX, Enabled: true}},
|
|
subscriptions: make(map[string]map[string]map[string]subscription),
|
|
bindings: make(map[string]map[string]byte),
|
|
pending: make(map[string]bool),
|
|
nextRetry: make(map[string]time.Time),
|
|
}
|
|
err := r.Subscribe(context.Background(), "events.offline", platformmq.AtLeastOnce, func(context.Context, platformmq.Message) {})
|
|
if !errors.Is(err, platformmq.ErrUnavailable) {
|
|
t.Fatalf("Subscribe() error = %v, want ErrUnavailable", err)
|
|
}
|
|
if len(r.subscriptions[ProviderEMQX]) != 0 {
|
|
t.Fatalf("offline legacy declaration remained: %#v", r.subscriptions[ProviderEMQX])
|
|
}
|
|
}
|
|
|
|
func TestReloadableExposesKafkaNamedClient(t *testing.T) {
|
|
r := &Reloadable{}
|
|
client := r.Client(" KAFKA ")
|
|
if client == nil {
|
|
t.Fatal("kafka named client is nil")
|
|
}
|
|
if client.Connected() {
|
|
t.Fatal("unconfigured kafka named client reported connected")
|
|
}
|
|
}
|
|
|
|
func TestConfigStringsDecodesJSONArrays(t *testing.T) {
|
|
values := map[string]any{"brokers": []any{" kafka-1:9092 ", "", "kafka-2:9092"}}
|
|
brokers := configStrings(values, "brokers")
|
|
if len(brokers) != 2 || brokers[0] != "kafka-1:9092" || brokers[1] != "kafka-2:9092" {
|
|
t.Fatalf("brokers = %#v", brokers)
|
|
}
|
|
}
|
|
|
|
func TestConfigStringsNormalizesTypedSlices(t *testing.T) {
|
|
values := map[string]any{"brokers": []string{" kafka-1:9092 ", "", "kafka-2:9092"}}
|
|
brokers := configStrings(values, "brokers")
|
|
if len(brokers) != 2 || brokers[0] != "kafka-1:9092" || brokers[1] != "kafka-2:9092" {
|
|
t.Fatalf("brokers = %#v", brokers)
|
|
}
|
|
}
|
|
|
|
func TestReloadableRejectsUnsupportedKafkaQoS(t *testing.T) {
|
|
r := &Reloadable{}
|
|
err := r.Register(platformmq.SubscriptionSet{
|
|
Owner: "orders",
|
|
Provider: ProviderKafka,
|
|
Topics: []platformmq.TopicSubscription{{Topic: "orders.created", QoS: platformmq.ExactlyOnce, Handler: func(context.Context, platformmq.Message) {}}},
|
|
})
|
|
if err == nil || !strings.Contains(err.Error(), "supports qos 0 or 1") {
|
|
t.Fatalf("Register() error = %v", err)
|
|
}
|
|
}
|
|
|
|
func TestRetryOnceHonorsPendingSubscriptionDeadline(t *testing.T) {
|
|
client := &fakeClient{}
|
|
r := &Reloadable{
|
|
clients: map[string]platformmq.Client{ProviderKafka: client},
|
|
configs: map[string]runtimeconfig.Config{
|
|
ProviderKafka: {Kind: "mq", Provider: ProviderKafka, Enabled: true},
|
|
},
|
|
subscriptions: map[string]map[string]map[string]subscription{
|
|
ProviderKafka: {
|
|
"orders.created": {"orders": {qos: platformmq.AtLeastOnce, handler: func(context.Context, platformmq.Message) {}}},
|
|
},
|
|
},
|
|
bindings: map[string]map[string]byte{ProviderKafka: {}},
|
|
pending: map[string]bool{ProviderKafka: true},
|
|
nextRetry: map[string]time.Time{ProviderKafka: time.Now().Add(time.Minute)},
|
|
}
|
|
r.retryOnce()
|
|
if len(client.subscribed) != 0 {
|
|
t.Fatalf("subscription retried before deadline: %v", client.subscribed)
|
|
}
|
|
r.nextRetry[ProviderKafka] = time.Now().Add(-time.Second)
|
|
r.retryOnce()
|
|
if len(client.subscribed) != 1 || client.subscribed[0] != "orders.created" {
|
|
t.Fatalf("subscription was not retried after deadline: %v", client.subscribed)
|
|
}
|
|
}
|