diff --git a/cmd/iot-service/main.go b/cmd/iot-service/main.go index f24ac33..303ae38 100644 --- a/cmd/iot-service/main.go +++ b/cmd/iot-service/main.go @@ -118,9 +118,11 @@ func run() int { go func() { defer wg.Done() - ready.SetBridge(true) + // ready.SetBridge управляется самим bridge: true только после успешной + // подписки (находка #3). Иначе «соединён, но не подписан» давал /ready=200 + // и молча терял телеметрию. defer ready.SetBridge(false) - if err := bridge.Run(ctx, cfg, sqsClient, log); err != nil && !errors.Is(err, context.Canceled) { + if err := bridge.Run(ctx, cfg, sqsClient, log, ready.SetBridge); err != nil && !errors.Is(err, context.Canceled) { log.Error("bridge stopped", "err", err) fatal.Store(true) cancel() // штатный shutdown, затем выход ≠ 0 diff --git a/internal/service/bridge/bridge.go b/internal/service/bridge/bridge.go index 5d1995e..afafa86 100644 --- a/internal/service/bridge/bridge.go +++ b/internal/service/bridge/bridge.go @@ -23,8 +23,18 @@ const telemetryTopicFilter = "+/telemetry/+" // Подписка обязательна (иначе bridge «глохнет»: соединение есть, а топик не слушается). const subscribeRetryInterval = 2 * time.Second -// Run — бесконечный цикл bridge: подключение → подписка → ожидание сигнала завершения. -func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, log *slog.Logger) error { +// setSubscribed — nil-safe вызов колбэка состояния подписки. +func setSubscribed(fn func(bool), v bool) { + if fn != nil { + fn(v) + } +} + +// Run — бесконечный цикл bridge. onSubscribed — колбэк состояния подписки +// (true после успешного SUBSCRIBE, false при потере соединения/сбое подписки): +// /ready должен отражать реальную подписку, а не факт запуска горутины +// (находка #3: «соединён, но не подписан» молча терял телеметрию при /ready=200). +func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, log *slog.Logger, onSubscribed func(bool)) error { log.Info("bridge: starting", "broker", cfg.MQTTBrokerURL, "client_id", cfg.MQTTClientID) // URL очереди получаем один раз при старте (как старый consumer). @@ -46,7 +56,7 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, log *sl handler := newMessageHandler(dispatcher, log) - client, err := connectMQTT(ctx, cfg, log, handler) + client, err := connectMQTT(ctx, cfg, log, handler, onSubscribed) if err != nil { dispatcher.closeAndWait() return err @@ -70,7 +80,7 @@ func (e errBridge) Error() string { return string(e) } // connectMQTT устанавливает подключение к EMQX с автореконнектом. // handler передаётся в OnConnectHandler: подписка повторяется при каждом подключении. -func connectMQTT(ctx context.Context, cfg *config.Config, log *slog.Logger, handler mqtt.MessageHandler) (mqtt.Client, error) { +func connectMQTT(ctx context.Context, cfg *config.Config, log *slog.Logger, handler mqtt.MessageHandler, onSubscribed func(bool)) (mqtt.Client, error) { opts := mqtt.NewClientOptions() opts.AddBroker(cfg.MQTTBrokerURL) opts.SetClientID(cfg.MQTTClientID) @@ -84,6 +94,7 @@ func connectMQTT(ctx context.Context, cfg *config.Config, log *slog.Logger, hand opts.SetConnectionLostHandler(func(_ mqtt.Client, err error) { log.Warn("bridge: MQTT connection lost, reconnecting...", "err", err) + setSubscribed(onSubscribed, false) }) opts.SetReconnectingHandler(func(_ mqtt.Client, _ *mqtt.ClientOptions) { log.Info("bridge: MQTT reconnecting...") @@ -98,6 +109,7 @@ func connectMQTT(ctx context.Context, cfg *config.Config, log *slog.Logger, hand token := c.Subscribe(telemetryTopicFilter, 1, handler) if token.WaitTimeout(10*time.Second) && token.Error() == nil { log.Info("bridge: subscribed", "filter", telemetryTopicFilter, "attempt", attempt) + setSubscribed(onSubscribed, true) return } log.Error("bridge: subscribe failed", "filter", telemetryTopicFilter, "attempt", attempt, "err", token.Error()) @@ -107,6 +119,9 @@ func connectMQTT(ctx context.Context, cfg *config.Config, log *slog.Logger, hand case <-time.After(subscribeRetryInterval): } } + // Подписка не удалась — bridge не готов (/ready=503) до следующего + // реконнекта. + setSubscribed(onSubscribed, false) log.Error("bridge: subscribe failed after retries, will retry on next reconnect", "filter", telemetryTopicFilter) })