优化结构

2026-08-22 10:49:48 +08:00
parent 2b78b195c1
commit 54f76bd913
11 changed files with 501 additions and 98 deletions
@@ -402,7 +402,7 @@ func (r *Reloadable) clientLocked(provider string) platformmq.Client {
 	return r.clients[provider]
 }

-func (r *Reloadable) reconcileProviderLocked(provider string) error {
+func (r *Reloadable) reconcileProviderLocked(provider string) (err error) {
 	client := r.clientLocked(provider)
 	desired := r.desiredSubscriptions(provider)
 	if client == nil || !client.Connected() {
@@ -419,6 +419,10 @@ func (r *Reloadable) reconcileProviderLocked(provider string) error {
 		current[topic] = qos
 	}
 	r.mu.RUnlock()
+	// Broker operations can partially succeed. Persist every successful
+	// change even when a later operation fails, otherwise the next retry will
+	// repeat stale unbinds and may never reach the remaining subscriptions.
+	defer func() { r.setBindings(provider, current) }()
 	for topic := range current {
 		if _, exists := desired[topic]; exists {
 			continue
@@ -436,20 +440,31 @@ func (r *Reloadable) reconcileProviderLocked(provider string) error {
 			if err := client.Unsubscribe(context.Background(), topic); err != nil {
 				return err
 			}
+			delete(current, topic)
 		}
 		if err := client.Subscribe(context.Background(), topic, qos, r.dispatcher(provider, topic)); err != nil {
 			return err
 		}
 		current[topic] = qos
 	}
-	r.mu.Lock()
-	r.bindings[provider] = current
-	r.mu.Unlock()
 	delete(r.pending, provider)
 	delete(r.nextRetry, provider)
 	return nil
 }

+func (r *Reloadable) setBindings(provider string, bindings map[string]byte) {
+	copyOfBindings := make(map[string]byte, len(bindings))
+	for topic, qos := range bindings {
+		copyOfBindings[topic] = qos
+	}
+	r.mu.Lock()
+	if r.bindings == nil {
+		r.bindings = make(map[string]map[string]byte)
+	}
+	r.bindings[provider] = copyOfBindings
+	r.mu.Unlock()
+}
+
 func (r *Reloadable) Register(set platformmq.SubscriptionSet) error {
 	normalized, err := platformmq.NormalizeSubscriptionSet(set)
 	if err != nil {
@@ -526,6 +541,9 @@ func (r *Reloadable) Publish(ctx context.Context, topic string, payload []byte,
 }

 func (r *Reloadable) PublishTo(ctx context.Context, provider, topic string, payload []byte, qos byte, retain bool) error {
+	if ctx == nil {
+		ctx = context.Background()
+	}
 	provider = strings.ToLower(strings.TrimSpace(provider))
 	client := r.clientLocked(provider)
 	if client == nil {
@@ -539,6 +557,9 @@ func (r *Reloadable) Subscribe(ctx context.Context, topic string, qos byte, hand
 }

 func (r *Reloadable) SubscribeTo(ctx context.Context, provider, topic string, qos byte, handler platformmq.Handler) error {
+	if ctx == nil {
+		ctx = context.Background()
+	}
 	if ctx != nil {
 		select {
 		case <-ctx.Done():
@@ -555,6 +576,9 @@ func (r *Reloadable) SubscribeTo(ctx context.Context, provider, topic string, qo
 		return err
 	}
 	if !r.ConnectedTo(provider) {
+		// The legacy API reports unavailable when no live broker exists. Do not
+		// leave a durable declaration behind when that call failed.
+		_ = r.Unregister(owner)
 		return platformmq.ErrUnavailable
 	}
 	return nil
@@ -565,6 +589,9 @@ func (r *Reloadable) Unsubscribe(ctx context.Context, topics ...string) error {
 }

 func (r *Reloadable) UnsubscribeFrom(ctx context.Context, provider string, topics ...string) error {
+	if ctx == nil {
+		ctx = context.Background()
+	}
 	if ctx != nil {
 		select {
 		case <-ctx.Done():
@@ -660,22 +687,24 @@ func (r *Reloadable) Close() error {
 }

 func (r *Reloadable) ensureStateLocked() {
-	if r.clients == nil {
-		r.clients = make(map[string]platformmq.Client)
-	}
 	if r.configs == nil {
 		r.configs = make(map[string]runtimeconfig.Config)
 	}
-	if r.subscriptions == nil {
-		r.subscriptions = make(map[string]map[string]map[string]subscription)
-	}
-	if r.bindings == nil {
-		r.bindings = make(map[string]map[string]byte)
-	}
 	if r.pending == nil {
 		r.pending = make(map[string]bool)
 	}
 	if r.nextRetry == nil {
 		r.nextRetry = make(map[string]time.Time)
 	}
+	r.mu.Lock()
+	if r.clients == nil {
+		r.clients = make(map[string]platformmq.Client)
+	}
+	if r.subscriptions == nil {
+		r.subscriptions = make(map[string]map[string]map[string]subscription)
+	}
+	if r.bindings == nil {
+		r.bindings = make(map[string]map[string]byte)
+	}
+	r.mu.Unlock()
 }
@@ -2,6 +2,7 @@ package mq

 import (
 	"context"
+	"errors"
 	"log/slog"
 	"testing"
 	"time"
@@ -11,14 +12,19 @@ import (
 )

 type fakeClient struct {
-	subscribed   []string
-	unsubscribed []string
-	handlers     map[string]platformmq.Handler
+	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)
 	}
@@ -27,6 +33,11 @@ func (f *fakeClient) Subscribe(_ context.Context, topic string, _ byte, handler
 }
 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 }
@@ -148,3 +159,53 @@ func TestReloadableDispatchesSameTopicToMultipleOwners(t *testing.T) {
 		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])
+	}
+}