diff --git a/internal/integration/mq/emqx.go b/internal/integration/mq/emqx.go index ae0dbec..a8c9c1e 100644 --- a/internal/integration/mq/emqx.go +++ b/internal/integration/mq/emqx.go @@ -15,8 +15,9 @@ import ( ) const ( - ProviderEMQX = "emqx" - ProviderRabbitMQ = "rabbitmq" + ProviderEMQX = platformmq.ProviderEMQX + ProviderRabbitMQ = platformmq.ProviderRabbitMQ + retryTick = time.Second ) // Reloadable owns the process-wide message clients. Configuration comes only @@ -25,8 +26,15 @@ type Reloadable struct { mu sync.RWMutex opMu sync.Mutex clients map[string]platformmq.Client - subscriptions map[string]map[string]subscription + configs map[string]runtimeconfig.Config + subscriptions map[string]map[string]map[string]subscription + bindings map[string]map[string]byte + pending map[string]bool + nextRetry map[string]time.Time stop []func() + retryStop chan struct{} + retryDone chan struct{} + closeOnce sync.Once logger *slog.Logger closed bool } @@ -47,7 +55,13 @@ func New(store *runtimeconfig.Store, logger *slog.Logger) (*Reloadable, func(), } r := &Reloadable{ clients: make(map[string]platformmq.Client), - subscriptions: make(map[string]map[string]subscription), + configs: make(map[string]runtimeconfig.Config), + 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), + retryStop: make(chan struct{}), + retryDone: make(chan struct{}), logger: logger, } if store != nil { @@ -58,6 +72,7 @@ func New(store *runtimeconfig.Store, logger *slog.Logger) (*Reloadable, func(), store.Subscribe("mq", ProviderRabbitMQ, func(config runtimeconfig.Config) { r.apply(ProviderRabbitMQ, config) }), ) } + go r.retryLoop() cleanup := func() { for _, stop := range r.stop { stop() @@ -118,26 +133,45 @@ func TestConfig(ctx context.Context, provider string, raw json.RawMessage) error } func (r *Reloadable) apply(provider string, config runtimeconfig.Config) { + provider = strings.ToLower(strings.TrimSpace(provider)) r.opMu.Lock() defer r.opMu.Unlock() if r.closed { return } + config.Provider = provider + config.Values = append(json.RawMessage(nil), config.Values...) + r.configs[provider] = config 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) + } +} + +func (r *Reloadable) activateLocked(provider string, config runtimeconfig.Config) error { client, err := newProviderClient(provider, config.Values) if err != nil { - r.logger.Warn("message integration unavailable", "mod", "mq", "provider", provider, "error", err) - return + return err } - if err = r.restoreSubscriptionsLocked(provider, client); err != nil { + bindings, err := r.restoreSubscriptionsLocked(provider, client) + if err != nil { _ = client.Close() - r.logger.Warn("restore message subscriptions failed", "mod", "mq", "provider", provider, "error", err) - return + return fmt.Errorf("restore message subscriptions: %w", err) } r.replaceClientLocked(provider, client) + r.mu.Lock() + r.bindings[provider] = bindings + r.mu.Unlock() + delete(r.pending, provider) + delete(r.nextRetry, provider) + return nil } func newProviderClient(provider string, raw json.RawMessage) (platformmq.Client, error) { @@ -148,33 +182,35 @@ func newProviderClient(provider string, raw json.RawMessage) (platformmq.Client, 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"), + 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"), + ReconnectInterval: configSeconds(values, "reconnect_interval"), }) 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"), + 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"), + ReconnectInterval: configSeconds(values, "reconnect_interval"), + TLS: configBool(values, "tls"), }) default: return nil, fmt.Errorf("unsupported message provider %q", provider) @@ -222,6 +258,7 @@ func (r *Reloadable) replaceClientLocked(provider string, next platformmq.Client old := r.clients[provider] if next == nil { delete(r.clients, provider) + delete(r.bindings, provider) } else { r.clients[provider] = next } @@ -231,13 +268,26 @@ func (r *Reloadable) replaceClientLocked(provider string, next platformmq.Client } } -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 +func (r *Reloadable) restoreSubscriptionsLocked(provider string, client platformmq.Client) (map[string]byte, error) { + desired := r.desiredSubscriptions(provider) + bindings := make(map[string]byte, len(desired)) + for topic, qos := range desired { + if err := client.Subscribe(context.Background(), topic, qos, r.dispatcher(provider, topic)); err != nil { + return nil, err } + bindings[topic] = qos } - return nil + 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 + } + return interval } func (r *Reloadable) client(provider string) platformmq.Client { diff --git a/pkg/mq/mq.go b/pkg/mq/mq.go index 0df0e98..9e2e6c2 100644 --- a/pkg/mq/mq.go +++ b/pkg/mq/mq.go @@ -10,6 +10,11 @@ import ( var ErrUnavailable = errors.New("message broker unavailable") +const ( + ProviderEMQX = "emqx" + ProviderRabbitMQ = "rabbitmq" +) + const ( AtMostOnce byte = 0 AtLeastOnce byte = 1 @@ -27,6 +32,37 @@ func (m Message) DecodeJSON(target any) error { return json.Unmarshal(m.Payload, type Handler func(context.Context, Message) +// TopicSubscription is one logical module subscription. The handler is kept +// by the runtime and is replayed after a client reconnects or is rebuilt. +type TopicSubscription struct { + Topic string + QoS byte + Handler Handler +} + +// SubscriptionSet is the complete subscription declaration for one module and +// provider. Registering the same owner/provider replaces its previous set. +// Owner must be a stable module name, not a request or goroutine identifier. +type SubscriptionSet struct { + Owner string + Provider string + Topics []TopicSubscription +} + +// SubscriptionRegistrar is the module-facing seam for durable-in-process +// subscription intent. Register does not require a live broker; the runtime +// will bind the declared topics when the provider becomes available. +type SubscriptionRegistrar interface { + Register(SubscriptionSet) error + Unregister(owner string) error +} + +// SubscriptionContributor lets a module expose its broker subscriptions +// without depending on the concrete integration implementation. +type SubscriptionContributor interface { + RegisterSubscriptions(SubscriptionRegistrar) error +} + type Client interface { Publish(context.Context, string, []byte, byte, bool) error Subscribe(context.Context, string, byte, Handler) error @@ -38,6 +74,7 @@ type Client interface { // Registry exposes named broker clients while preserving Client as the // default EMQX/MQTT boundary for existing modules. type Registry interface { + SubscriptionRegistrar Client(provider string) Client PublishTo(context.Context, string, string, []byte, byte, bool) error SubscribeTo(context.Context, string, string, byte, Handler) error diff --git a/pkg/mq/subscription.go b/pkg/mq/subscription.go new file mode 100644 index 0000000..6dec36f --- /dev/null +++ b/pkg/mq/subscription.go @@ -0,0 +1,55 @@ +package mq + +import ( + "fmt" + "strings" +) + +// NormalizeSubscriptionSet validates and canonicalizes a module declaration. +func NormalizeSubscriptionSet(set SubscriptionSet) (SubscriptionSet, error) { + set.Owner = strings.TrimSpace(set.Owner) + set.Provider = strings.ToLower(strings.TrimSpace(set.Provider)) + if set.Owner == "" { + return SubscriptionSet{}, fmt.Errorf("mq subscription owner is empty") + } + if set.Provider != ProviderEMQX && set.Provider != ProviderRabbitMQ { + return SubscriptionSet{}, fmt.Errorf("unsupported mq provider %q", set.Provider) + } + if len(set.Topics) == 0 { + return SubscriptionSet{}, fmt.Errorf("mq subscription topics are empty") + } + seen := make(map[string]struct{}, len(set.Topics)) + for index := range set.Topics { + set.Topics[index].Topic = strings.TrimSpace(set.Topics[index].Topic) + if set.Topics[index].Topic == "" { + return SubscriptionSet{}, fmt.Errorf("mq subscription topic at index %d is empty", index) + } + if set.Topics[index].QoS > ExactlyOnce { + return SubscriptionSet{}, fmt.Errorf("invalid mq qos %d for topic %q", set.Topics[index].QoS, set.Topics[index].Topic) + } + if set.Topics[index].Handler == nil { + return SubscriptionSet{}, fmt.Errorf("mq subscription handler is nil for topic %q", set.Topics[index].Topic) + } + if _, exists := seen[set.Topics[index].Topic]; exists { + return SubscriptionSet{}, fmt.Errorf("duplicate mq subscription topic %q", set.Topics[index].Topic) + } + seen[set.Topics[index].Topic] = struct{}{} + } + return set, nil +} + +// ApplySubscriptions activates dependency-bearing module contributors. +func ApplySubscriptions(registrar SubscriptionRegistrar, contributors ...SubscriptionContributor) error { + if registrar == nil { + return fmt.Errorf("mq subscription registrar is nil") + } + for _, contributor := range contributors { + if contributor == nil { + continue + } + if err := contributor.RegisterSubscriptions(registrar); err != nil { + return err + } + } + return nil +}