Browse Source

feat(web): refresh laundry status on MQTT messages via server-sent events

main
Domagoj Zecevic 3 weeks ago
parent
commit
ef46fc849a
  1. 27
      .ai/HANDOFF.md
  2. 108
      .ai/PLAN.md
  3. 19
      .ai/REVIEW.md
  4. 1
      .ai/TASKS.md
  5. 41
      README.md
  6. 18
      ROADMAP.md
  7. 41
      internal/sensor/hub.go
  8. 75
      internal/sensor/hub_test.go
  9. 10
      internal/sensor/sampler.go
  10. 35
      internal/sensor/sampler_test.go
  11. 78
      internal/web/handler.go
  12. 202
      internal/web/handler_test.go
  13. 9
      internal/web/static/index.html
  14. 27
      main.go
  15. 44
      main_test.go

27
.ai/HANDOFF.md

@ -60,3 +60,30 @@ Append-only role handoff log. Each role adds one entry when its step is complete
- Status: done. Commit message read from the T-003 next_task entry. - Status: done. Commit message read from the T-003 next_task entry.
- All three tasks completed and reviewed; one separate Conventional Commit per task. - All three tasks completed and reviewed; one separate Conventional Commit per task.
### T-004 — plan — 2026-10-12T12:06:52Z
| Field | Value |
|-------|-------|
| Agent | claude |
| Summary | Planned SSE-based instant page refresh on each valid MQTT message (Hub, Sampler.OnReceive, /api/events, EventSource with polling fallback) including README and comment updates |
| Files Changed | .ai/PLAN.md, .ai/TASKS.md, .ai/HANDOFF.md, ROADMAP.md |
| Next Role | implement |
### T-004 — next_task (dev/all_task) — 2026-10-12T13:18:18Z
| Field | Value |
|-------|-------|
| Agent | codex |
| Summary | Added MQTT-triggered SSE refresh, non-blocking broadcasts, keep-alives, stream shutdown, polling fallback, tests, and endpoint documentation |
| Validation | fmt/vet/test/race PASS; real Mosquitto/Chrome SSE smoke PASS; page fallback/coalescing checks PASS; review PASS after one recorder synchronization retry |
| Commit | feat(web): refresh laundry status on MQTT messages via server-sent events |
| Next Role | dev commit |
- Includes the existing T-004 roadmap and planning artifacts. No new dependencies.
- PR limitation: `aide pr --dry-run` refuses main; repository instructions require staying on the current branch.
### T-004 — commit_task (all_task) — 2026-10-12T13:18:29Z
- Status: done. Commit message read from the T-004 next_task entry.
- All four board tasks are done; T-004 is the only task completed in this invocation and receives its own Conventional Commit.

108
.ai/PLAN.md

