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. Показание ниже порога проверено отдельно — правило корректно не срабатывает.
66 lines
1.8 KiB
Go
66 lines
1.8 KiB
Go
package rabbitmq
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"time"
|
|
)
|
|
|
|
// ReadingEvent mirrors ingest-service's newReadingEvent wire format.
|
|
type ReadingEvent struct {
|
|
DeviceID string `json:"device_id"`
|
|
ZoneID string `json:"zone_id"`
|
|
SensorType string `json:"sensor_type"`
|
|
Value float64 `json:"value"`
|
|
RecordedAt time.Time `json:"recorded_at"`
|
|
}
|
|
|
|
// HandleResult tells ConsumeReadings how to settle a delivery.
|
|
type HandleResult int
|
|
|
|
const (
|
|
Ack HandleResult = iota // processed successfully
|
|
NackRequeue // transient failure (e.g. gRPC unreachable) — try again later
|
|
NackDiscard // permanent failure (e.g. bad data) — retrying won't help
|
|
)
|
|
|
|
// Handler processes one reading event and decides how it should be settled.
|
|
type Handler func(ctx context.Context, event ReadingEvent) HandleResult
|
|
|
|
// ConsumeReadings blocks, delivering messages to handler, until ctx is
|
|
// cancelled or the delivery channel closes (e.g. connection lost).
|
|
func (c *Client) ConsumeReadings(ctx context.Context, handler Handler) error {
|
|
deliveries, err := c.consumeCh.ConsumeWithContext(ctx, NewReadingQueue, "", false, false, false, false, nil)
|
|
if err != nil {
|
|
return fmt.Errorf("consume %s: %w", NewReadingQueue, err)
|
|
}
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil
|
|
case d, ok := <-deliveries:
|
|
if !ok {
|
|
return fmt.Errorf("delivery channel for %s closed", NewReadingQueue)
|
|
}
|
|
|
|
var event ReadingEvent
|
|
if err := json.Unmarshal(d.Body, &event); err != nil {
|
|
// Poison message: no amount of retrying fixes malformed JSON.
|
|
_ = d.Nack(false, false)
|
|
continue
|
|
}
|
|
|
|
switch handler(ctx, event) {
|
|
case Ack:
|
|
_ = d.Ack(false)
|
|
case NackRequeue:
|
|
_ = d.Nack(false, true)
|
|
case NackDiscard:
|
|
_ = d.Nack(false, false)
|
|
}
|
|
}
|
|
}
|
|
}
|