Public: anyone can read this; the crew can change it.
Migrated from github.com/Trevato/SO-ARM100
workbenchthis machine
/care refreshes itself when dosing changes: a mode change, a resume or a stop sends an event on /care/events #620
Check · Review · Land done · Live — Landed; nothing deploys from this branch, so this is the end of its path.
smith/arm:smith/618-care-refreshes-itself-when-dosing-change → main 1 commit fork 1 thread · 1 unresolved
3 files
changed +81
−1
Software/so101/src/care-dosing.test.ts +52
| @@ -9,6 +9,7 @@ import { renderCare } from "./care-page.ts"; | ||
| 9 | 9 | import { careCommand, dosingLines } from "./care-terminal.ts"; |
| 10 | 10 | import { BUILT_IN_DIR, loadProfiles } from "./care-profile.ts"; |
| 11 | 11 | import { Doser, LOOPS, topOff } from "./doser.ts"; |
| 12 | import { handleCare } from "./care-routes.ts"; | |
| 12 | 13 | |
| 13 | 14 | const spinach = loadProfiles(BUILT_IN_DIR).find((p) => p.id === "spinach")!; |
| 14 | 15 | |
| @@ -214,3 +215,54 @@ test("controller fixes: only the owner changes modes or resumes, modes persist, | ||
| 214 | 215 | test("the loops are the three pumps", () => { |
| 215 | 216 | expect(LOOPS).toEqual(["nutrientA", "nutrientB", "phDown"]); |
| 216 | 217 | }); |
| 218 | ||
| 219 | test("/care/events: a mode change, a resume, a halt and a dose each send one event to the owner, and a guest hears nothing of dosing", async () => { | |
| 220 | const log = new CareLog(); | |
| 221 | const doser = fakeDoser(log); | |
| 222 | const readBody = async () => Buffer.alloc(0); | |
| 223 | let owner = true; | |
| 224 | const server = createServer((req, res) => { | |
| 225 | handleCare(req, res, new URL(req.url!, "http://x").pathname, { profile: spinach, log, doser, readBody, who: () => ({ owner, known: owner, signedIn: null }) }).then((hit) => hit || (res.writeHead(404), res.end())); | |
| 226 | }); | |
| 227 | await new Promise<void>((ok) => server.listen(0, "127.0.0.1", ok)); | |
| 228 | const url = `http://127.0.0.1:${(server.address() as { port: number }).port}/care/events`; | |
| 229 | const listen = async () => { | |
| 230 | const res = await fetch(url); | |
| 231 | const reader = res.body!.getReader(); | |
| 232 | let heard = ""; | |
| 233 | (async () => { for (;;) { const r = await reader.read(); if (r.done) return; heard += new TextDecoder().decode(r.value); } })().catch(() => {}); | |
| 234 | await Bun.sleep(30); | |
| 235 | return { events: () => heard.split("event: care").length - 1, stop: () => reader.cancel() }; | |
| 236 | }; | |
| 237 | try { | |
| 238 | const mine = await listen(); | |
| 239 | owner = false; | |
| 240 | const guest = await listen(); | |
| 241 | const settle = () => Bun.sleep(40); | |
| 242 | doser.setMode("phDown", "auto", "owner"); | |
| 243 | await settle(); | |
| 244 | expect(mine.events()).toBe(1); | |
| 245 | doser.resume("owner"); | |
| 246 | await settle(); | |
| 247 | expect(mine.events()).toBe(2); | |
| 248 | log.add({ type: "measurement", kind: "ph", value: 13 }); // outside what water can read: a dose halts | |
| 249 | await settle(); | |
| 250 | const logged = mine.events(); | |
| 251 | expect((await doser.dose("phDown", "owner")).ok).toBe(false); | |
| 252 | await settle(); | |
| 253 | expect(doser.halted).not.toBeNull(); | |
| 254 | expect(mine.events()).toBe(logged + 1); | |
| 255 | doser.resume("owner"); | |
| 256 | log.add({ type: "measurement", kind: "ph", value: 7.2 }); | |
| 257 | await settle(); | |
| 258 | const before = mine.events(); | |
| 259 | expect((await doser.dose("phDown", "owner")).ok).toBe(true); // the dose also writes a log entry: still one event | |
| 260 | await settle(); | |
| 261 | expect(mine.events()).toBe(before + 1); | |
| 262 | expect(guest.events()).toBe(3); // the three log entries only: no mode, resume or halt | |
| 263 | await mine.stop(); | |
| 264 | await guest.stop(); | |
| 265 | } finally { | |
| 266 | server.close(); | |
| 267 | } | |
| 268 | }); |
Software/so101/src/care-routes.ts +10 −1
| @@ -58,7 +58,16 @@ export async function handleCare(req: IncomingMessage, res: ServerResponse, path | ||
| 58 | 58 | if (req.method === "GET" && pathname === "/care/events") { |
| 59 | 59 | res.writeHead(200, { "content-type": "text/event-stream", "cache-control": "no-store", "x-accel-buffering": "no" }); |
| 60 | 60 | res.write("retry: 5000\n\n"); |
| 61 | const stop = ctx.log.onChange(() => res.write("event: care\ndata: changed\n\n")); | |
| 61 | // one event per change, even when a dose also writes a log entry; dosing is the owner's, so a guest hears nothing of it | |
| 62 | let queued = false; | |
| 63 | const ping = () => { | |
| 64 | if (queued) return; | |
| 65 | queued = true; | |
| 66 | setTimeout(() => { queued = false; res.write("event: care\ndata: changed\n\n"); }, 0); | |
| 67 | }; | |
| 68 | const stopLog = ctx.log.onChange(ping); | |
| 69 | const stopDoser = me.owner ? ctx.doser?.onChange(ping) : undefined; | |
| 70 | const stop = () => { stopLog(); stopDoser?.(); }; | |
| 62 | 71 | const beat = setInterval(() => res.write(": \n\n"), 25_000); |
| 63 | 72 | res.on("close", () => { stop(); clearInterval(beat); }); |
| 64 | 73 | return true; |
Software/so101/src/doser.ts +19
| @@ -110,6 +110,7 @@ export class Doser { | ||
| 110 | 110 | readonly loops: Record<LoopId, LoopConfig>; |
| 111 | 111 | private s: Saved = { halted: null, lastDoseAt: 0, usedReading: {}, pending: null, doses: [], modes: {} }; |
| 112 | 112 | private busy = false; |
| 113 | private readonly listeners = new Set<() => void>(); | |
| 113 | 114 | |
| 114 | 115 | private readonly d: DoserDeps; |
| 115 | 116 | |
| @@ -130,6 +131,15 @@ export class Doser { | ||
| 130 | 131 | } |
| 131 | 132 | } |
| 132 | 133 | |
| 134 | /** Calls `fn` after a mode change, a resume, a halt or a dose; returns the way to stop. */ | |
| 135 | onChange(fn: () => void): () => void { | |
| 136 | this.listeners.add(fn); | |
| 137 | return () => void this.listeners.delete(fn); | |
| 138 | } | |
| 139 | private changed() { | |
| 140 | for (const fn of this.listeners) fn(); | |
| 141 | } | |
| 142 | ||
| 133 | 143 | private now() { |
| 134 | 144 | return this.d.now?.() ?? Date.now(); |
| 135 | 145 | } |
| @@ -149,6 +159,7 @@ export class Doser { | ||
| 149 | 159 | this.s.halted = null; |
| 150 | 160 | this.s.pending = null; |
| 151 | 161 | this.save(); |
| 162 | this.changed(); | |
| 152 | 163 | } |
| 153 | 164 | /** Only the owner changes a mode, and it is kept across a restart. */ |
| 154 | 165 | setMode(loop: LoopId, mode: Mode, cause: Cause) { |
| @@ -158,6 +169,7 @@ export class Doser { | ||
| 158 | 169 | this.loops[loop].mode = mode; |
| 159 | 170 | this.s.modes[loop] = mode; |
| 160 | 171 | this.save(); |
| 172 | this.changed(); | |
| 161 | 173 | } |
| 162 | 174 | /** What a page or the terminal shows: per loop its mode, what it would add and what it has used today. */ |
| 163 | 175 | status() { |
| @@ -238,6 +250,13 @@ export class Doser { | ||
| 238 | 250 | * With no `ml`, the suggested amount. |
| 239 | 251 | */ |
| 240 | 252 | async dose(loop: LoopId, cause: Cause, ml?: number): Promise<Outcome> { |
| 253 | const halted = this.s.halted; | |
| 254 | const out = await this.run(loop, cause, ml); | |
| 255 | if (out.ok || this.s.halted !== halted) this.changed(); | |
| 256 | return out; | |
| 257 | } | |
| 258 | ||
| 259 | private async run(loop: LoopId, cause: Cause, ml?: number): Promise<Outcome> { | |
| 241 | 260 | const c = this.loops[loop]; |
| 242 | 261 | if (this.busy) return refuse("a dose is running"); |
| 243 | 262 | if (this.s.halted) return refuse(`stopped: ${this.s.halted}`); |
The setTimeout is not cleared when the stream closes. A ping queued just before disconnect writes to a closed response. Keep the handle and clear it in the close handler.