优化结构

2026-08-21 21:37:30 +08:00
parent a2ce3ae218
commit e4ba0dced9
41 changed files with 3646 additions and 364 deletions
@@ -2,140 +2,294 @@ package mq

 import (
 	"context"
+	"encoding/json"
+	"fmt"
 	"log/slog"
+	"strconv"
+	"strings"
 	"sync"
 	"time"

-	"kra/internal/conf"
-	"kra/pkg/mq"
+	"kra/internal/integration/runtimeconfig"
+	platformmq "kra/pkg/mq"
 )

-// Reloadable follows the runtime snapshot and keeps a single shared EMQX
-// connection for all modules in this process.
+const (
+	ProviderEMQX     = "emqx"
+	ProviderRabbitMQ = "rabbitmq"
+)
+
+// Reloadable owns the process-wide message clients. Configuration comes only
+// from sys_integration_configs through runtimeconfig.Store.
 type Reloadable struct {
 	mu            sync.RWMutex
 	opMu          sync.Mutex
-	current       mq.Client
-	subscriptions map[string]subscription
-	stop          func()
+	clients       map[string]platformmq.Client
+	subscriptions map[string]map[string]subscription
+	stop          []func()
 	logger        *slog.Logger
 	closed        bool
 }

 type subscription struct {
 	qos     byte
-	handler mq.Handler
+	handler platformmq.Handler
 }

