cacto
abc83faf05
Слушает telemetry.new_reading (ручной ack/nack, реквизишн только при транспортных ошибках gRPC), кэширует активные automation_rules в памяти с периодическим обновлением из PostgreSQL (JOIN с devices — резолвит внутренние ID в external_id, которым оперируют MQTT/Redis/gRPC). Условия правил (>,<,>=,<=,=,!=) оцениваются обобщённо, без привязки к конкретным типам устройств. При срабатывании вызывает device-control-service по gRPC и публикует automation.rule_triggered в RabbitMQ. Проверено сквозным тестом через docker compose на полном пайплайне: mosquitto_pub → ingest-service (ClickHouse + telemetry.new_reading) → rule-engine-service (совпадение правила) → device-control-service (gRPC → Redis desired_state + MQTT-команда) → automation.rule_triggered. Показание ниже порога проверено отдельно — правило корректно не срабатывает.
63 lines
1.8 KiB
Go
63 lines
1.8 KiB
Go
// Package rabbitmq consumes ingest-service's "new reading" events and
|
|
// publishes rule-trigger events for notification-service. One connection is
|
|
// shared, with separate channels for consuming and publishing, since that's
|
|
// the amqp best practice (a channel is not meant to be used concurrently for
|
|
// unrelated flows).
|
|
package rabbitmq
|
|
|
|
import (
|
|
"fmt"
|
|
|
|
amqp "github.com/rabbitmq/amqp091-go"
|
|
)
|
|
|
|
const (
|
|
NewReadingQueue = "telemetry.new_reading" // declared by ingest-service; declared here too, idempotently
|
|
RuleTriggeredQueue = "automation.rule_triggered"
|
|
)
|
|
|
|
type Client struct {
|
|
conn *amqp.Connection
|
|
consumeCh *amqp.Channel
|
|
publishCh *amqp.Channel
|
|
}
|
|
|
|
func Connect(url string) (*Client, error) {
|
|
conn, err := amqp.Dial(url)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("dial rabbitmq: %w", err)
|
|
}
|
|
|
|
consumeCh, err := conn.Channel()
|
|
if err != nil {
|
|
conn.Close()
|
|
return nil, fmt.Errorf("open consume channel: %w", err)
|
|
}
|
|
// Process one reading at a time: simple, demonstrable backpressure —
|
|
// the next delivery isn't sent until the current one is acked/nacked.
|
|
if err := consumeCh.Qos(1, 0, false); err != nil {
|
|
conn.Close()
|
|
return nil, fmt.Errorf("set qos: %w", err)
|
|
}
|
|
if _, err := consumeCh.QueueDeclare(NewReadingQueue, true, false, false, false, nil); err != nil {
|
|
conn.Close()
|
|
return nil, fmt.Errorf("declare queue %s: %w", NewReadingQueue, err)
|
|
}
|
|
|
|
publishCh, err := conn.Channel()
|
|
if err != nil {
|
|
conn.Close()
|
|
return nil, fmt.Errorf("open publish channel: %w", err)
|
|
}
|
|
if _, err := publishCh.QueueDeclare(RuleTriggeredQueue, true, false, false, false, nil); err != nil {
|
|
conn.Close()
|
|
return nil, fmt.Errorf("declare queue %s: %w", RuleTriggeredQueue, err)
|
|
}
|
|
|
|
return &Client{conn: conn, consumeCh: consumeCh, publishCh: publishCh}, nil
|
|
}
|
|
|
|
func (c *Client) Close() error {
|
|
return c.conn.Close()
|
|
}
|