@ -2,79 +2,51 @@
Status: **ready_for_implement** Status: **ready_for_implement**
Goal: implement the scope defined in `ROADMAP.md` — a phone-first webpage, served by a small Go backend in Docker, that shows whether the laundry is done based on the last 3 one-minute power samples from `tele/tasmota_washingmashine_C21183/SENSOR`. Goal: implement `ROADMAP.md` Priority 2 — the page refreshes itself whenever a new valid MQTT power message arrives, using Server-Sent Events (SSE). T-001..T-003 (Priority 1) are done and stay unchanged in behavior.
## Scope ## Scope
- Go module `mqtt_washingmachine` (Go 1.26), standard library plus `github.com/eclipse/paho.mqtt.golang` for MQTT. - Standard library only; no new dependencies.
- Layout: - Out of scope: live-reading line, WebSocket, changes to the status rule, JSON contract of `/api/status`, or 1-minute sampling.
- `main.go`: config from env, wiring, graceful shutdown.
- `internal/status/`: pure logic (sample window, status evaluation). ## Design decisions (from ROADMAP.md Priority 2)
- `internal/sensor/`: Tasmota payload parsing, MQTT subscription, per-minute sampler. - Event source: a valid payload accepted by `Sampler.Receive` (about every 10 s). Invalid payloads return an error and never notify.
- `internal/web/`: HTTP handlers and embedded static page (`go:embed`). - Hub: new `internal/sensor/hub.go` with a concurrency-safe `Hub` (zero value usable):
- Docker packaging and documentation. - `Subscribe() (events <-chan struct{}, cancel func())`: each subscriber gets a channel with buffer 1.
- Out of scope: persistence, TLS/auth, history beyond 3 samples, multiple devices. - `Notify()`: non-blocking send to every subscriber (drop if the buffer is full, so a slow client never blocks MQTT handling and bursts coalesce).
- `cancel` removes the subscriber; safe to call more than once.
## Design decisions (from ROADMAP.md) - `Sampler` gets an optional exported field `OnReceive func()`, set before use (documented as such), called after a valid payload is stored and the mutex is released. Zero value stays ready to use. `main.go` sets `sampler.OnReceive = hub.Notify`.
- Threshold: 3.00 W. A sample is "idle" only if `power < 3.0`; `>= 3.0` is "active". - Endpoint `GET /api/events` (`text/event-stream`, `Cache-Control: no-store`):
- Sampling: the MQTT handler stores only the latest `ENERGY.Power` value and its receive time. A ticker fires every `SAMPLE_INTERVAL` (default 60s). At each tick: if a message was received since the previous tick, record its power as a sample; otherwise record a missing sample. Then clear the "received" flag. - On connect: flush headers and send an initial `event: update` so the client syncs immediately (covers reconnects).
- Window: ring of the last 3 samples (oldest first). Each sample is `{Power float64, Valid bool}`. - Per notification: write `event: update\ndata: {}\n\n` and flush.
- Status evaluation (first match wins): - Keep-alive: SSE comment line `: keep-alive\n\n` every 25 s (interval a package-level variable so tests can shorten it).
1. fewer than 3 samples recorded since start -> `waiting`, "Waiting for data...", grey - Clear the write deadline for this stream with `http.NewResponseController(w).SetWriteDeadline(time.Time{})` so the server-wide 10 s `WriteTimeout` does not cut it; keep `WriteTimeout` as is for other routes. If the ResponseWriter does not support flushing, return 500.
2. any of the last 3 samples missing -> `incomplete`, "Incomplete data", grey - Exit on `r.Context().Done()` (client gone or server shutdown) and call `cancel`.
3. any sample `>= 3.0` -> `running`, "Laundry not yet done", red - Graceful shutdown: `server.Shutdown` waits for active handlers; SSE handlers must also stop on shutdown. Register `server.RegisterOnShutdown(cancelStreams)` with a context the handler selects on (e.g. a base context passed to `NewHandler`), so shutdown is not delayed to the 5 s timeout.
4. otherwise (all 3 `< 3.0`) -> `done`, "Laundry done", default text color - `NewHandler` signature becomes `NewHandler(snapshot func() status.Result, events *sensor.Hub)`; `web` imports `sensor` only for the Hub type (or accept an interface `Subscribe() (<-chan struct{}, func())` to avoid the import; prefer the interface).
- JSON contract for `GET /api/status`: - Page (`index.html`):
```json - Open `new EventSource('/api/events')`; on `update` call `refresh()`. `EventSource` reconnects automatically; the `open` event after a reconnect triggers a refresh via the initial `update`.
{"state":"running","message":"Laundry not yet done","readings":[{"power":47.7,"valid":true},{"power":0,"valid":false}]} - Keep `setInterval(refresh, 60000)` as a fallback. Coalesce overlapping refreshes (skip or reuse if one is in flight) so bursts do not stack requests.
``` - If `EventSource` is unavailable, rely on polling only.
`readings` has 0–3 entries, oldest first. The page renders the display text, color class, and `47.70W` / `--` formatting.
- Config env vars: `MQTT_BROKER` (default `tcp://192.168.1.12:1883`), `MQTT_TOPIC` (default `tele/tasmota_washingmashine_C21183/SENSOR`), `MQTT_USERNAME`, `MQTT_PASSWORD` (optional), `HTTP_ADDR` (default `:8080`), `SAMPLE_INTERVAL` (Go duration, default `60s`).
- MQTT client uses auto-reconnect and re-subscribes in the `OnConnect` handler.
## Acceptance Criteria ## Acceptance Criteria
- `GET /` returns a white-background page with a viewport meta tag, a very large status text, and the last 3 readings below it in normal-size black text formatted like `47.70W`; missing readings show `--`. - A valid MQTT message triggers an SSE `update` event to all connected clients within the same moment; the page then re-fetches `/api/status`.
- The page fetches `/api/status` on load and every 60 s; no horizontal scroll at 360 px width. - Invalid payloads produce no event.
- Status states and colors match the table in `ROADMAP.md` exactly, including `>= 3.00 W` = not done. - A new connection receives an immediate `update`; idle connections receive a `: keep-alive` comment at the configured interval.
- The backend collects samples continuously with no page open; a page load immediately reflects the current state. - The stream survives longer than the 10 s `WriteTimeout`; other routes keep it.
- A minute with no MQTT message produces a missing sample and `incomplete` state. - A slow or stuck subscriber never blocks `Receive`/MQTT handling or other subscribers.
- Malformed payloads or payloads without `ENERGY.Power` are ignored (logged), without crashing or changing the latest value. - Disconnected clients are removed from the hub (no goroutine or subscriber leak); shutdown closes streams promptly.
- `docker build` succeeds and the container runs with env-var configuration; README documents setup. - `/api/status` JSON, status rule, and 60 s polling fallback unchanged.
- `go fmt ./...`, `go vet ./...`, `go test ./...` all pass. - `go fmt ./...`, `go vet ./...`, `go test ./...` (also `-race` for sensor and web) pass.
## Tasks ## Tasks
### T-001 — Status logic and Tasmota payload parsing ### T-004 — SSE push so the page refreshes on each MQTT message
Tests first, then implementation.
- `internal/status/status_test.go`, `internal/status/status.go`: window of 3, `Evaluate`, boundary tests (2.99 -> idle, 3.00 -> active, 0 W all three -> done, mixed -> running, gap -> incomplete, <3 samples -> waiting).
- `internal/sensor/payload_test.go`, `internal/sensor/payload.go`: parse the sample payload from the roadmap (`Power` 139.60 etc.), missing `ENERGY`, missing `Power`, invalid JSON.
- `go.mod` (module init, no external deps yet in this task).
- Files to change: `go.mod`, `internal/status/*`, `internal/sensor/payload*.go`.
- Documentation: package doc comments; no README change yet.
### T-002 — Sampler, MQTT subscription, HTTP server and page
Tests first, then implementation. Tests first, then implementation.
- `internal/sensor/sampler_test.go`, `internal/sensor/sampler.go`: latest-value store plus ticker-driven sampling into the window, with an injectable tick channel/clock for deterministic tests (message between ticks -> valid sample, none -> missing). - `internal/sensor/hub_test.go`, `internal/sensor/hub.go`: subscribe/notify delivery to multiple subscribers, coalescing when buffer is full (Notify never blocks), cancel removes subscriber and is idempotent, concurrent use under `-race`.
- `internal/sensor/mqtt.go`: paho client, auto-reconnect, subscribe in `OnConnect`, feeds payloads to the parser then the sampler (thin glue, not unit-tested against a real broker). - `internal/sensor/sampler_test.go`, `internal/sensor/sampler.go`: `OnReceive` called once per valid payload, not called for invalid payload, nil `OnReceive` is safe.
- `internal/web/handler_test.go`, `internal/web/handler.go`, `internal/web/static/index.html`: `/` serves the embedded page, `/api/status` returns the JSON contract (test with `httptest`, including state/message/readings per state). Page styling: white background, `font-size` about 3rem+ bold for status via `clamp()`, readings in black at normal size, red/grey classes, `meta viewport`, 60 s `setInterval` fetch, graceful display if the fetch fails. - `internal/web/handler_test.go`, `internal/web/handler.go`: `/api/events` headers (`text/event-stream`, `no-store`), initial `update`, `update` after a notification, keep-alive with a shortened interval, handler returns and cancels on client context cancel, non-flushable writer -> 500; existing tests updated for the new `NewHandler` signature; page test asserts `EventSource` and `/api/events` are present and `60000` fallback remains.
- `main.go`: env config with defaults, wiring, signal-based shutdown. - `internal/web/static/index.html`: EventSource wiring, in-flight refresh coalescing, polling fallback.
- `go.mod`, `go.sum`: add `github.com/eclipse/paho.mqtt.golang`. - `main.go`, `main_test.go`: create the hub, set `sampler.OnReceive`, pass to `NewHandler`, shutdown handling for streams; keep `WriteTimeout`; test that shutdown completes promptly with an open stream if feasible.
- Documentation: code comments; README deferred to T-003. - Documentation (explicit scope): `README.md` — describe `GET /api/events` (event format, keep-alive), that the page refreshes on each MQTT message with 60 s polling as fallback, and update the statements at lines ~13 and ~76 that say the page only refreshes every 60 s; note that reverse proxies must not buffer `text/event-stream`. Update code comments in `main.go` (`loadConfig` comment says the page always polls every 60 s) and package docs for `Hub` and `OnReceive`. `ROADMAP.md` already updated.
- Files to change: `internal/sensor/hub.go`, `internal/sensor/hub_test.go`, `internal/sensor/sampler.go`, `internal/sensor/sampler_test.go`, `internal/web/handler.go`, `internal/web/handler_test.go`, `internal/web/static/index.html`, `main.go`, `main_test.go`, `README.md`. Stage new files with `git add`.
### T-003 — Docker packaging and README - Suggested commit: `feat(web): refresh the laundry page instantly on each MQTT message via server-sent events`
- `Dockerfile` (multi-stage: golang builder with `CGO_ENABLED=0`, final `scratch` or `distroless/static` image, `EXPOSE 8080`).
- `docker-compose.yml` example with the env vars from this plan.
- `.dockerignore`.
- `README.md`: replace the project TODOs (Overview, Getting Started) with what it does, status rules, env var table, `docker compose up` instructions, the Tasmota `TelePeriod` note (10 s, any value works since sampling is per minute), and limitations (in-memory state, LAN only, no auth). Keep the existing AI Workflow section untouched. English only.
- Files to change: `Dockerfile`, `docker-compose.yml`, `.dockerignore`, `README.md`.
- Validation extra: `docker build .` if Docker is available; otherwise note it in the handoff.
## Validation
- `go fmt ./...`
- `go vet ./...`
- `go test ./...`
- Manual check after T-002: run with `HTTP_ADDR=:8080`, publish test payloads with `mosquitto_pub` (if available), open `/` at a phone-width viewport.
## Commit subjects (suggested)
- T-001: `feat(status): evaluate laundry state from the last three power samples`
- T-002: `feat(web): show live laundry status page fed by MQTT power readings`
- T-003: `feat(docker): package laundry status service with Docker and README`