-func New(runtime *conf.Runtime, logger *slog.Logger) (*Reloadable, func(), error) {
+type namedClient struct {
+	owner    *Reloadable
+	provider string
+}
+
+func New(store *runtimeconfig.Store, logger *slog.Logger) (*Reloadable, func(), error) {
 	if logger == nil {
 		logger = slog.Default()
 	}
-	r := &Reloadable{logger: logger, subscriptions: make(map[string]subscription)}
-	if runtime != nil {
-		var config *conf.AdminBackend_MQ
-		if admin := runtime.Admin(); admin != nil {
-			config = admin.GetMq()
-		}
-		r.replace(config)
-		r.stop = runtime.Subscribe(func(_ *conf.Data, admin *conf.AdminBackend) {
-			if admin != nil {
-				r.replace(admin.GetMq())
-			}
-		})
+	r := &Reloadable{
+		clients:       make(map[string]platformmq.Client),
+		subscriptions: make(map[string]map[string]subscription),
+		logger:        logger,
+	}
+	if store != nil {
+		r.apply(ProviderEMQX, storeConfig(store, ProviderEMQX))
+		r.apply(ProviderRabbitMQ, storeConfig(store, ProviderRabbitMQ))
+		r.stop = append(r.stop,
+			store.Subscribe("mq", ProviderEMQX, func(config runtimeconfig.Config) { r.apply(ProviderEMQX, config) }),
+			store.Subscribe("mq", ProviderRabbitMQ, func(config runtimeconfig.Config) { r.apply(ProviderRabbitMQ, config) }),
+		)
 	}
 	cleanup := func() {
-		if r.stop != nil {
-			r.stop()
+		for _, stop := range r.stop {
+			stop()
 		}
 		_ = r.Close()
 	}
 	return r, cleanup, nil
 }

-func (r *Reloadable) replace(config *conf.AdminBackend_MQ) {
+func storeConfig(store *runtimeconfig.Store, provider string) runtimeconfig.Config {
+	config, _ := store.Get("mq", provider)
+	return config
+}
+
+func (r *Reloadable) apply(provider string, config runtimeconfig.Config) {
 	r.opMu.Lock()
 	defer r.opMu.Unlock()
 	if r.closed {
 		return
 	}
-	if config == nil {
-		config = &conf.AdminBackend_MQ{}
-	}
-	cfg := mq.Config{Enabled: config.Enabled, Broker: config.Broker, ClientID: config.ClientId, Username: config.Username, Password: config.Password, CleanSession: config.CleanSession}
-	if config.KeepAlive > 0 {
-		cfg.KeepAlive = time.Duration(config.KeepAlive) * time.Second
-	}
-	if config.ConnectTimeout > 0 {
-		cfg.ConnectTimeout = time.Duration(config.ConnectTimeout) * time.Second
-	}
-	client, err := mq.NewMQTT(cfg)
-	if err != nil {
-		r.logger.Warn("emqx unavailable", "mod", "mq", "error", err)
+	if !config.Enabled {
+		r.replaceClientLocked(provider, nil)
 		return
 	}
-	if config.Enabled {
-		if err = r.restoreSubscriptions(context.Background(), client); err != nil {
-			_ = client.Close()
-			r.logger.Warn("restore emqx subscriptions failed", "mod", "mq", "error", err)
-			return
-		}
+	client, err := newProviderClient(provider, config.Values)
+	if err != nil {
+		r.logger.Warn("message integration unavailable", "mod", "mq", "provider", provider, "error", err)
+		return
 	}
+	if err = r.restoreSubscriptionsLocked(provider, client); err != nil {
+		_ = client.Close()
+		r.logger.Warn("restore message subscriptions failed", "mod", "mq", "provider", provider, "error", err)
+		return
+	}
+	r.replaceClientLocked(provider, client)
+}
+
+func newProviderClient(provider string, raw json.RawMessage) (platformmq.Client, error) {
+	values := map[string]any{}
+	if err := json.Unmarshal(raw, &values); err != nil {
+		return nil, fmt.Errorf("decode %s configuration: %w", provider, err)
+	}
+	switch provider {
+	case ProviderEMQX:
+		return platformmq.NewMQTT(platformmq.Config{
+			Enabled:        true,
+			Broker:         configText(values, "broker"),
+			ClientID:       configText(values, "client_id"),
+			Username:       configText(values, "username"),
+			Password:       configText(values, "password"),
+			KeepAlive:      configSeconds(values, "keep_alive"),
+			CleanSession:   configBool(values, "clean_session"),
+			ConnectTimeout: configSeconds(values, "connect_timeout"),
+		})
+	case ProviderRabbitMQ:
+		return platformmq.NewRabbitMQ(platformmq.RabbitMQConfig{
+			Enabled:        true,
+			Host:           configText(values, "host"),
+			Port:           configInt(values, "port"),
+			Username:       configText(values, "username"),
+			Password:       configText(values, "password"),
+			VHost:          configText(values, "vhost"),
+			Exchange:       configText(values, "exchange"),
+			ExchangeType:   configText(values, "exchange_type"),
+			Queue:          configText(values, "queue"),
+			RoutingKey:     configText(values, "routing_key"),
+			Durable:        configBool(values, "durable"),
+			AutoDelete:     configBool(values, "auto_delete"),
+			PrefetchCount:  configInt(values, "prefetch_count"),
+			Heartbeat:      configSeconds(values, "heartbeat"),
+			ConnectTimeout: configSeconds(values, "connect_timeout"),
+			TLS:            configBool(values, "tls"),
+		})
+	default:
+		return nil, fmt.Errorf("unsupported message provider %q", provider)
+	}
+}
+
+func configText(values map[string]any, key string) string {
+	value, ok := values[key]
+	if !ok || value == nil {
+		return ""
+	}
+	return strings.TrimSpace(fmt.Sprint(value))
+}
+
+func configInt(values map[string]any, key string) int {
+	switch value := values[key].(type) {
+	case float64:
+		return int(value)
+	case int:
+		return value
+	case json.Number:
+		parsed, _ := strconv.Atoi(string(value))
+		return parsed
+	default:
+		parsed, _ := strconv.Atoi(configText(values, key))
+		return parsed
+	}
+}
+
+func configSeconds(values map[string]any, key string) time.Duration {
+	seconds := configInt(values, key)
+	if seconds <= 0 {
+		return 0
+	}
+	return time.Duration(seconds) * time.Second
+}
+
+func configBool(values map[string]any, key string) bool {
+	value, _ := values[key].(bool)
+	return value
+}
+
+func (r *Reloadable) replaceClientLocked(provider string, next platformmq.Client) {
 	r.mu.Lock()
-	old := r.current
-	r.current = client
+	old := r.clients[provider]
+	if next == nil {
+		delete(r.clients, provider)
+	} else {
+		r.clients[provider] = next
+	}
 	r.mu.Unlock()
 	if old != nil {
 		_ = old.Close()
 	}
 }

-func (r *Reloadable) restoreSubscriptions(ctx context.Context, client mq.Client) error {
-	for topic, item := range r.subscriptions {
-		if err := client.Subscribe(ctx, topic, item.qos, item.handler); err != nil {
+func (r *Reloadable) restoreSubscriptionsLocked(provider string, client platformmq.Client) error {
+	for topic, item := range r.subscriptions[provider] {
+		if err := client.Subscribe(context.Background(), topic, item.qos, item.handler); err != nil {
 			return err
 		}
 	}
 	return nil
 }

-func (r *Reloadable) client() mq.Client { r.mu.RLock(); defer r.mu.RUnlock(); return r.current }
-func (r *Reloadable) Publish(ctx context.Context, topic string, payload []byte, qos byte, retain bool) error {
-	c := r.client()
-	if c == nil {
-		return mq.ErrUnavailable
-	}
-	return c.Publish(ctx, topic, payload, qos, retain)
+func (r *Reloadable) client(provider string) platformmq.Client {
+	provider = strings.ToLower(strings.TrimSpace(provider))
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	return r.clients[provider]
 }
-func (r *Reloadable) Subscribe(ctx context.Context, topic string, qos byte, handler mq.Handler) error {
+
+func (r *Reloadable) Client(provider string) platformmq.Client {
+	provider = strings.ToLower(strings.TrimSpace(provider))
+	if provider != ProviderEMQX && provider != ProviderRabbitMQ {
+		return nil
+	}
+	return &namedClient{owner: r, provider: provider}
+}
+
+func (c *namedClient) Publish(ctx context.Context, topic string, payload []byte, qos byte, retain bool) error {
+	return c.owner.PublishTo(ctx, c.provider, topic, payload, qos, retain)
+}
+func (c *namedClient) Subscribe(ctx context.Context, topic string, qos byte, handler platformmq.Handler) error {
+	return c.owner.SubscribeTo(ctx, c.provider, topic, qos, handler)
+}
+func (c *namedClient) Unsubscribe(ctx context.Context, topics ...string) error {
+	return c.owner.UnsubscribeFrom(ctx, c.provider, topics...)
+}
+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()
-	c := r.client()
-	if c == nil {
-		return mq.ErrUnavailable
+	client := r.client(provider)
+	if client == nil {
+		return platformmq.ErrUnavailable
 	}
-	if err := c.Subscribe(ctx, topic, qos, handler); err != nil {
+	if err := client.Subscribe(ctx, topic, qos, handler); err != nil {
 		return err
 	}
-	r.subscriptions[topic] = subscription{qos: qos, handler: handler}
+	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()
-	c := r.client()
-	if c == nil {
-		return mq.ErrUnavailable
+	client := r.client(provider)
+	if client == nil {
+		return platformmq.ErrUnavailable
 	}
-	if err := c.Unsubscribe(ctx, topics...); err != nil {
+	if err := client.Unsubscribe(ctx, topics...); err != nil {
 		return err
 	}
 	for _, topic := range topics {
-		delete(r.subscriptions, topic)
+		delete(r.subscriptions[provider], topic)
 	}
 	return nil
 }
-func (r *Reloadable) Connected() bool { c := r.client(); return c != nil && c.Connected() }
+
+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]
+	return client != nil && client.Connected()
+}
+
 func (r *Reloadable) Close() error {
 	r.opMu.Lock()
 	defer r.opMu.Unlock()
@@ -144,11 +298,16 @@ func (r *Reloadable) Close() error {
 	}
 	r.closed = true
 	r.mu.Lock()
-	old := r.current
-	r.current = nil
+	clients := make([]platformmq.Client, 0, len(r.clients))
+	for provider, client := range r.clients {
+		clients = append(clients, client)
+		delete(r.clients, provider)
+	}
 	r.mu.Unlock()
-	if old != nil {
-		return old.Close()
+	for _, client := range clients {
+		if client != nil {
+			_ = client.Close()
+		}
 	}
 	return nil
 }
@@ -26,28 +26,33 @@ func (*fakeClient) Close() error    { return nil }

 func TestReloadableTracksSubscriptions(t *testing.T) {
 	client := &fakeClient{}
-	r := &Reloadable{current: client, subscriptions: make(map[string]subscription)}
+	r := &Reloadable{
+		clients:       map[string]platformmq.Client{ProviderEMQX: client},
+		subscriptions: make(map[string]map[string]subscription),
+	}
 	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["orders/+/paid"]; !ok {
+	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["orders/+/paid"]; ok {
+	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]subscription{
-		"orders/+/paid": {qos: platformmq.AtLeastOnce, handler: func(context.Context, platformmq.Message) {}},
+	r := &Reloadable{subscriptions: map[string]map[string]subscription{
+		ProviderEMQX: {
+			"orders/+/paid": {qos: platformmq.AtLeastOnce, handler: func(context.Context, platformmq.Message) {}},
+		},
 	}}
-	if err := r.restoreSubscriptions(context.Background(), client); err != nil {
+	if err := r.restoreSubscriptionsLocked(ProviderEMQX, client); err != nil {
 		t.Fatal(err)
 	}
 	if len(client.subscribed) != 1 || client.subscribed[0] != "orders/+/paid" {