cacto
3eb84aae6c
Добавлен gRPC-контракт proto/device_control (TurnOn/TurnOff/SetLevel), общий Go-модуль proto/ для переиспользования сгенерированного кода. device-control-service принимает команды по gRPC, обновляет desired_state в Redis и публикует их в MQTT; слушает devices/+/telemetry и devices/+/ack для отметки живости устройства и обновления reported_state; горутина health-check переводит устройство в offline по таймауту и публикует событие в RabbitMQ (device.status_changed) для будущего notification-service. Проверено сквозным тестом через docker compose: grpcurl TurnOn/SetLevel меняет desired_state и уходит в MQTT, mosquitto_pub с ack обновляет reported_state и статус, health-check автоматически переводит устройство в offline по истечении таймаута.
79 lines
1.9 KiB
Go
79 lines
1.9 KiB
Go
// Package rabbitmq publishes device online/offline transitions for
|
|
// notification-service to consume. Delivery reliability matters more than
|
|
// latency here, so messages are persistent and the queue is durable.
|
|
package rabbitmq
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"time"
|
|
|
|
amqp "github.com/rabbitmq/amqp091-go"
|
|
)
|
|
|
|
const StatusChangedQueue = "device.status_changed"
|
|
|
|
type Publisher struct {
|
|
conn *amqp.Connection
|
|
ch *amqp.Channel
|
|
}
|
|
|
|
func Connect(url string) (*Publisher, error) {
|
|
conn, err := amqp.Dial(url)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("dial rabbitmq: %w", err)
|
|
}
|
|
|
|
ch, err := conn.Channel()
|
|
if err != nil {
|
|
conn.Close()
|
|
return nil, fmt.Errorf("open channel: %w", err)
|
|
}
|
|
|
|
if _, err := ch.QueueDeclare(StatusChangedQueue, true, false, false, false, nil); err != nil {
|
|
ch.Close()
|
|
conn.Close()
|
|
return nil, fmt.Errorf("declare queue %s: %w", StatusChangedQueue, err)
|
|
}
|
|
|
|
return &Publisher{conn: conn, ch: ch}, nil
|
|
}
|
|
|
|
func (p *Publisher) Close() error {
|
|
if err := p.ch.Close(); err != nil {
|
|
p.conn.Close()
|
|
return err
|
|
}
|
|
return p.conn.Close()
|
|
}
|
|
|
|
type statusChangedEvent struct {
|
|
DeviceID string `json:"device_id"`
|
|
Status string `json:"status"`
|
|
At time.Time `json:"at"`
|
|
}
|
|
|
|
// PublishStatusChanged emits one event when a device transitions online or offline.
|
|
func (p *Publisher) PublishStatusChanged(ctx context.Context, deviceID, status string) error {
|
|
body, err := json.Marshal(statusChangedEvent{
|
|
DeviceID: deviceID,
|
|
Status: status,
|
|
At: time.Now(),
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("marshal event: %w", err)
|
|
}
|
|
|
|
err = p.ch.PublishWithContext(ctx, "", StatusChangedQueue, false, false, amqp.Publishing{
|
|
ContentType: "application/json",
|
|
DeliveryMode: amqp.Persistent,
|
|
Timestamp: time.Now(),
|
|
Body: body,
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("publish status change for device %s: %w", deviceID, err)
|
|
}
|
|
return nil
|
|
}
|