19
.ai/REVIEW.md

@ -35,3 +35,22 @@
- Environment retry during implementation: Docker build metadata was read-only under the sandbox; approved rerun built successfully. No review validation retries or weakened assertions. - Environment retry during implementation: Docker build metadata was read-only under the sandbox; approved rerun built successfully. No review validation retries or weakened assertions.
- Workflow limitation: `aide pr --dry-run` rejected the current `main` branch. AGENTS.md requires working in the current branch, so no branch change or PR creation was performed. - Workflow limitation: `aide pr --dry-run` rejected the current `main` branch. AGENTS.md requires working in the current branch, so no branch change or PR creation was performed.
- Verdict: PASS, approved for the per-task commit under `all_task`. - Verdict: PASS, approved for the per-task commit under `all_task`.
## Task: T-004
### Review Round 1 — 2026-10-12T13:17:23Z
- Mechanical retry 1 (`go test -race ./...`): the test recorder's body read before advancing synthetic time was not synchronized with the subsequent keep-alive write. Synchronize the recorder fixture's writes and body reads; retain all assertions and expected behavior. No product race was reported.
- Retry 1 result: fmt/vet/test/race PASS after synchronizing the test recorder; no assertions weakened or removed.
### Final verification — 2026-10-12T13:18:18Z
- Findings: no required fixes or semantic concerns. Reviewed the staged and unstaged diff against T-004; status rules, JSON contract, and sampling remain unchanged. Hub broadcasts coalesce without blocking; callbacks execute after storage outside the sampler lock; SSE subscribers and blocked writes clean up on cancellation.
- Tests first: Go tests failed on missing Hub/callback/handler symbols before implementation; temporary page behavior checks failed on missing EventSource before page changes. Shutdown completion was fixed during implementation without changing its assertion.
- Validation: `go fmt ./...`, `go vet ./...`, `go test ./...`, `go test -race ./...`, and staged/unstaged diff checks PASS.
- E2E: NOT_CONFIGURED as a repository command. Temporary `node /tmp/mqtt-washingmachine-t004-smoke.mjs` PASS with isolated Mosquitto and real Chrome: initial events, multiple subscribers, invalid payload suppression, disconnect/reconnect, streams surviving the actual 10-second timeout, actual 25-second keep-alives, browser refresh on MQTT, sampled readings at 360px, and prompt clean SIGTERM shutdown with active streams.
- Page behavior: temporary `node /tmp/mqtt-washingmachine-t004-page-test.mjs` PASS for burst coalescing, 60-second polling, no EventSource support, and failed-fetch recovery. Screenshot inspected: `/tmp/mqtt-washingmachine-t004-mobile.png`.
- Documentation: README covers SSE format, refresh behavior, sampling, keep-alives, proxy buffering, and manual checks; AI Workflow section is byte-for-byte unchanged.
- Environment retries: Go formatting/testing needed writable build-cache access; staging needed writable Git-index access. Approved reruns passed.
- Workflow limitation: `aide pr --dry-run` refuses the current `main` branch. Kept the current branch as instructed; no PR created.
- Verdict: PASS, approved for the per-task commit under `all_task`.

1
.ai/TASKS.md

