Практика: поток показаний приборов учёта (Meter readings stream) ​

Счётчики воды присылают показания, а тревога, контроль связи и суточный расход подписаны на поток: подписчики хранят состояние между событиями, а пропажу связи видно только по тику часов.

Базовый пример: Наблюдатель. Там каждое событие самодостаточно: пришла цена, подписчик отреагировал. Здесь подписчикам нужна история (когда прибор выходил на связь, сколько набралось за сутки), а самое важное событие - то, которого не было: прибор замолчал.

Первая практика темы, доменные события заказа, про диспетчер с типами событий и каскадами. Эта про поток однотипных событий со временем: явные тики вместо системных часов, подписчики с состоянием и подписчик, который уходит сам, сделав дело.

Задача ​

Управляющая компания ставит в домах счётчики воды с радиомодулем. Каждый прибор раз в 10-30 минут присылает, сколько литров прошло с прошлого показания. Сначала был один обработчик: записать показание в базу. Потом диспетчеры попросили SMS при прорыве (больше 500 литров за одно показание), инженеры - список приборов, которые пропали со связи больше чем на час, бухгалтерия - расход за сутки, а монтажники - отметку, что новый прибор вышел на связь.

Если всё это писать в обработчике показания, он знает о SMS-шлюзе, бухгалтерии и монтаже, и упавший шлюз роняет запись расхода. А пропажу связи из обработчика показаний не увидеть вовсе: молчащий прибор ничего не присылает, и обработчик не вызывается.

Что урезано ​

• Показание - расход за интервал в литрах. Настоящие счётчики часто шлют накопленное значение, и разницу считает слой приёма до потока.
• Поток синхронный и в одном процессе. В проекте показания приходят из MQTT или очереди, а тик присылает планировщик раз в несколько минут.
• SMS - функция notify, журнал и тревоги - функции, которые пишут строку.
• Время - целые секунды Unix, сутки считаются по UTC. Часовые пояса домов, опоздавшие и повторные показания не учитываются.

Решение ​

Издатель MeterStream рассылает два вида событий: показание (publish) и тик часов (tick). Контракт подписчика MeterObserver - два метода, onReading и onTick. Тик нужен, потому что наблюдатель реагирует только на события: чтобы заметить тишину, кто-то должен сказать "сейчас 23:40". Время приходит в тике явно, подписчики не читают системные часы, и тесты двигают время числами.

Подписчики:

• ThresholdAlarm отправляет notify, если за одно показание прошло больше лимита.
• ConnectionWatch помнит, когда каждый прибор выходил на связь. На тике помечает молчащих дольше интервала и сообщает о них один раз, на показании от пропавшего прибора сообщает, что он вернулся.
• DailyUsage суммирует литры по прибору и дню UTC.
• Commissioning ждёт первое показание нового прибора, записывает время и отписывается сам, прямо во время рассылки.

subscribe возвращает функцию отписки. publish и tick ловят исключение каждого подписчика, заворачивают в SubscriberFailed с именем подписчика и продолжают рассылку. Список ошибок получает вызывающий.

Структура ​

Время и тишина ​

Подписчик с состоянием живёт между событиями, и его логика зависит от времени. Если он сам вызывает time(), тест "прибор молчит больше часа" ждёт час или подменяет системные часы. Поэтому время здесь - данные: у показания своё at, у тика своё now.

ConnectionWatch сравнивает now - lastSeen с интервалом строго: ровно 3600 секунд тишины - ещё на связи, 3601 - пропал. Пропавший прибор помечается, чтобы следующий тик не слал ту же тревогу снова. Приборы обходятся в порядке идентификаторов, иначе порядок тревог зависел бы от порядка ключей в словаре.

Отписка во время рассылки: Commissioning вызывает свою функцию отписки внутри onReading. Рассылка идёт по снимку списка подписчиков, поэтому сосед, подписанный следом, всё равно получает это показание, а со следующего показания Commissioning уже не вызывается.

Как это собирается ​

