From 2b78b195c1beca260c1ab631d8cbcf65154d9314 Mon Sep 17 00:00:00 2001 From: Yvan <8574526@qq,com> Date: Sat, 22 Aug 2026 08:25:25 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E7=BB=93=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/biz/integration_config.go | 6 + internal/biz/integration_config_definition.go | 6 +- internal/integration/mq/emqx.go | 431 ++++++++++++++---- internal/integration/mq/emqx_test.go | 99 +++- internal/integration/provider.go | 1 + pkg/mq/mq.go | 17 +- pkg/mq/mqtt.go | 5 +- pkg/mq/rabbitmq.go | 72 ++- pkg/mq/subscription_test.go | 232 ++++++++++ .../view/systemTools/integration/config.vue | 61 ++- 10 files changed, 815 insertions(+), 115 deletions(-) create mode 100644 pkg/mq/subscription_test.go diff --git a/internal/biz/integration_config.go b/internal/biz/integration_config.go index 2bb1cfe..7431a5b 100644 --- a/internal/biz/integration_config.go +++ b/internal/biz/integration_config.go @@ -233,6 +233,9 @@ func validateCommunicationIntegrationConfig(kind, provider string, values map[st if timeout := integrationInt64(values, "connect_timeout", 0); timeout <= 0 { return errors.New("emqx connect_timeout 必须大于 0") } + if interval := integrationInt64(values, "reconnect_interval", 0); interval <= 0 { + return errors.New("emqx reconnect_interval 必须大于 0") + } case IntegrationKindMQ + "/rabbitmq": port := integrationInt64(values, "port", 0) if port < 1 || port > 65535 { @@ -251,6 +254,9 @@ func validateCommunicationIntegrationConfig(kind, provider string, values map[st if integrationInt64(values, "connect_timeout", 0) <= 0 { return errors.New("rabbitmq connect_timeout 必须大于 0") } + if integrationInt64(values, "reconnect_interval", 0) <= 0 { + return errors.New("rabbitmq reconnect_interval 必须大于 0") + } case IntegrationKindWebSocket + "/melody": path := integrationText(values, "path") if !strings.HasPrefix(path, "/") { diff --git a/internal/biz/integration_config_definition.go b/internal/biz/integration_config_definition.go index ab1339d..7d21365 100644 --- a/internal/biz/integration_config_definition.go +++ b/internal/biz/integration_config_definition.go @@ -92,7 +92,7 @@ var integrationDefinitions = map[string][]IntegrationConfigDefinition{ IntegrationKindMQ: { { Kind: IntegrationKindMQ, Provider: "emqx", Name: "EMQX", Description: "EMQX MQTT 消息服务", - Defaults: map[string]any{"broker": "tcp://127.0.0.1:1883", "client_id": "kra", "username": "", "password": "", "keep_alive": 30, "clean_session": true, "connect_timeout": 10}, + Defaults: map[string]any{"broker": "tcp://127.0.0.1:1883", "client_id": "kra", "username": "", "password": "", "keep_alive": 30, "clean_session": true, "connect_timeout": 10, "reconnect_interval": 5}, Fields: []IntegrationConfigField{ {Key: "broker", Label: "Broker 地址", Type: "text", Required: true, Placeholder: "tcp://127.0.0.1:1883"}, {Key: "client_id", Label: "客户端 ID", Type: "text", Required: true, Placeholder: "kra"}, @@ -101,11 +101,12 @@ var integrationDefinitions = map[string][]IntegrationConfigDefinition{ {Key: "keep_alive", Label: "心跳间隔（秒）", Type: "number", Required: true}, {Key: "clean_session", Label: "清理会话", Type: "switch", Description: "连接时不恢复 Broker 端保存的旧会话。"}, {Key: "connect_timeout", Label: "连接超时（秒）", Type: "number", Required: true}, + {Key: "reconnect_interval", Label: "重连退避上限（秒）", Type: "number", Required: true, Description: "网络中断后自动重连的最大退避间隔。"}, }, }, { Kind: IntegrationKindMQ, Provider: "rabbitmq", Name: "RabbitMQ", Description: "RabbitMQ AMQP 消息队列", - Defaults: map[string]any{"host": "127.0.0.1", "port": 5672, "username": "guest", "password": "guest", "vhost": "/", "exchange": "kra", "exchange_type": "topic", "queue": "kra", "routing_key": "#", "durable": true, "auto_delete": false, "prefetch_count": 10, "heartbeat": 10, "connect_timeout": 10, "tls": false}, + Defaults: map[string]any{"host": "127.0.0.1", "port": 5672, "username": "guest", "password": "guest", "vhost": "/", "exchange": "kra", "exchange_type": "topic", "queue": "kra", "routing_key": "#", "durable": true, "auto_delete": false, "prefetch_count": 10, "heartbeat": 10, "connect_timeout": 10, "reconnect_interval": 5, "tls": false}, Fields: []IntegrationConfigField{ {Key: "host", Label: "主机", Type: "text", Required: true, Placeholder: "127.0.0.1"}, {Key: "port", Label: "端口", Type: "number", Required: true}, @@ -121,6 +122,7 @@ var integrationDefinitions = map[string][]IntegrationConfigDefinition{ {Key: "prefetch_count", Label: "预取数量", Type: "number"}, {Key: "heartbeat", Label: "心跳间隔（秒）", Type: "number"}, {Key: "connect_timeout", Label: "连接超时（秒）", Type: "number", Required: true}, + {Key: "reconnect_interval", Label: "重连退避上限（秒）", Type: "number", Required: true, Description: "网络中断后自动重连的最大退避间隔。"}, {Key: "tls", Label: "启用 TLS", Type: "switch"}, }, }, diff --git a/internal/integration/mq/emqx.go b/internal/integration/mq/emqx.go index a8c9c1e..2a46be7 100644 --- a/internal/integration/mq/emqx.go +++ b/internal/integration/mq/emqx.go @@ -20,8 +20,8 @@ const ( retryTick = time.Second ) -// Reloadable owns the process-wide message clients. Configuration comes only -// from sys_integration_configs through runtimeconfig.Store. +// Reloadable owns process-wide message clients and the logical subscription +// declarations used to restore them after reconnects or configuration reloads. type Reloadable struct { mu sync.RWMutex opMu sync.Mutex @@ -31,10 +31,10 @@ type Reloadable struct { bindings map[string]map[string]byte pending map[string]bool nextRetry map[string]time.Time + legacySeq uint64 stop []func() retryStop chan struct{} retryDone chan struct{} - closeOnce sync.Once logger *slog.Logger closed bool } @@ -139,19 +139,22 @@ func (r *Reloadable) apply(provider string, config runtimeconfig.Config) { if r.closed { return } + r.ensureStateLocked() + config.Kind = "mq" config.Provider = provider config.Values = append(json.RawMessage(nil), config.Values...) r.configs[provider] = config + r.replaceClientLocked(provider, nil) if !config.Enabled { delete(r.pending, provider) delete(r.nextRetry, provider) - r.replaceClientLocked(provider, nil) return } if err := r.activateLocked(provider, config); err != nil { - r.pending[provider] = true - r.nextRetry[provider] = time.Now().Add(configRetryInterval(config.Values)) - r.logger.Warn("message integration unavailable", "mod", "mq", "provider", provider, "error", err) + r.scheduleRetryLocked(provider, config.Values) + if r.logger != nil { + r.logger.Warn("message integration unavailable", "mod", "mq", "provider", provider, "error", err) + } } } @@ -167,6 +170,9 @@ func (r *Reloadable) activateLocked(provider string, config runtimeconfig.Config } r.replaceClientLocked(provider, client) r.mu.Lock() + if r.bindings == nil { + r.bindings = make(map[string]map[string]byte) + } r.bindings[provider] = bindings r.mu.Unlock() delete(r.pending, provider) @@ -253,8 +259,89 @@ func configBool(values map[string]any, key string) bool { return value } +func (r *Reloadable) retryLoop() { + defer close(r.retryDone) + ticker := time.NewTicker(retryTick) + defer ticker.Stop() + for { + select { + case <-r.retryStop: + return + case <-ticker.C: + r.retryOnce() + } + } +} + +func (r *Reloadable) retryOnce() { + r.opMu.Lock() + defer r.opMu.Unlock() + if r.closed { + return + } + r.ensureStateLocked() + now := time.Now() + for _, provider := range []string{ProviderEMQX, ProviderRabbitMQ} { + config, exists := r.configs[provider] + if !exists || !config.Enabled { + continue + } + client := r.clientLocked(provider) + if client != nil && client.Connected() { + if r.pending[provider] { + if err := r.reconcileProviderLocked(provider); err != nil { + r.scheduleRetryLocked(provider, config.Values) + } + } + continue + } + if client != nil { + if recovering, ok := client.(interface{ Reconnecting() bool }); ok && recovering.Reconnecting() { + continue + } + } + if retryAt := r.nextRetry[provider]; !retryAt.IsZero() && now.Before(retryAt) { + continue + } + r.replaceClientLocked(provider, nil) + if err := r.activateLocked(provider, config); err != nil { + r.scheduleRetryLocked(provider, config.Values) + if r.logger != nil { + r.logger.Warn("message integration reconnect failed", "mod", "mq", "provider", provider, "error", err) + } + } + } +} + +func (r *Reloadable) scheduleRetryLocked(provider string, raw json.RawMessage) { + if r.pending == nil { + r.pending = make(map[string]bool) + } + if r.nextRetry == nil { + r.nextRetry = make(map[string]time.Time) + } + r.pending[provider] = true + r.nextRetry[provider] = time.Now().Add(configRetryInterval(raw)) +} + +func configRetryInterval(raw json.RawMessage) time.Duration { + values := map[string]any{} + _ = json.Unmarshal(raw, &values) + interval := configSeconds(values, "reconnect_interval") + if interval <= 0 { + return 5 * time.Second + } + return interval +} + func (r *Reloadable) replaceClientLocked(provider string, next platformmq.Client) { r.mu.Lock() + if r.clients == nil { + r.clients = make(map[string]platformmq.Client) + } + if r.bindings == nil { + r.bindings = make(map[string]map[string]byte) + } old := r.clients[provider] if next == nil { delete(r.clients, provider) @@ -263,7 +350,7 @@ func (r *Reloadable) replaceClientLocked(provider string, next platformmq.Client r.clients[provider] = next } r.mu.Unlock() - if old != nil { + if old != nil && old != next { _ = old.Close() } } @@ -280,23 +367,241 @@ func (r *Reloadable) restoreSubscriptionsLocked(provider string, client platform return bindings, nil } -func configRetryInterval(raw json.RawMessage) time.Duration { - values := map[string]any{} - _ = json.Unmarshal(raw, &values) - interval := configSeconds(values, "reconnect_interval") - if interval <= 0 { - return 5 * time.Second +func (r *Reloadable) desiredSubscriptions(provider string) map[string]byte { + result := make(map[string]byte) + r.mu.RLock() + defer r.mu.RUnlock() + for topic, owners := range r.subscriptions[provider] { + for _, item := range owners { + if qos, exists := result[topic]; !exists || item.qos > qos { + result[topic] = item.qos + } + } } - return interval + return result } -func (r *Reloadable) client(provider string) platformmq.Client { - provider = strings.ToLower(strings.TrimSpace(provider)) +func (r *Reloadable) dispatcher(provider, topic string) platformmq.Handler { + return func(ctx context.Context, message platformmq.Message) { + r.mu.RLock() + owners := r.subscriptions[provider][topic] + handlers := make([]platformmq.Handler, 0, len(owners)) + for _, item := range owners { + handlers = append(handlers, item.handler) + } + r.mu.RUnlock() + for _, handler := range handlers { + handler(ctx, message) + } + } +} + +func (r *Reloadable) clientLocked(provider string) platformmq.Client { r.mu.RLock() defer r.mu.RUnlock() return r.clients[provider] } +func (r *Reloadable) reconcileProviderLocked(provider string) error { + client := r.clientLocked(provider) + desired := r.desiredSubscriptions(provider) + if client == nil || !client.Connected() { + if len(desired) == 0 { + delete(r.pending, provider) + delete(r.nextRetry, provider) + return nil + } + return platformmq.ErrUnavailable + } + r.mu.RLock() + current := make(map[string]byte, len(r.bindings[provider])) + for topic, qos := range r.bindings[provider] { + current[topic] = qos + } + r.mu.RUnlock() + for topic := range current { + if _, exists := desired[topic]; exists { + continue + } + if err := client.Unsubscribe(context.Background(), topic); err != nil { + return err + } + delete(current, topic) + } + for topic, qos := range desired { + if oldQoS, exists := current[topic]; exists && oldQoS == qos { + continue + } + if _, exists := current[topic]; exists { + if err := client.Unsubscribe(context.Background(), topic); err != nil { + return err + } + } + 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) Register(set platformmq.SubscriptionSet) error { + normalized, err := platformmq.NormalizeSubscriptionSet(set) + if err != nil { + return err + } + r.opMu.Lock() + defer r.opMu.Unlock() + if r.closed { + return platformmq.ErrUnavailable + } + r.ensureStateLocked() + r.mu.Lock() + if r.subscriptions[normalized.Provider] == nil { + r.subscriptions[normalized.Provider] = make(map[string]map[string]subscription) + } + for topic, owners := range r.subscriptions[normalized.Provider] { + delete(owners, normalized.Owner) + if len(owners) == 0 { + delete(r.subscriptions[normalized.Provider], topic) + } + } + for _, item := range normalized.Topics { + if r.subscriptions[normalized.Provider][item.Topic] == nil { + r.subscriptions[normalized.Provider][item.Topic] = make(map[string]subscription) + } + r.subscriptions[normalized.Provider][item.Topic][normalized.Owner] = subscription{qos: item.QoS, handler: item.Handler} + } + r.mu.Unlock() + if err := r.reconcileProviderLocked(normalized.Provider); err != nil { + config := r.configs[normalized.Provider] + r.scheduleRetryLocked(normalized.Provider, config.Values) + if r.logger != nil { + r.logger.Warn("message subscription bind deferred", "mod", "mq", "provider", normalized.Provider, "owner", normalized.Owner, "error", err) + } + } + return nil +} + +func (r *Reloadable) Unregister(owner string) error { + owner = strings.TrimSpace(owner) + if owner == "" { + return fmt.Errorf("mq subscription owner is empty") + } + r.opMu.Lock() + defer r.opMu.Unlock() + if r.closed { + return nil + } + r.ensureStateLocked() + r.mu.Lock() + providers := make([]string, 0, len(r.subscriptions)) + for provider, topics := range r.subscriptions { + providers = append(providers, provider) + for topic, owners := range topics { + delete(owners, owner) + if len(owners) == 0 { + delete(topics, topic) + } + } + } + r.mu.Unlock() + for _, provider := range providers { + if err := r.reconcileProviderLocked(provider); err != nil { + if config, ok := r.configs[provider]; ok { + r.scheduleRetryLocked(provider, config.Values) + } + } + } + return nil +} + +func (r *Reloadable) Publish(ctx context.Context, topic string, payload []byte, qos byte, retain bool) error { + return r.PublishTo(ctx, ProviderEMQX, topic, payload, qos, retain) +} + +func (r *Reloadable) PublishTo(ctx context.Context, provider, topic string, payload []byte, qos byte, retain bool) error { + provider = strings.ToLower(strings.TrimSpace(provider)) + client := r.clientLocked(provider) + if client == nil { + return platformmq.ErrUnavailable + } + return client.Publish(ctx, topic, payload, qos, retain) +} + +func (r *Reloadable) Subscribe(ctx context.Context, topic string, qos byte, handler platformmq.Handler) error { + return r.SubscribeTo(ctx, ProviderEMQX, topic, qos, handler) +} + +func (r *Reloadable) SubscribeTo(ctx context.Context, provider, topic string, qos byte, handler platformmq.Handler) error { + if ctx != nil { + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + } + provider = strings.ToLower(strings.TrimSpace(provider)) + r.opMu.Lock() + r.legacySeq++ + owner := fmt.Sprintf("legacy/%s/%d", provider, r.legacySeq) + r.opMu.Unlock() + if err := r.Register(platformmq.SubscriptionSet{Owner: owner, Provider: provider, Topics: []platformmq.TopicSubscription{{Topic: topic, QoS: qos, Handler: handler}}}); err != nil { + return err + } + if !r.ConnectedTo(provider) { + return platformmq.ErrUnavailable + } + return nil +} + +func (r *Reloadable) Unsubscribe(ctx context.Context, topics ...string) error { + return r.UnsubscribeFrom(ctx, ProviderEMQX, topics...) +} + +func (r *Reloadable) UnsubscribeFrom(ctx context.Context, provider string, topics ...string) error { + if ctx != nil { + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + } + provider = strings.ToLower(strings.TrimSpace(provider)) + if len(topics) == 0 { + return fmt.Errorf("mq topics are empty") + } + r.opMu.Lock() + defer r.opMu.Unlock() + r.ensureStateLocked() + r.mu.Lock() + for _, rawTopic := range topics { + topic := strings.TrimSpace(rawTopic) + owners := r.subscriptions[provider][topic] + for owner := range owners { + if strings.HasPrefix(owner, "legacy/") { + delete(owners, owner) + } + } + if len(owners) == 0 { + delete(r.subscriptions[provider], topic) + } + } + r.mu.Unlock() + if err := r.reconcileProviderLocked(provider); err != nil { + if config, ok := r.configs[provider]; ok { + r.scheduleRetryLocked(provider, config.Values) + } + return err + } + return nil +} + func (r *Reloadable) Client(provider string) platformmq.Client { provider = strings.ToLower(strings.TrimSpace(provider)) if provider != ProviderEMQX && provider != ProviderRabbitMQ { @@ -317,81 +622,24 @@ func (c *namedClient) Unsubscribe(ctx context.Context, topics ...string) error { func (c *namedClient) Connected() bool { return c.owner.ConnectedTo(c.provider) } func (*namedClient) Close() error { return nil } -func (r *Reloadable) Publish(ctx context.Context, topic string, payload []byte, qos byte, retain bool) error { - return r.PublishTo(ctx, ProviderEMQX, topic, payload, qos, retain) -} - -func (r *Reloadable) PublishTo(ctx context.Context, provider, topic string, payload []byte, qos byte, retain bool) error { - provider = strings.ToLower(strings.TrimSpace(provider)) - r.mu.RLock() - defer r.mu.RUnlock() - client := r.clients[provider] - if client == nil { - return platformmq.ErrUnavailable - } - return client.Publish(ctx, topic, payload, qos, retain) -} - -func (r *Reloadable) Subscribe(ctx context.Context, topic string, qos byte, handler platformmq.Handler) error { - return r.SubscribeTo(ctx, ProviderEMQX, topic, qos, handler) -} - -func (r *Reloadable) SubscribeTo(ctx context.Context, provider, topic string, qos byte, handler platformmq.Handler) error { - provider = strings.ToLower(strings.TrimSpace(provider)) - r.opMu.Lock() - defer r.opMu.Unlock() - client := r.client(provider) - if client == nil { - return platformmq.ErrUnavailable - } - if err := client.Subscribe(ctx, topic, qos, handler); err != nil { - return err - } - if r.subscriptions[provider] == nil { - r.subscriptions[provider] = make(map[string]subscription) - } - r.subscriptions[provider][topic] = subscription{qos: qos, handler: handler} - return nil -} - -func (r *Reloadable) Unsubscribe(ctx context.Context, topics ...string) error { - return r.UnsubscribeFrom(ctx, ProviderEMQX, topics...) -} - -func (r *Reloadable) UnsubscribeFrom(ctx context.Context, provider string, topics ...string) error { - provider = strings.ToLower(strings.TrimSpace(provider)) - r.opMu.Lock() - defer r.opMu.Unlock() - client := r.client(provider) - if client == nil { - return platformmq.ErrUnavailable - } - if err := client.Unsubscribe(ctx, topics...); err != nil { - return err - } - for _, topic := range topics { - delete(r.subscriptions[provider], topic) - } - return nil -} - func (r *Reloadable) Connected() bool { return r.ConnectedTo(ProviderEMQX) } func (r *Reloadable) ConnectedTo(provider string) bool { provider = strings.ToLower(strings.TrimSpace(provider)) - r.mu.RLock() - defer r.mu.RUnlock() - client := r.clients[provider] + client := r.clientLocked(provider) return client != nil && client.Connected() } func (r *Reloadable) Close() error { r.opMu.Lock() - defer r.opMu.Unlock() if r.closed { + r.opMu.Unlock() return nil } r.closed = true + if r.retryStop != nil { + close(r.retryStop) + } r.mu.Lock() clients := make([]platformmq.Client, 0, len(r.clients)) for provider, client := range r.clients { @@ -399,6 +647,10 @@ func (r *Reloadable) Close() error { delete(r.clients, provider) } r.mu.Unlock() + r.opMu.Unlock() + if r.retryDone != nil { + <-r.retryDone + } for _, client := range clients { if client != nil { _ = client.Close() @@ -406,3 +658,24 @@ func (r *Reloadable) Close() error { } return nil } + +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) + } +} diff --git a/internal/integration/mq/emqx_test.go b/internal/integration/mq/emqx_test.go index 53615d3..f333c82 100644 --- a/internal/integration/mq/emqx_test.go +++ b/internal/integration/mq/emqx_test.go @@ -2,19 +2,27 @@ package mq import ( "context" + "log/slog" "testing" + "time" + "kra/internal/integration/runtimeconfig" platformmq "kra/pkg/mq" ) type fakeClient struct { subscribed []string unsubscribed []string + handlers map[string]platformmq.Handler } func (*fakeClient) Publish(context.Context, string, []byte, byte, bool) error { return nil } -func (f *fakeClient) Subscribe(_ context.Context, topic string, _ byte, _ platformmq.Handler) error { +func (f *fakeClient) Subscribe(_ context.Context, topic string, _ byte, handler platformmq.Handler) error { f.subscribed = append(f.subscribed, topic) + 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 { @@ -28,7 +36,11 @@ func TestReloadableTracksSubscriptions(t *testing.T) { client := &fakeClient{} r := &Reloadable{ clients: map[string]platformmq.Client{ProviderEMQX: client}, - subscriptions: make(map[string]map[string]subscription), + 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 { @@ -47,15 +59,92 @@ func TestReloadableTracksSubscriptions(t *testing.T) { func TestReloadableRestoresSubscriptions(t *testing.T) { client := &fakeClient{} - r := &Reloadable{subscriptions: map[string]map[string]subscription{ + r := &Reloadable{subscriptions: map[string]map[string]map[string]subscription{ ProviderEMQX: { - "orders/+/paid": {qos: platformmq.AtLeastOnce, handler: func(context.Context, platformmq.Message) {}}, + "orders/+/paid": { + "orders": {qos: platformmq.AtLeastOnce, handler: func(context.Context, platformmq.Message) {}}, + }, }, }} - if err := r.restoreSubscriptionsLocked(ProviderEMQX, client); err != nil { + 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") + } } diff --git a/internal/integration/provider.go b/internal/integration/provider.go index fb372a3..f719e3b 100644 --- a/internal/integration/provider.go +++ b/internal/integration/provider.go @@ -25,6 +25,7 @@ var ProviderSet = wire.NewSet( mqintegration.New, wire.Bind(new(mq.Client), new(*mqintegration.Reloadable)), wire.Bind(new(mq.Registry), new(*mqintegration.Reloadable)), + wire.Bind(new(mq.SubscriptionRegistrar), new(*mqintegration.Reloadable)), websocketintegration.New, wire.Bind(new(platformws.Hub), new(*websocketintegration.Server)), wire.Bind(new(biz.FileStorage), new(*storage.Reloadable)), diff --git a/pkg/mq/mq.go b/pkg/mq/mq.go index 9e2e6c2..eba48c6 100644 --- a/pkg/mq/mq.go +++ b/pkg/mq/mq.go @@ -83,14 +83,15 @@ type Registry interface { } type Config struct { - Enabled bool - Broker string - ClientID string - Username string - Password string - KeepAlive time.Duration - CleanSession bool - ConnectTimeout time.Duration + Enabled bool + Broker string + ClientID string + Username string + Password string + KeepAlive time.Duration + CleanSession bool + ConnectTimeout time.Duration + ReconnectInterval time.Duration } func PublishJSON(ctx context.Context, client Client, topic string, value any, qos byte, retain bool) error { diff --git a/pkg/mq/mqtt.go b/pkg/mq/mqtt.go index ccba0c8..e9c090e 100644 --- a/pkg/mq/mqtt.go +++ b/pkg/mq/mqtt.go @@ -28,9 +28,12 @@ func NewMQTT(cfg Config) (*MQTT, error) { if cfg.ConnectTimeout <= 0 { cfg.ConnectTimeout = 10 * time.Second } + if cfg.ReconnectInterval <= 0 { + cfg.ReconnectInterval = 5 * time.Second + } opts := paho.NewClientOptions().AddBroker(cfg.Broker).SetClientID(cfg.ClientID).SetUsername(cfg.Username).SetPassword(cfg.Password) opts.SetKeepAlive(cfg.KeepAlive).SetCleanSession(cfg.CleanSession).SetConnectTimeout(cfg.ConnectTimeout).SetAutoReconnect(true) - opts.SetResumeSubs(true).SetOrderMatters(false) + opts.SetMaxReconnectInterval(cfg.ReconnectInterval).SetResumeSubs(true).SetOrderMatters(false) c := &MQTT{} token := paho.NewClient(opts) connect := token.Connect() diff --git a/pkg/mq/rabbitmq.go b/pkg/mq/rabbitmq.go index 6a3fc62..574a94c 100644 --- a/pkg/mq/rabbitmq.go +++ b/pkg/mq/rabbitmq.go @@ -6,28 +6,30 @@ import ( "fmt" "strings" "sync" + "sync/atomic" "time" amqp "github.com/rabbitmq/amqp091-go" ) type RabbitMQConfig struct { - Enabled bool - Host string - Port int - Username string - Password string - VHost string - Exchange string - ExchangeType string - Queue string - RoutingKey string - Durable bool - AutoDelete bool - PrefetchCount int - Heartbeat time.Duration - ConnectTimeout time.Duration - TLS bool + Enabled bool + Host string + Port int + Username string + Password string + VHost string + Exchange string + ExchangeType string + Queue string + RoutingKey string + Durable bool + AutoDelete bool + PrefetchCount int + Heartbeat time.Duration + ConnectTimeout time.Duration + ReconnectInterval time.Duration + TLS bool } type rabbitSubscription struct { @@ -52,6 +54,7 @@ type RabbitMQ struct { consuming bool stop chan struct{} closed bool + reconnecting atomic.Bool } func NewRabbitMQ(config RabbitMQConfig) (*RabbitMQ, error) { @@ -75,7 +78,13 @@ func NewRabbitMQ(config RabbitMQConfig) (*RabbitMQ, error) { }.String() connection, err := amqp.DialConfig(address, amqp.Config{ Heartbeat: config.Heartbeat, - Recovery: &amqp.Recovery{}, + Recovery: &amqp.Recovery{ + ReconnectionConfig: &amqp.ReconnectionConfig{MaxRetryCount: 5, RetryInterval: config.ReconnectInterval}, + OnTopologyEntityError: func(_ *amqp.Connection, entity amqp.TopologyRecoveryEntity) bool { + amqp.Logger.Printf("rabbitmq topology recovery failed: %s", entity.Error()) + return false + }, + }, }) if err != nil { return nil, fmt.Errorf("connect rabbitmq: %w", err) @@ -116,6 +125,9 @@ func NewRabbitMQ(config RabbitMQConfig) (*RabbitMQ, error) { client.consumeChannel = consumeChannel client.config = config client.consumerTag = fmt.Sprintf("kra-%d", time.Now().UnixNano()) + stateChanges := make(chan *amqp.StateChanged, 16) + connection.NotifyStateChange(stateChanges) + go client.watchState(stateChanges) return client, nil } @@ -148,6 +160,9 @@ func defaultRabbitMQConfig(config RabbitMQConfig) RabbitMQConfig { if config.ConnectTimeout <= 0 { config.ConnectTimeout = 10 * time.Second } + if config.ReconnectInterval <= 0 { + config.ReconnectInterval = 5 * time.Second + } return config } @@ -371,7 +386,28 @@ func (c *RabbitMQ) Connected() bool { } c.mu.RLock() defer c.mu.RUnlock() - return !c.closed && c.connection != nil && !c.connection.IsClosed() + return !c.closed && !c.reconnecting.Load() && c.connection != nil && !c.connection.IsClosed() +} + +// Reconnecting reports whether the AMQP driver is actively recovering the +// current connection. The outer runtime waits for this state to settle before +// deciding whether a fresh client must be built. +func (c *RabbitMQ) Reconnecting() bool { + return c != nil && !c.closed && c.reconnecting.Load() +} + +func (c *RabbitMQ) watchState(states <-chan *amqp.StateChanged) { + for state := range states { + if state == nil { + continue + } + switch state.To { + case amqp.StateReconnecting: + c.reconnecting.Store(true) + case amqp.StateOpen, amqp.StateClosed: + c.reconnecting.Store(false) + } + } } func (c *RabbitMQ) Close() error { diff --git a/pkg/mq/subscription_test.go b/pkg/mq/subscription_test.go new file mode 100644 index 0000000..eb45f93 --- /dev/null +++ b/pkg/mq/subscription_test.go @@ -0,0 +1,232 @@ +package mq + +import ( + "context" + "errors" + "reflect" + "strings" + "testing" +) + +func TestNormalizeSubscriptionSetValidDeclaration(t *testing.T) { + handler := func(context.Context, Message) {} + set := SubscriptionSet{ + Owner: " orders ", + Provider: " RABBITMQ ", + Topics: []TopicSubscription{ + {Topic: " orders.created ", QoS: AtMostOnce, Handler: handler}, + {Topic: "orders.updated", QoS: AtLeastOnce, Handler: handler}, + {Topic: "orders.deleted", QoS: ExactlyOnce, Handler: handler}, + }, + } + + got, err := NormalizeSubscriptionSet(set) + if err != nil { + t.Fatalf("NormalizeSubscriptionSet() error = %v", err) + } + + want := SubscriptionSet{ + Owner: "orders", + Provider: ProviderRabbitMQ, + Topics: []TopicSubscription{ + {Topic: "orders.created", QoS: AtMostOnce, Handler: handler}, + {Topic: "orders.updated", QoS: AtLeastOnce, Handler: handler}, + {Topic: "orders.deleted", QoS: ExactlyOnce, Handler: handler}, + }, + } + if got.Owner != want.Owner || got.Provider != want.Provider { + t.Fatalf("normalized identity = %#v, want owner=%q provider=%q", got, want.Owner, want.Provider) + } + if len(got.Topics) != len(want.Topics) { + t.Fatalf("normalized topic count = %d, want %d", len(got.Topics), len(want.Topics)) + } + for index := range want.Topics { + if got.Topics[index].Topic != want.Topics[index].Topic { + t.Errorf("topic[%d] = %q, want %q", index, got.Topics[index].Topic, want.Topics[index].Topic) + } + if got.Topics[index].QoS != want.Topics[index].QoS { + t.Errorf("qos[%d] = %d, want %d", index, got.Topics[index].QoS, want.Topics[index].QoS) + } + if reflect.ValueOf(got.Topics[index].Handler).Pointer() != reflect.ValueOf(want.Topics[index].Handler).Pointer() { + t.Errorf("handler[%d] was changed", index) + } + } +} + +func TestNormalizeSubscriptionSetRejectsInvalidDeclarations(t *testing.T) { + handler := func(context.Context, Message) {} + base := func() SubscriptionSet { + return SubscriptionSet{ + Owner: "orders", + Provider: ProviderEMQX, + Topics: []TopicSubscription{ + {Topic: "orders.created", QoS: AtLeastOnce, Handler: handler}, + }, + } + } + + tests := []struct { + name string + set SubscriptionSet + wantErr string + }{ + { + name: "empty owner", + set: func() SubscriptionSet { set := base(); set.Owner = " "; return set }(), + wantErr: "owner is empty", + }, + { + name: "empty provider", + set: func() SubscriptionSet { set := base(); set.Provider = " "; return set }(), + wantErr: "unsupported mq provider", + }, + { + name: "unsupported provider", + set: func() SubscriptionSet { set := base(); set.Provider = "kafka"; return set }(), + wantErr: "unsupported mq provider", + }, + { + name: "empty topics", + set: func() SubscriptionSet { set := base(); set.Topics = nil; return set }(), + wantErr: "topics are empty", + }, + { + name: "empty topic", + set: func() SubscriptionSet { + set := base() + set.Topics[0].Topic = " " + return set + }(), + wantErr: "topic at index 0 is empty", + }, + { + name: "duplicate topic", + set: func() SubscriptionSet { + set := base() + set.Topics = append(set.Topics, TopicSubscription{ + Topic: " orders.created ", QoS: AtMostOnce, Handler: handler, + }) + return set + }(), + wantErr: "duplicate mq subscription topic", + }, + { + name: "nil handler", + set: func() SubscriptionSet { + set := base() + set.Topics[0].Handler = nil + return set + }(), + wantErr: "handler is nil", + }, + { + name: "invalid qos", + set: func() SubscriptionSet { + set := base() + set.Topics[0].QoS = ExactlyOnce + 1 + return set + }(), + wantErr: "invalid mq qos", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + _, err := NormalizeSubscriptionSet(test.set) + if err == nil { + t.Fatal("NormalizeSubscriptionSet() error = nil, want error") + } + if !strings.Contains(err.Error(), test.wantErr) { + t.Fatalf("error = %q, want substring %q", err, test.wantErr) + } + }) + } +} + +type subscriptionTestRegistrar struct { + sets []SubscriptionSet +} + +func (r *subscriptionTestRegistrar) Register(set SubscriptionSet) error { + r.sets = append(r.sets, set) + return nil +} + +func (*subscriptionTestRegistrar) Unregister(string) error { return nil } + +type subscriptionTestContributor struct { + name string + set SubscriptionSet + order *[]string + err error +} + +func (c *subscriptionTestContributor) RegisterSubscriptions(registrar SubscriptionRegistrar) error { + *c.order = append(*c.order, c.name) + if c.err != nil { + return c.err + } + return registrar.Register(c.set) +} + +func TestApplySubscriptionsAppliesMultipleContributorsInOrder(t *testing.T) { + order := make([]string, 0, 2) + registrar := &subscriptionTestRegistrar{} + handler := func(context.Context, Message) {} + first := &subscriptionTestContributor{ + name: "orders", + order: &order, + set: SubscriptionSet{ + Owner: "orders", + Provider: ProviderRabbitMQ, + Topics: []TopicSubscription{{Topic: "orders.created", Handler: handler}}, + }, + } + second := &subscriptionTestContributor{ + name: "notifications", + order: &order, + set: SubscriptionSet{ + Owner: "notifications", + Provider: ProviderEMQX, + Topics: []TopicSubscription{{Topic: "notifications.sent", Handler: handler}}, + }, + } + + if err := ApplySubscriptions(registrar, first, nil, second); err != nil { + t.Fatalf("ApplySubscriptions() error = %v", err) + } + if !reflect.DeepEqual(order, []string{"orders", "notifications"}) { + t.Fatalf("contributor order = %#v, want %#v", order, []string{"orders", "notifications"}) + } + if len(registrar.sets) != 2 { + t.Fatalf("registered sets = %d, want 2", len(registrar.sets)) + } + if registrar.sets[0].Owner != "orders" || registrar.sets[1].Owner != "notifications" { + t.Fatalf("registered owners = %q, %q", registrar.sets[0].Owner, registrar.sets[1].Owner) + } +} + +func TestApplySubscriptionsPropagatesContributorError(t *testing.T) { + order := make([]string, 0, 2) + registrar := &subscriptionTestRegistrar{} + wantErr := errors.New("registration failed") + failing := &subscriptionTestContributor{name: "failing", order: &order, err: wantErr} + following := &subscriptionTestContributor{name: "following", order: &order} + + err := ApplySubscriptions(registrar, failing, following) + if !errors.Is(err, wantErr) { + t.Fatalf("ApplySubscriptions() error = %v, want %v", err, wantErr) + } + if !reflect.DeepEqual(order, []string{"failing"}) { + t.Fatalf("contributors called = %#v, want %#v", order, []string{"failing"}) + } + if len(registrar.sets) != 0 { + t.Fatalf("registered sets = %d, want 0", len(registrar.sets)) + } +} + +func TestApplySubscriptionsRejectsNilRegistrar(t *testing.T) { + if err := ApplySubscriptions(nil); err == nil || !strings.Contains(err.Error(), "registrar is nil") { + t.Fatalf("ApplySubscriptions(nil) error = %v, want nil-registrar error", err) + } +} diff --git a/web/src/view/systemTools/integration/config.vue b/web/src/view/systemTools/integration/config.vue index e02158d..8a58134 100644 --- a/web/src/view/systemTools/integration/config.vue +++ b/web/src/view/systemTools/integration/config.vue @@ -116,6 +116,7 @@ class="field-control" :min="numberConstraint(field.key).min" :max="numberConstraint(field.key).max" + :precision="numberConstraint(field.key).integer ? 0 : undefined" :step="1" controls-position="right" @update:model-value="clearFieldError(selected, field.key)" @@ -231,10 +232,21 @@ const TARGETS = { } const TARGET_ORDER = Object.keys(TARGETS) +const RECONNECT_INTERVAL_KEY = 'reconnect_interval' +const RECONNECT_DEFAULT_SECONDS = 5 +const MQ_RECONNECT_TARGETS = new Set(['mq/emqx', 'mq/rabbitmq']) +const RECONNECT_INTERVAL_FIELD = { + key: RECONNECT_INTERVAL_KEY, + label: '重连间隔（秒）', + type: 'number', + required: true, + description: '连接中断后再次尝试连接的等待时间，最小 1 秒。' +} const NUMBER_CONSTRAINTS = { port: { min: 1, max: 65535 }, keep_alive: { min: 1 }, connect_timeout: { min: 1 }, + reconnect_interval: { min: 1, integer: true }, prefetch_count: { min: 0 }, heartbeat: { min: 0 }, max_message_size: { min: 0 }, @@ -264,13 +276,56 @@ const errorKey = (item, fieldKey) => `${integrationKey(item)}:${fieldKey}` const listKey = (item, fieldKey) => `${integrationKey(item)}:${fieldKey}` const cloneConfig = (value) => JSON.parse(JSON.stringify(value || {})) +const communicationFields = (item) => { + const fields = Array.isArray(item.fields) + ? item.fields.map((field) => + field.key === RECONNECT_INTERVAL_KEY + ? { + ...field, + label: field.label || RECONNECT_INTERVAL_FIELD.label, + type: 'number', + required: true, + description: field.description || RECONNECT_INTERVAL_FIELD.description + } + : { ...field } + ) + : [] + if ( + !MQ_RECONNECT_TARGETS.has(integrationKey(item)) || + fields.some((field) => field.key === RECONNECT_INTERVAL_KEY) + ) { + return fields + } + + const connectTimeoutIndex = fields.findIndex( + (field) => field.key === 'connect_timeout' + ) + fields.splice( + connectTimeoutIndex < 0 ? fields.length : connectTimeoutIndex + 1, + 0, + { ...RECONNECT_INTERVAL_FIELD } + ) + return fields +} + const normalizeIntegration = (item) => { + const config = cloneConfig(item.config) + const fields = communicationFields(item) + if ( + MQ_RECONNECT_TARGETS.has(integrationKey(item)) && + (config[RECONNECT_INTERVAL_KEY] === null || + typeof config[RECONNECT_INTERVAL_KEY] === 'undefined' || + config[RECONNECT_INTERVAL_KEY] === '') + ) { + config[RECONNECT_INTERVAL_KEY] = RECONNECT_DEFAULT_SECONDS + } + const normalized = { ...item, enabled: Boolean(item.enabled), configured: Boolean(item.configured), - config: cloneConfig(item.config), - fields: Array.isArray(item.fields) ? item.fields : [] + config, + fields } normalized._savedEnabled = normalized.enabled normalized._savedConfig = cloneConfig(normalized.config) @@ -359,6 +414,8 @@ const validate = (item, enabled = item.enabled) => { const constraint = numberConstraint(field.key) if (!Number.isFinite(number)) { message = `${field.label}必须是数字` + } else if (constraint.integer && !Number.isInteger(number)) { + message = `${field.label}必须是整数` } else if (constraint.min !== undefined && number < constraint.min) { message = `${field.label}不能小于 ${constraint.min}` } else if (constraint.max !== undefined && number > constraint.max) {