@ -23,3 +23,4 @@ Command expectations:
| T-001 | Status evaluation logic (3-sample window, 3.00 W threshold, waiting/incomplete/running/done) and Tasmota payload parsing, with tests first | done | All state/boundary tests pass (2.99 idle, 3.00 active, gap -> incomplete, <3 samples -> waiting); payload parser handles valid, missing ENERGY/Power, and invalid JSON; fmt/vet/test pass | 2026-10-12T11:03:20Z: tests first, fmt/vet/test PASS; review PASS | none | | T-001 | Status evaluation logic (3-sample window, 3.00 W threshold, waiting/incomplete/running/done) and Tasmota payload parsing, with tests first | done | All state/boundary tests pass (2.99 idle, 3.00 active, gap -> incomplete, <3 samples -> waiting); payload parser handles valid, missing ENERGY/Power, and invalid JSON; fmt/vet/test pass | 2026-10-12T11:03:20Z: tests first, fmt/vet/test PASS; review PASS | none |
| T-002 | Per-minute sampler, MQTT subscription, HTTP `/api/status` and phone-first page, `main.go` env config | done | Sampler records valid/missing samples per tick; `/api/status` matches JSON contract; page has white background, large status, `47.70W` readings, viewport meta, 60 s refresh; fmt/vet/test pass | 2026-10-12T11:17:06Z: fmt/vet/test/race PASS; MQTT and 360px Chrome smoke PASS | none | | T-002 | Per-minute sampler, MQTT subscription, HTTP `/api/status` and phone-first page, `main.go` env config | done | Sampler records valid/missing samples per tick; `/api/status` matches JSON contract; page has white background, large status, `47.70W` readings, viewport meta, 60 s refresh; fmt/vet/test pass | 2026-10-12T11:17:06Z: fmt/vet/test/race PASS; MQTT and 360px Chrome smoke PASS | none |
| T-003 | Dockerfile, docker-compose example, .dockerignore, README documentation | done | Image builds (or noted if Docker unavailable); README documents rules, env vars, deployment, limitations; existing AI Workflow section untouched | 2026-10-12T11:22:18Z: fmt/vet/test PASS; Docker build/runtime PASS; README workflow preserved | none | | T-003 | Dockerfile, docker-compose example, .dockerignore, README documentation | done | Image builds (or noted if Docker unavailable); README documents rules, env vars, deployment, limitations; existing AI Workflow section untouched | 2026-10-12T11:22:18Z: fmt/vet/test PASS; Docker build/runtime PASS; README workflow preserved | none |
| T-004 | SSE push (`GET /api/events`) so the page refreshes on each valid MQTT message: sensor Hub, `Sampler.OnReceive`, SSE handler with keep-alive and write-deadline override, page EventSource with 60 s polling fallback, README/comment updates, tests first | done | Valid message -> `update` event to all clients; invalid -> none; immediate `update` on connect; keep-alive sent; stream outlives 10 s WriteTimeout; slow subscriber never blocks Receive; disconnects/shutdown clean up; `/api/status` unchanged; README documents endpoint and updated refresh behavior; fmt/vet/test (+race) pass | 2026-10-12T13:18:18Z: fmt/vet/test/race PASS; MQTT/SSE/Chrome smoke PASS; review PASS | none |

41
README.md

@ -10,7 +10,9 @@ readings appear as `--`.
The backend runs continuously, even with no page open. Every 60 seconds it records The backend runs continuously, even with no page open. Every 60 seconds it records
the latest valid message received since the previous tick, or a missing sample if the latest valid message received since the previous tick, or a missing sample if
none arrived. It keeps only the last three samples in memory, oldest first. The none arrived. It keeps only the last three samples in memory, oldest first. The
page fetches the current status on load and every 60 seconds. page fetches the current status on load and whenever a valid MQTT power message
arrives, using Server-Sent Events (SSE). It also polls every 60 seconds as a fallback.
The readings still show the last three interval samples, not a live measurement.
### Status rules ### Status rules
@ -73,7 +75,8 @@ port in the Compose `ports` mapping. To change only the published host port, edi
the left side of `8080:8080` and leave `HTTP_ADDR` unchanged. the left side of `8080:8080` and leave `HTTP_ADDR` unchanged.
Changing `SAMPLE_INTERVAL` affects backend sampling and the three-sample window; Changing `SAMPLE_INTERVAL` affects backend sampling and the three-sample window;
the browser still refreshes every 60 seconds. Invalid or nonpositive intervals the browser still refreshes on valid MQTT messages with a 60-second polling
fallback. Invalid or nonpositive intervals
stop startup with an error. The service retries an unavailable broker in the stop startup with an error. The service retries an unavailable broker in the
background and subscribes again after reconnecting. HTTP remains available during background and subscribes again after reconnecting. HTTP remains available during
broker outages; intervals without valid messages become missing data. broker outages; intervals without valid messages become missing data.
@ -121,8 +124,10 @@ mosquitto_pub -h <test-broker> -t <test-topic> -m '{"ENERGY":{"Power":2.99}}'
After three valid samples, `/api/status` should report `done`. Publish `3.00` to After three valid samples, `/api/status` should report `done`. Publish `3.00` to
get `running`, then stop publishing for a sample interval to get `incomplete`. get `running`, then stop publishing for a sample interval to get `incomplete`.
Reload the page to inspect each result immediately. Check the page at 360 px width Keep the page open: valid messages trigger status fetches without a reload, while
for readable text and no horizontal scrolling. missing intervals appear on the next fallback poll when no messages arrive.
Readings only change after a sampling tick. Check the page at 360 px width for
readable text and no horizontal scrolling.
### Status API ### Status API
@ -136,6 +141,34 @@ readings ordered oldest first. Responses are not cached. For example:
Use `valid` to distinguish a missing sample from a real zero-watt reading. Startup Use `valid` to distinguish a missing sample from a real zero-watt reading. Startup
returns `{"state":"waiting","message":"Waiting for data...","readings":[]}`. returns `{"state":"waiting","message":"Waiting for data...","readings":[]}`.
### Status events
`GET /api/events` streams `text/event-stream` with `Cache-Control: no-store`.
It sends an immediate event on connection (including reconnection), then an
`update` notification after each valid MQTT power message:
```text
event: update
data: {}
```
Each event ends with a blank line. Events contain no measurements; fetch
`/api/status` for the current sampled state. Invalid payloads do not notify
clients. Bursts may coalesce for slow subscribers without delaying MQTT handling
or other clients. An idle connection receives `: keep-alive` followed by a blank
line every 25 seconds. Streams outlive the ordinary HTTP write timeout and close
when the client disconnects or the service shuts down.
The page uses `EventSource`, which reconnects automatically. It refreshes on each
`update`, skips overlapping status fetches, and retains the 60-second poll even
when streaming is unavailable or the browser has no `EventSource` support.
If using a reverse proxy, disable buffering for `text/event-stream` and allow
idle connections longer than the 25-second keep-alive interval.
To inspect events manually, run `curl -N http://localhost:8080/api/events` while
publishing valid and invalid payloads to the test topic.
### Limitations ### Limitations
- In-memory state only: a restart clears the window. With the default interval, - In-memory state only: a restart clears the window. With the default interval,

18
ROADMAP.md

@ -36,3 +36,21 @@ Objective: show the laundry status from the last 3 one-minute power samples of t
### Notes ### Notes
- State is in memory only. After a backend restart the page shows "Waiting for data..." until 3 samples are collected (about 3 minutes). - State is in memory only. After a backend restart the page shows "Waiting for data..." until 3 samples are collected (about 3 minutes).
## Priority 2
Objective: the page refreshes itself whenever a new MQTT power message arrives, instead of waiting for the 60 s poll.
### Decided
- Mechanism: Server-Sent Events. New endpoint `GET /api/events` (name adjustable during planning) streams an event each time the backend receives a valid `ENERGY.Power` message (about every 10 s).
- The event is only a notification. The page reacts by re-fetching `/api/status`, so the status rule, JSON contract and 1-minute sampling stay unchanged.
- Readings still show the last 3 one-minute samples; they visibly change once per minute. No live-reading line is added.
- Page uses `EventSource` (auto-reconnect). The 60 s poll stays as a fallback if the stream is down; on reconnect the page refreshes once.
- Backend: the sampler (or MQTT handler) broadcasts to all connected SSE clients without blocking on slow clients. The HTTP `WriteTimeout` of 10 s must not kill the stream (per-route override or periodic keep-alive).
- Invalid payloads do not trigger an event.
- Docs: README documents the new endpoint and behavior.
### Open
- None.