Поток собирает воркер приёма показаний при старте: подписывает тревогу, контроль связи и суточный расход, а для каждого нового прибора из заявки на монтаж - Commissioning. Дальше воркер читает сообщения из очереди и зовёт publish, а планировщик раз в несколько минут зовёт tick. Ошибки подписчиков воркер пишет в журнал и идёт дальше: показание уже получили все остальные.

В примере за вечер приходят показания трёх приборов. На 640 литрах у m1 уходит SMS. Новый m3 выходит на связь в 22:30, монтаж отмечен, Commissioning отписался. Тик в 23:40 замечает, что m2 молчит 6000 секунд, а показание после полуночи возвращает его на связь. Суточный расход раскладывает литры m2 на два дня.

Сценарий целиком с выводом: пример на PHP, он же на Go, Python и TypeScript в панели кода.

Реализации ​

Поведение и числа одинаковы во всех языках: лимит 500 литров, интервал тишины 3600 секунд, сутки по UTC, ошибка подписчика "<имя> failed: <причина>".

Запуск: make test T=patterns/behavioral/observer/practice-meters

Что проверяют тесты ​

Тесты одинаковы во всех языках, время задаётся числами от 22:00 15 октября 2026 UTC.

• Контроль связи, таблица из четырёх сценариев тиков: 3600 секунд тишины - на связи; 3601 - тревога; второй тик не повторяет тревогу; показание после тревоги возвращает прибор.
• Суточный расход: 100 и 50 литров до полуночи UTC и 70 после дают 150 и 70 в разных днях.
• Тревога: 500 литров - тишина, 501 - SMS с текстом.
• Ошибка подписчика: SMS-шлюз бросает sms gateway down, рассылка возвращает одну ошибку alarm failed: sms gateway down, а суточный расход, подписанный следом, всё равно посчитал 900 литров.
• Отписка во время рассылки: Commissioning отписывается на первом показании, сосед получает оба показания (30 литров), а время выхода на связь не перезаписано вторым.

Где граница ​

• Приборов тысячи и десятки тысяч: обходить всех на каждом тике дорого. Тогда "пропажа связи" становится задачей хранилища (запрос "последнее показание старше часа") или таймера на каждый прибор, а не подписчика в памяти.
• Показания опаздывают и приходят не по порядку: тик в 23:40 может пометить прибор пропавшим, хотя его показание за 23:30 ещё в очереди. Это уже обработка потоков с водяными знаками (watermark), как в Kafka Streams или Flink.
• Состояние подписчиков живёт в памяти и теряется при перезапуске воркера. Суточный расход в проекте пишется в базу, а подписчик только обновляет её.
• Подписчик один и не меняется - прямой вызов из обработчика проще потока.

Отличия между языками ​

• PHP - массивы копируются при присваивании, и foreach по свойству-массиву уже работает со снимком: unset из обработчика не сдвинет соседей. Отписка - замыкание, захватившее $id. Ошибки ловятся по Exception, а Error (баг в коде) пролетает наружу.
• Go - ошибок-исключений нет, подписчик возвращает error, и поток собирает их сам; SubscriberFailed реализует Unwrap, поэтому errors.Is находит исходную причину. Без slices.Clone отписка через DeleteFunc сдвинула бы элементы того же массива, по которому идёт цикл, и сосед был бы пропущен. Словари обходятся в отсортированном порядке, иначе вывод менялся бы от запуска к запуску.
• Python - Protocol проверяет подписчиков без наследования; снимок обязателен, иначе изменение словаря во время обхода бросит RuntimeError. Отписка - вложенная функция над pop(sub_id, None), повторный вызов безопасен.
• TypeScript - обход Map переживает удаление и без снимка (удалённый элемент просто не встретится), но снимок делает поведение одинаковым с другими языками и не зависит от этой тонкости. Время - числа секунд, Date нужен только для форматирования дня.

Связанные темы ​

• Наблюдатель (базовый пример)
• Практика: доменные события заказа - типы событий, каскады через очередь, ошибки подписчиков
• Практика Декоратора: слои над медленным отчётом - тоже конвейер вокруг источника данных, но слои вкладываются друг в друга, а подписчики стоят рядом