cactoandClaude Sonnet 5
938700cf86
Subscribes to devices/+/telemetry, validates payloads, buffers writes into ClickHouse (flush on size or interval), and emits a per-reading event on RabbitMQ for rule-engine-service to consume later. Adds the initial Postgres and ClickHouse schema migrations, a Dockerfile, and wires the service into docker-compose. Verified end-to-end via docker compose + mosquitto_pub: row lands in ClickHouse and the event reaches the telemetry.new_reading queue. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
85 lines
2.0 KiB
Go
85 lines
2.0 KiB
Go
package batch
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
"log/slog"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/telemetry"
|
|
)
|
|
|
|
func discardLogger() *slog.Logger {
|
|
return slog.New(slog.NewTextHandler(io.Discard, nil))
|
|
}
|
|
|
|
func reading(deviceID string) telemetry.Reading {
|
|
return telemetry.Reading{DeviceID: deviceID, ZoneID: "z1", SensorType: "temperature", Value: 1}
|
|
}
|
|
|
|
func awaitBatch(t *testing.T, flushed chan []telemetry.Reading, timeout time.Duration) []telemetry.Reading {
|
|
t.Helper()
|
|
select {
|
|
case batch := <-flushed:
|
|
return batch
|
|
case <-time.After(timeout):
|
|
t.Fatal("timed out waiting for flush")
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func TestBatcher_FlushesOnSize(t *testing.T) {
|
|
flushed := make(chan []telemetry.Reading, 1)
|
|
b := New(3, time.Hour, time.Second, func(_ context.Context, r []telemetry.Reading) error {
|
|
flushed <- r
|
|
return nil
|
|
}, discardLogger())
|
|
b.Start()
|
|
defer b.Stop()
|
|
|
|
b.Add(reading("a"))
|
|
b.Add(reading("b"))
|
|
b.Add(reading("c")) // hits maxSize, should trigger an immediate flush
|
|
|
|
batch := awaitBatch(t, flushed, time.Second)
|
|
if len(batch) != 3 {
|
|
t.Fatalf("got %d readings, want 3", len(batch))
|
|
}
|
|
}
|
|
|
|
func TestBatcher_FlushesOnInterval(t *testing.T) {
|
|
flushed := make(chan []telemetry.Reading, 1)
|
|
b := New(100, 20*time.Millisecond, time.Second, func(_ context.Context, r []telemetry.Reading) error {
|
|
flushed <- r
|
|
return nil
|
|
}, discardLogger())
|
|
b.Start()
|
|
defer b.Stop()
|
|
|
|
b.Add(reading("a"))
|
|
|
|
batch := awaitBatch(t, flushed, time.Second)
|
|
if len(batch) != 1 {
|
|
t.Fatalf("got %d readings, want 1", len(batch))
|
|
}
|
|
}
|
|
|
|
func TestBatcher_StopFlushesRemainder(t *testing.T) {
|
|
flushed := make(chan []telemetry.Reading, 1)
|
|
b := New(100, time.Hour, time.Second, func(_ context.Context, r []telemetry.Reading) error {
|
|
flushed <- r
|
|
return nil
|
|
}, discardLogger())
|
|
b.Start()
|
|
|
|
b.Add(reading("a"))
|
|
b.Add(reading("b"))
|
|
b.Stop()
|
|
|
|
batch := awaitBatch(t, flushed, time.Second)
|
|
if len(batch) != 2 {
|
|
t.Fatalf("got %d readings, want 2", len(batch))
|
|
}
|
|
}
|