41
internal/sensor/hub.go

@ -0,0 +1,41 @@
package sensor
import "sync"
// Hub broadcasts refresh notifications. Its zero value is ready to use, and all
// methods are concurrency-safe. Bursts coalesce so slow subscribers never block
// the sender or other subscribers.
type Hub struct {
mu sync.Mutex
subscribers map[chan struct{}]struct{}
}
// Subscribe registers a buffered notification channel. The caller must cancel
// when finished; cancellation is idempotent and leaves the channel open.
func (h *Hub) Subscribe() (<-chan struct{}, func()) {
events := make(chan struct{}, 1)
h.mu.Lock()
if h.subscribers == nil {
h.subscribers = make(map[chan struct{}]struct{})
}
h.subscribers[events] = struct{}{}
h.mu.Unlock()
return events, func() {
h.mu.Lock()
delete(h.subscribers, events)
h.mu.Unlock()
}
}
// Notify queues at most one pending notification per subscriber without waiting
// for readers. Notifications carry no data; clients fetch the current snapshot.
func (h *Hub) Notify() {
h.mu.Lock()
defer h.mu.Unlock()
for events := range h.subscribers {
select {
case events <- struct{}{}:
default:
}
}
}

75
internal/sensor/hub_test.go

@ -0,0 +1,75 @@
package sensor
import (
"sync"
"testing"
"time"
)
func TestHubDeliveryCoalescingAndCancel(t *testing.T) {
var hub Hub
slow, cancelSlow := hub.Subscribe()
fast, cancelFast := hub.Subscribe()
defer cancelFast()
done := make(chan struct{})
go func() {
for range 1000 {
hub.Notify()
}
close(done)
}()
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("slow subscribers blocked notification")
}
for _, events := range []<-chan struct{}{slow, fast} {
select {
case <-events:
default:
t.Fatal("subscriber missed notification")
}
select {
case <-events:
t.Fatal("burst was not coalesced")
default:
}
}
cancelSlow()
cancelSlow()
hub.Notify()
select {
case _, ok := <-slow:
if ok {
t.Fatal("cancelled subscriber received a new event")
}
default:
}
select {
case <-fast:
default:
t.Fatal("remaining subscriber missed notification")
}
}
func TestHubConcurrentUse(t *testing.T) {
var hub Hub
hub.Notify() // The zero value also supports notification without subscribers.
var workers sync.WaitGroup
for range 10 {
workers.Go(func() {
for range 100 {
events, cancel := hub.Subscribe()
hub.Notify()
select {
case <-events:
default:
t.Error("subscriber missed notification")
}
cancel()
cancel()
}
})
}
workers.Wait()
}

10
internal/sensor/sampler.go

@ -11,6 +11,11 @@ import (
// Sampler collects the latest valid message in each interval, independently of // Sampler collects the latest valid message in each interval, independently of
// HTTP requests. Its zero value is ready to use; all methods are concurrency-safe. // HTTP requests. Its zero value is ready to use; all methods are concurrency-safe.
type Sampler struct { type Sampler struct {
// OnReceive, if set, runs after each valid message is stored and the lock
// is released. Set it before using the sampler; it must be concurrency-safe
// and return promptly so it does not delay MQTT handling.
OnReceive func()
mu sync.Mutex mu sync.Mutex
window status.Window window status.Window
power float64 power float64
@ -25,8 +30,11 @@ func (s *Sampler) Receive(payload []byte) error {
return err return err
} }
s.mu.Lock() s.mu.Lock()
defer s.mu.Unlock()
s.power, s.receivedAt, s.received = power, time.Now(), true s.power, s.receivedAt, s.received = power, time.Now(), true
s.mu.Unlock()
if s.OnReceive != nil {
s.OnReceive()
}
return nil return nil
} }

35
internal/sensor/sampler_test.go

