409 lines
12 KiB
Go
409 lines
12 KiB
Go
package mq
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log/slog"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"kra/internal/integration/runtimeconfig"
|
|
platformmq "kra/pkg/mq"
|
|
)
|
|
|
|
const (
|
|
ProviderEMQX = platformmq.ProviderEMQX
|
|
ProviderRabbitMQ = platformmq.ProviderRabbitMQ
|
|
retryTick = time.Second
|
|
)
|
|
|
|
// 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
|
|
clients map[string]platformmq.Client
|
|
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
|
|
}
|
|
|
|
type subscription struct {
|
|
qos byte
|
|
handler platformmq.Handler
|
|
}
|
|
|
|
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{
|
|
clients: make(map[string]platformmq.Client),
|
|
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 {
|
|
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) }),
|
|
)
|
|
}
|
|
go r.retryLoop()
|
|
cleanup := func() {
|
|
for _, stop := range r.stop {
|
|
stop()
|
|
}
|
|
_ = r.Close()
|
|
}
|
|
return r, cleanup, nil
|
|
}
|
|
|
|
func storeConfig(store *runtimeconfig.Store, provider string) runtimeconfig.Config {
|
|
config, _ := store.Get("mq", provider)
|
|
return config
|
|
}
|
|
|
|
// TestConfig creates a short-lived provider client and closes it immediately.
|
|
// For RabbitMQ this also checks the configured exchange and queue topology.
|
|
func TestConfig(ctx context.Context, provider string, raw json.RawMessage) error {
|
|
if ctx != nil {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
default:
|
|
}
|
|
}
|
|
provider = strings.ToLower(strings.TrimSpace(provider))
|
|
if provider == ProviderEMQX {
|
|
values := map[string]any{}
|
|
if err := json.Unmarshal(raw, &values); err != nil {
|
|
return fmt.Errorf("decode %s configuration: %w", provider, err)
|
|
}
|
|
baseID := configText(values, "client_id")
|
|
values["client_id"] = fmt.Sprintf("%s-test-%d", baseID, time.Now().UnixNano())
|
|
encoded, err := json.Marshal(values)
|
|
if err != nil {
|
|
return fmt.Errorf("encode %s test configuration: %w", provider, err)
|
|
}
|
|
raw = encoded
|
|
}
|
|
client, err := newProviderClient(provider, raw)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if client == nil || !client.Connected() {
|
|
if client != nil {
|
|
_ = client.Close()
|
|
}
|
|
return platformmq.ErrUnavailable
|
|
}
|
|
closeErr := client.Close()
|
|
if ctx != nil {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
default:
|
|
}
|
|
}
|
|
return closeErr
|
|
}
|
|
|
|
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 {
|
|
return err
|
|
}
|
|
bindings, err := r.restoreSubscriptionsLocked(provider, client)
|
|
if err != nil {
|
|
_ = client.Close()
|
|
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) {
|
|
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"),
|
|
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"),
|
|
ReconnectInterval: configSeconds(values, "reconnect_interval"),
|
|
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.clients[provider]
|
|
if next == nil {
|
|
delete(r.clients, provider)
|
|
delete(r.bindings, provider)
|
|
} else {
|
|
r.clients[provider] = next
|
|
}
|
|
r.mu.Unlock()
|
|
if old != nil {
|
|
_ = old.Close()
|
|
}
|
|
}
|
|
|
|
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 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 {
|
|
provider = strings.ToLower(strings.TrimSpace(provider))
|
|
r.mu.RLock()
|
|
defer r.mu.RUnlock()
|
|
return r.clients[provider]
|
|
}
|
|
|
|
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()
|
|
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]
|
|
return client != nil && client.Connected()
|
|
}
|
|
|
|
func (r *Reloadable) Close() error {
|
|
r.opMu.Lock()
|
|
defer r.opMu.Unlock()
|
|
if r.closed {
|
|
return nil
|
|
}
|
|
r.closed = true
|
|
r.mu.Lock()
|
|
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()
|
|
for _, client := range clients {
|
|
if client != nil {
|
|
_ = client.Close()
|
|
}
|
|
}
|
|
return nil
|
|
}
|