mm::node::NotificationBus Class Reference

Polls a fixed set of sources and broadcasts a notification when one of them changes. More...

#include <notification_bus.h>

Classes
struct	Source
	One feature's contribution: when to speak, and what to say. More...

Public Types
using	PublishFn = std::function< void(std::string topic, std::string json)>
	Receives a finished message and the topic to put it on. Wired to WebSocketServer::publish at the root, the same shape as MonitoringManager::setPublish.

Public Member Functions
	NotificationBus ()
	~NotificationBus ()
	Requests the stop and joins.
	NotificationBus (const NotificationBus &)=delete
NotificationBus &	operator= (const NotificationBus &)=delete
	NotificationBus (NotificationBus &&)=delete
NotificationBus &	operator= (NotificationBus &&)=delete
void	setPublish (PublishFn publish)
	Sets where finished messages go. Call before start.
void	addSource (Source source)
	Registers a source. Call before start.
void	start ()
	Starts the poll thread.
void	stop ()
	Stops the poll thread and joins it. Returns immediately if it is not running, and does not wait out the poll interval.

Detailed Description

Polls a fixed set of sources and broadcasts a notification when one of them changes.

Why a pump and not a callback. A real-time task cannot call a WebSocket server: a publish allocates, uWebSockets is only safely callable from its own loop, and both can block. But the RT thread is the only code that sees a fault at the moment it happens. So the RT side stores plain scalars into storage that already exists and moves on, and something off the RT thread notices and formats. This is that something, written once instead of once per feature.

The change signal is a version counter, not a flag. A source's revision is a cheap read of a counter the producer only ever increases; this class keeps a private last-seen copy and never writes back. A counter has no clear step to lose an update against, and several bumps between two polls coalesce into one message carrying the current state — which is the wanted behaviour, not a compromise. Only inequality is used, so a producer that restarts its counter at zero is reported rather than ignored.

The producer should bump its counter with std::memory_order_release after writing the scalars that describe the change, and a render should read the counter with std::memory_order_acquire before those scalars. That pairing is what stops a message pairing a fresh counter with stale fields.

This class names no server type. The seam is a PublishFn wired at the composition root, the same shape as MonitoringManager::setPublish.

Every notification goes to one topic, kNotificationTopic, so a client subscribes once and receives every event — including ones added in a later version. A topic per source would look tidier and would break that: uWebSockets 20.77 matches topic names exactly, with no wildcard, so "give me all notifications" would not be expressible and a new source would silently reach no existing client. Clients tell events apart by data.event, which is in the payload either way. Per-topic fan-out is for the monitoring streams, where a client chooses what it pays for.

Not for per-cycle telemetry. Latency here is the poll cadence, which is right for a state change and wrong for a signal trace — that is a monitoring subscription on the recorder ring.

Member Typedef Documentation

PublishFn

using mm::node::NotificationBus::PublishFn = std::function<void(std::string topic, std::string json)>

Receives a finished message and the topic to put it on. Wired to WebSocketServer::publish at the root, the same shape as MonitoringManager::setPublish.

Constructor & Destructor Documentation

NotificationBus() [1/3]

mm::node::NotificationBus::NotificationBus ( )

default

~NotificationBus()

mm::node::NotificationBus::~NotificationBus

(

)

Requests the stop and joins.

NotificationBus() [2/3]

mm::node::NotificationBus::NotificationBus ( const NotificationBus &  )

delete

NotificationBus() [3/3]

mm::node::NotificationBus::NotificationBus ( NotificationBus &&  )

delete

Member Function Documentation

addSource()

void mm::node::NotificationBus::addSource

(

Source

source

)

Registers a source. Call before start.

Membership is fixed once the thread runs, the same rule CyclicTask follows, which is what lets the poll loop read the registry with no lock.

operator=() [1/2]

NotificationBus & mm::node::NotificationBus::operator= ( const NotificationBus &  )

delete

operator=() [2/2]

NotificationBus & mm::node::NotificationBus::operator= ( NotificationBus &&  )

delete

setPublish()

void mm::node::NotificationBus::setPublish

(

PublishFn

publish

)

Sets where finished messages go. Call before start.

Leaving it unset is supported: every source still renders, so one that logs keeps logging.

start()

void mm::node::NotificationBus::start

(

)

Starts the poll thread.

Each source's last-seen mark is taken here, so a producer that was already counting before this call raises no message for what it had already recorded.

stop()

void mm::node::NotificationBus::stop

(

)

Stops the poll thread and joins it. Returns immediately if it is not running, and does not wait out the poll interval.

---

The documentation for this class was generated from the following files:

• libs/node/notification_bus.h
• libs/node/notification_bus.cc