@ -83,6 +83,41 @@ func TestSamplerClosedTicks(t *testing.T) {
} }
} }
func TestSamplerOnReceive(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
var sampler Sampler
calls := 0
ticks := make(chan time.Time)
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
go sampler.Run(ctx, ticks)
sampler.OnReceive = func() {
calls++
// Both operations acquire the sampler lock: the callback must run
// after storage and outside that lock.
ticks <- time.Now()
synctest.Wait()
got := sampler.Status().Readings
if got[len(got)-1] != (status.Sample{Power: 47.7, Valid: true}) {
t.Errorf("callback ran before storage: %+v", got)
}
}
for range 2 {
if err := sampler.Receive([]byte(`{"ENERGY":{"Power":47.7}}`)); err != nil {
t.Fatal(err)
}
}
for _, payload := range []string{`broken`, `{}`, `{"ENERGY":{}}`, `{"ENERGY":{"Power":"bad"}}`} {
if err := sampler.Receive([]byte(payload)); err == nil {
t.Fatalf("accepted %s", payload)
}
}
if calls != 2 {
t.Fatalf("callback calls = %d; want 2", calls)
}
})
}
func TestSamplerConcurrentAccess(t *testing.T) { func TestSamplerConcurrentAccess(t *testing.T) {
var sampler Sampler var sampler Sampler
ctx, cancel := context.WithCancel(t.Context()) ctx, cancel := context.WithCancel(t.Context())

78
internal/web/handler.go

@ -2,9 +2,12 @@
package web package web
import ( import (
"context"
_ "embed" _ "embed"
"encoding/json" "encoding/json"
"errors"
"net/http" "net/http"
"time"
"mqtt_washingmachine/internal/status" "mqtt_washingmachine/internal/status"
) )
@ -12,9 +15,17 @@ import (
//go:embed static/index.html //go:embed static/index.html
var indexHTML []byte var indexHTML []byte
var keepAliveInterval = 25 * time.Second
// EventSource provides coalesced refresh notifications and subscriber cleanup.
type EventSource interface {
Subscribe() (<-chan struct{}, func())
}
// NewHandler fetches a fresh snapshot for each API request. Disabling caching // NewHandler fetches a fresh snapshot for each API request. Disabling caching
// keeps reloads and polling from showing a previous laundry state. // keeps reloads, push refreshes, and polling from showing a previous state.
func NewHandler(snapshot func() status.Result) http.Handler { // The server must cancel request contexts on shutdown to close active streams.
func NewHandler(snapshot func() status.Result, events EventSource) http.Handler {
mux := http.NewServeMux() mux := http.NewServeMux()
mux.HandleFunc("GET /{$}", func(w http.ResponseWriter, r *http.Request) { mux.HandleFunc("GET /{$}", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/html; charset=utf-8") w.Header().Set("Content-Type", "text/html; charset=utf-8")
@ -26,5 +37,68 @@ func NewHandler(snapshot func() status.Result) http.Handler {
w.Header().Set("Cache-Control", "no-store") w.Header().Set("Cache-Control", "no-store")
json.NewEncoder(w).Encode(snapshot()) json.NewEncoder(w).Encode(snapshot())
}) })
mux.HandleFunc("GET /api/events", func(w http.ResponseWriter, r *http.Request) {
serveEvents(w, r, events)
})
return mux return mux
} }
func serveEvents(w http.ResponseWriter, r *http.Request, source EventSource) {
if _, ok := w.(http.Flusher); !ok {
http.Error(w, "Streaming is not supported", http.StatusInternalServerError)
return
}
controller := http.NewResponseController(w)
// Streaming outlives the server's ordinary request write timeout. Test
// recorders may support flushing without implementing write deadlines.
if err := controller.SetWriteDeadline(time.Time{}); err != nil && !errors.Is(err, http.ErrNotSupported) {
http.Error(w, "Cannot start stream", http.StatusInternalServerError)
return
}
// Cancellation must also interrupt a write blocked by an unresponsive
// client. Wait for this callback before returning so it cannot use the
// ResponseWriter after the handler finishes.
interrupted := make(chan struct{})
stop := context.AfterFunc(r.Context(), func() {
controller.SetWriteDeadline(time.Now())
close(interrupted)
})
defer func() {
if !stop() {
<-interrupted
// Let net/http finish an idle stream's chunked response normally.
// A failed blocked write already carries its error in the writer.
controller.SetWriteDeadline(time.Time{})
}
}()
events, cancel := source.Subscribe()
defer cancel()
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-store")
write := func(frame string) bool {
if _, err := w.Write([]byte(frame)); err != nil {
return false
}
return controller.Flush() == nil
}
const update = "event: update\ndata: {}\n\n"
if !write(update) {
return
}
ticker := time.NewTicker(keepAliveInterval)
defer ticker.Stop()
for {
select {
case <-r.Context().Done():
return
case <-events:
if !write(update) {
return
}
case <-ticker.C:
if !write(": keep-alive\n\n") {
return
}
}
}
}

202
internal/web/handler_test.go

@ -1,13 +1,21 @@
package web package web
import ( import (
"bufio"
"context"
"encoding/json" "encoding/json"
"errors"
"io"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"reflect" "reflect"
"strings" "strings"
"sync"
"testing" "testing"
"testing/synctest"
"time"
"mqtt_washingmachine/internal/sensor"
"mqtt_washingmachine/internal/status" "mqtt_washingmachine/internal/status"
) )
@ -23,7 +31,7 @@ func TestStatusAPI(t *testing.T) {
{"done", "done", "Laundry done", []status.Sample{{Power: 2.99, Valid: true}, {Valid: true}, {Valid: true}}}, {"done", "done", "Laundry done", []status.Sample{{Power: 2.99, Valid: true}, {Valid: true}, {Valid: true}}},
} { } {
t.Run(tt.name, func(t *testing.T) { t.Run(tt.name, func(t *testing.T) {
handler := NewHandler(func() status.Result { return status.Evaluate(tt.readings) }) handler := NewHandler(func() status.Result { return status.Evaluate(tt.readings) }, &sensor.Hub{})
response := httptest.NewRecorder() response := httptest.NewRecorder()
handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/api/status", nil)) handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/api/status", nil))
if response.Code != http.StatusOK { if response.Code != http.StatusOK {
@ -48,13 +56,13 @@ func TestStatusAPI(t *testing.T) {
} }
func TestPageAndRouting(t *testing.T) { func TestPageAndRouting(t *testing.T) {
handler := NewHandler(func() status.Result { return status.Evaluate(nil) }) handler := NewHandler(func() status.Result { return status.Evaluate(nil) }, &sensor.Hub{})
page := httptest.NewRecorder() page := httptest.NewRecorder()
handler.ServeHTTP(page, httptest.NewRequest(http.MethodGet, "/", nil)) handler.ServeHTTP(page, httptest.NewRequest(http.MethodGet, "/", nil))
if page.Code != http.StatusOK || !strings.HasPrefix(page.Header().Get("Content-Type"), "text/html") { if page.Code != http.StatusOK || !strings.HasPrefix(page.Header().Get("Content-Type"), "text/html") {
t.Fatalf("page response: %+v", page.Result()) t.Fatalf("page response: %+v", page.Result())
} }
for _, text := range []string{`name="viewport"`, "Waiting for data...", "/api/status", "60000", "toFixed(2)", "--", "clamp("} { for _, text := range []string{`name="viewport"`, "Waiting for data...", "/api/status", "60000", "EventSource", "/api/events", "toFixed(2)", "--", "clamp("} {
if !strings.Contains(page.Body.String(), text) { if !strings.Contains(page.Body.String(), text) {
t.Errorf("page missing %q", text) t.Errorf("page missing %q", text)
} }
@ -65,6 +73,7 @@ func TestPageAndRouting(t *testing.T) {
}{ }{
{http.MethodGet, "/missing", http.StatusNotFound}, {http.MethodGet, "/missing", http.StatusNotFound},
{http.MethodPost, "/api/status", http.StatusMethodNotAllowed}, {http.MethodPost, "/api/status", http.StatusMethodNotAllowed},
{http.MethodPost, "/api/events", http.StatusMethodNotAllowed},
{http.MethodPost, "/", http.StatusMethodNotAllowed}, {http.MethodPost, "/", http.StatusMethodNotAllowed},
} { } {
response := httptest.NewRecorder() response := httptest.NewRecorder()
@ -77,7 +86,7 @@ func TestPageAndRouting(t *testing.T) {
func TestStatusAPIReadsCurrentSnapshot(t *testing.T) { func TestStatusAPIReadsCurrentSnapshot(t *testing.T) {
var window status.Window var window status.Window
handler := NewHandler(func() status.Result { return status.Evaluate(window.Readings()) }) handler := NewHandler(func() status.Result { return status.Evaluate(window.Readings()) }, &sensor.Hub{})
for range 3 { for range 3 {
window.Add(status.Sample{Valid: true}) window.Add(status.Sample{Valid: true})
response := httptest.NewRecorder() response := httptest.NewRecorder()
@ -91,3 +100,188 @@ func TestStatusAPIReadsCurrentSnapshot(t *testing.T) {
} }
} }
} }
type trackedEvents struct {
events chan struct{}
subscribed bool
cancelled bool
}
func (e *trackedEvents) Subscribe() (<-chan struct{}, func()) {
e.subscribed = true
return e.events, func() { e.cancelled = true }
}
// Synthetic time orders timer wakeups but does not establish a race-detector
// happens-before edge from a body inspection to the next timed write.
type streamRecorder struct {
*httptest.ResponseRecorder
mu sync.Mutex
}
func (w *streamRecorder) Write(p []byte) (int, error) {
w.mu.Lock()
defer w.mu.Unlock()
return w.ResponseRecorder.Write(p)
}
func (w *streamRecorder) body() string {
w.mu.Lock()
defer w.mu.Unlock()
return w.Body.String()
}
func TestEventsInitialNotificationKeepAliveAndCleanup(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
oldInterval := keepAliveInterval
keepAliveInterval = time.Second
defer func() { keepAliveInterval = oldInterval }()
events := &trackedEvents{events: make(chan struct{}, 1)}
handler := NewHandler(func() status.Result { return status.Evaluate(nil) }, events)
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
request := httptest.NewRequest(http.MethodGet, "/api/events", nil).WithContext(ctx)
response := &streamRecorder{ResponseRecorder: httptest.NewRecorder()}
done := make(chan struct{})
go func() { handler.ServeHTTP(response, request); close(done) }()
synctest.Wait()
if !events.subscribed || !response.Flushed || response.Code != http.StatusOK {
t.Fatal("stream did not subscribe and flush immediately")
}
if response.Header().Get("Content-Type") != "text/event-stream" || response.Header().Get("Cache-Control") != "no-store" {
t.Fatalf("stream headers: %v", response.Header())
}
if got := response.body(); got != "event: update\ndata: {}\n\n" {
t.Fatalf("initial event = %q", got)
}
events.events <- struct{}{}
synctest.Wait()
if got := strings.Count(response.body(), "event: update\n"); got != 2 {
t.Fatalf("update count = %d; want 2", got)
}
time.Sleep(time.Second)
synctest.Wait()
if !strings.HasSuffix(response.body(), ": keep-alive\n\n") {
t.Fatalf("missing keep-alive: %q", response.body())
}
cancel()
synctest.Wait()
select {
case <-done:
default:
t.Fatal("handler did not exit after disconnect")
}
if !events.cancelled {
t.Fatal("disconnected subscriber was not removed")
}
})
}
type nonFlushingWriter struct{ http.ResponseWriter }
func TestEventsRejectNonFlushingWriter(t *testing.T) {
events := &trackedEvents{events: make(chan struct{}, 1)}
handler := NewHandler(func() status.Result { return status.Evaluate(nil) }, events)
response := httptest.NewRecorder()
handler.ServeHTTP(nonFlushingWriter{response}, httptest.NewRequest(http.MethodGet, "/api/events", nil))
if response.Code != http.StatusInternalServerError || events.subscribed {
t.Fatalf("unsupported stream: status %d, subscribed %v", response.Code, events.subscribed)
}
}
type blockedStreamWriter struct {
*httptest.ResponseRecorder
started chan struct{}
unblock chan struct{}
once sync.Once
cleared bool
}
func (w *blockedStreamWriter) Write(p []byte) (int, error) {
close(w.started)
<-w.unblock
return 0, errors.New("write deadline exceeded")
}
func (w *blockedStreamWriter) SetWriteDeadline(deadline time.Time) error {
if deadline.IsZero() {
w.cleared = true
} else {
w.once.Do(func() { close(w.unblock) })
}
return nil
}
func TestEventsCancellationInterruptsBlockedWrite(t *testing.T) {
events := &trackedEvents{events: make(chan struct{}, 1)}
handler := NewHandler(func() status.Result { return status.Evaluate(nil) }, events)
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
response := &blockedStreamWriter{ResponseRecorder: httptest.NewRecorder(), started: make(chan struct{}), unblock: make(chan struct{})}
done := make(chan struct{})
go func() {
handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/api/events", nil).WithContext(ctx))
close(done)
}()
select {
case <-response.started:
case <-time.After(time.Second):
t.Fatal("stream did not start writing")
}
cancel()
select {
case <-done:
case <-time.After(time.Second):
response.once.Do(func() { close(response.unblock) })
<-done
t.Fatal("cancellation did not interrupt the blocked write")
}
if !events.cancelled || !response.cleared {
t.Fatalf("cancelled=%v, write deadline cleared=%v", events.cancelled, response.cleared)
}
}
func TestEventsSurviveWriteTimeoutAndBroadcastValidMessages(t *testing.T) {
var hub sensor.Hub
sampler := &sensor.Sampler{OnReceive: hub.Notify}
server := httptest.NewUnstartedServer(NewHandler(sampler.Status, &hub))
server.Config.WriteTimeout = 50 * time.Millisecond
server.Start()
defer server.Close()
client := &http.Client{Timeout: 3 * time.Second}
readers := make([]*bufio.Reader, 2)
for i := range readers {
response, err := client.Get(server.URL + "/api/events")
if err != nil {
t.Fatal(err)
}
defer response.Body.Close()
readers[i] = bufio.NewReader(response.Body)
assertUpdate(t, readers[i])
}
time.Sleep(100 * time.Millisecond)
if err := sampler.Receive([]byte(`broken`)); err == nil {
t.Fatal("invalid payload accepted")
}
if err := sampler.Receive([]byte(`{"ENERGY":{"Power":47.7}}`)); err != nil {
t.Fatal(err)
}
for _, reader := range readers {
assertUpdate(t, reader)
}
if server.Config.WriteTimeout != 50*time.Millisecond {
t.Fatal("global write timeout changed")
}
}
func assertUpdate(t *testing.T, reader io.Reader) {
t.Helper()
want := "event: update\ndata: {}\n\n"
got := make([]byte, len(want))
if _, err := io.ReadFull(reader, got); err != nil {
t.Fatal(err)
}
if string(got) != want {
t.Fatalf("event = %q; want %q", got, want)
}
}

9
internal/web/static/index.html

@ -29,8 +29,11 @@
<script> <script>
const statusText = document.getElementById('status'); const statusText = document.getElementById('status');
const readings = document.getElementById('readings'); const readings = document.getElementById('readings');
let refreshing = false;
async function refresh() { async function refresh() {
if (refreshing) return;
refreshing = true;
try { try {
const response = await fetch('/api/status', { cache: 'no-store', signal: AbortSignal.timeout(10000) }); const response = await fetch('/api/status', { cache: 'no-store', signal: AbortSignal.timeout(10000) });
if (!response.ok) throw new Error('Status request failed'); if (!response.ok) throw new Error('Status request failed');
@ -47,11 +50,17 @@
statusText.textContent = 'Status unavailable'; statusText.textContent = 'Status unavailable';
statusText.className = 'unavailable'; statusText.className = 'unavailable';
readings.replaceChildren(); readings.replaceChildren();
} finally {
refreshing = false;
} }
} }
refresh(); refresh();
setInterval(refresh, 60000); setInterval(refresh, 60000);
if (typeof EventSource !== 'undefined') {
const events = new EventSource('/api/events');
events.addEventListener('update', refresh);
}
</script> </script>
</body> </body>
</html> </html>

27
main.go

@ -24,7 +24,8 @@ type config struct {
} }
// Empty variables use defaults. SAMPLE_INTERVAL accepts positive Go durations; // Empty variables use defaults. SAMPLE_INTERVAL accepts positive Go durations;
// it controls backend sampling only, while the page always polls every 60 s. // it controls backend sampling only. The page refreshes on valid MQTT messages,
// with a fixed 60 s polling fallback.
func loadConfig() (config, error) { func loadConfig() (config, error) {
env := func(key, fallback string) string { env := func(key, fallback string) string {
if value := os.Getenv(key); value != "" { if value := os.Getenv(key); value != "" {
@ -58,6 +59,8 @@ func run(ctx context.Context, cfg config) error {
ticker := time.NewTicker(cfg.SampleInterval) ticker := time.NewTicker(cfg.SampleInterval)
defer ticker.Stop() defer ticker.Stop()
var sampler sensor.Sampler var sampler sensor.Sampler
var hub sensor.Hub
sampler.OnReceive = hub.Notify
done := make(chan struct{}) done := make(chan struct{})
go func() { go func() {
defer close(done) defer close(done)
@ -66,12 +69,7 @@ func run(ctx context.Context, cfg config) error {
defer func() { cancel(); <-done }() defer func() { cancel(); <-done }()
client := sensor.StartMQTT(cfg.MQTTConfig, &sampler, slog.Default()) client := sensor.StartMQTT(cfg.MQTTConfig, &sampler, slog.Default())
defer client.Disconnect(250) defer client.Disconnect(250)
server := &http.Server{ server := newHTTPServer(ctx, web.NewHandler(sampler.Status, &hub))
Handler: web.NewHandler(sampler.Status),
ReadHeaderTimeout: 5 * time.Second,
WriteTimeout: 10 * time.Second,
IdleTimeout: 60 * time.Second,
}
errorsCh := make(chan error, 1) errorsCh := make(chan error, 1)
go func() { errorsCh <- server.Serve(listener) }() go func() { errorsCh <- server.Serve(listener) }()
slog.Info("Laundry status server listening", "address", listener.Addr().String()) slog.Info("Laundry status server listening", "address", listener.Addr().String())
@ -89,6 +87,21 @@ func run(ctx context.Context, cfg config) error {
} }
} }
// newHTTPServer cancels stream contexts when shutdown starts, allowing active
// SSE handlers (including blocked writers) to exit before Shutdown waits on them.
func newHTTPServer(ctx context.Context, handler http.Handler) *http.Server {
streams, cancelStreams := context.WithCancel(ctx)
server := &http.Server{
Handler: handler,
BaseContext: func(net.Listener) context.Context { return streams },
ReadHeaderTimeout: 5 * time.Second,
WriteTimeout: 10 * time.Second,
IdleTimeout: 60 * time.Second,
}
server.RegisterOnShutdown(cancelStreams)
return server
}
func main() { func main() {
cfg, err := loadConfig() cfg, err := loadConfig()
if err == nil { if err == nil {

44
main_test.go

@ -1,8 +1,15 @@
package main package main
import ( import (
"context"
"io"
"net"
"net/http"
"testing" "testing"
"time" "time"
"mqtt_washingmachine/internal/sensor"
"mqtt_washingmachine/internal/web"
) )
func TestLoadConfigDefaults(t *testing.T) { func TestLoadConfigDefaults(t *testing.T) {
@ -18,6 +25,43 @@ func TestLoadConfigDefaults(t *testing.T) {
} }
} }
func TestHTTPServerShutdownWithOpenStream(t *testing.T) {
var sampler sensor.Sampler
var hub sensor.Hub
server := newHTTPServer(t.Context(), web.NewHandler(sampler.Status, &hub))
if server.WriteTimeout != 10*time.Second {
t.Fatalf("write timeout = %v", server.WriteTimeout)
}
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
defer server.Close()
served := make(chan error, 1)
go func() { served <- server.Serve(listener) }()
client := &http.Client{Timeout: 3 * time.Second}
response, err := client.Get("http://" + listener.Addr().String() + "/api/events")
if err != nil {
t.Fatal(err)
}
defer response.Body.Close()
initial := make([]byte, len("event: update\ndata: {}\n\n"))
if _, err := io.ReadFull(response.Body, initial); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(t.Context(), time.Second)
defer cancel()
if err := server.Shutdown(ctx); err != nil {
t.Fatalf("open stream delayed shutdown: %v", err)
}
if body, err := io.ReadAll(response.Body); err != nil || len(body) != 0 {
t.Fatalf("stream did not close cleanly: %q, %v", body, err)
}
if err := <-served; err != http.ErrServerClosed {
t.Fatalf("Serve = %v", err)
}
}
func TestLoadConfigOverrides(t *testing.T) { func TestLoadConfigOverrides(t *testing.T) {
for key, value := range map[string]string{"MQTT_BROKER": "tcp://localhost:1884", "MQTT_TOPIC": "test/power", "MQTT_USERNAME": "tester", "MQTT_PASSWORD": "test-password", "HTTP_ADDR": "127.0.0.1:9090", "SAMPLE_INTERVAL": "250ms"} { for key, value := range map[string]string{"MQTT_BROKER": "tcp://localhost:1884", "MQTT_TOPIC": "test/power", "MQTT_USERNAME": "tester", "MQTT_PASSWORD": "test-password", "HTTP_ADDR": "127.0.0.1:9090", "SAMPLE_INTERVAL": "250ms"} {
t.Setenv(key, value) t.Setenv(key, value)

Loading…
Cancel
Save