Mads::Bridge Class Reference

The Bridge class represents metadata for an agent. More...

#include <bridge.hpp>

Inheritance diagram for Mads::Bridge:

Public Member Functions
	Bridge (std::string name, std::string settings_path)
	Constructs a Bridge object with the given name and settings path.
void	route ()
	Routes any line received via STDIN to the mads network.
Public Member Functions inherited from Mads::Agent
	Agent (std::string name, std::string settings_uri)
	Constructs an Agent object with the given name and settings path. If the settings path is a URI, the agent will connect to the broker with a REQ/REP socket and request the same settings file loaded by the broker.
void	init (std::string name, std::string settings_uri, bool crypto=false, std::filesystem::path const &key_dir="", bool install_watchdog=true)
	Initializes the agent.
void	init (bool crypto=false, bool install_watchdog=true)
	Initializes the agent.
void	fetch_settings (bool crypto=false)
	Acquires settings (and any broker-served attachment) without binding the agent to a settings section.
virtual	~Agent ()
void	install_loop_watchdog (uint8_t max_count=3)
	Install a watch thread to ensure exit from loops.
void	save_settings (const std::string path=SETTINGS_PATH)
	Save settings read from broker to file.
nlohmann::json	get_settings ()
	Get all settings as JSON.
virtual void	info (std::ostream &out=std::cout)
	Prints information about the agent.
void	connect (std::chrono::milliseconds delay=std::chrono::milliseconds(250))
	Connects the agent to the publish and subscribe endpoints.
bool	wait_for_connection (std::chrono::milliseconds timeout)
	Blocks until the publisher socket's connection is confirmed by a real ZMQ_EVENT_CONNECTED/ZMQ_EVENT_HANDSHAKE_SUCCEEDED event, or timeout elapses (ZMQ_DEVELOPMENT.md §2.1).
Mads::LinkState	link_state () const
	The current state of this agent's link to the broker: up or down, why (last_handshake – e.g. a ZMQ_EVENT_HANDSHAKE_FAILED_AUTH turns a CURVE rejection into a fact instead of a bare receive timeout), since when, and how often it has dropped and recovered.
void	disconnect ()
	Disconnects the agent from the publish and subscribe endpoints.
void	shutdown ()
	Performs a coordinated shutdown of the agent.
void	set_cross (bool cross)
	Sets the cross flag.
void	enable_remote_control (bool threaded=false)
	Enables remote control for the agent.
void	enable_threaded_remote_control ()
void	register_event (const event_type event=event_type::marker, const nlohmann::json &info=nlohmann::json(), const std::string &info_name="info")
	Registers an event.
void	publish (nlohmann::json payload, std::string topic="")
	Publishes a message with the given JSON payload.
void	publish (const char *payload, size_t len, nlohmann::json meta=nlohmann::json{{"format", "raw"}}, std::string topic="")
	Publishes a message with the given binary blob payload.
void	publish (const std::vector< unsigned char > &payload, nlohmann::json meta=nlohmann::json{{"format", "raw"}}, std::string topic="")
	Publishes a message with the given binary blob payload.
message_type	receive (bool dont_block=false)
	Receives a message from the subscribe socket.
bool	receive_raw_message (std::string &topic, std::vector< std::string > &parts, bool dont_block=false)
	Receives a message without any JSON/blob (de)serialization (P3).
void	publish_raw_message (const std::string &topic, const std::vector< std::string > &parts)
	Publishes a raw multi-part message with no JSON encoding, no automatic field-stamping (agent_id/hostname/timestamp/...), no compression, and no frame header (P3).
void	loop (loop_fun_t const &lambda, std::chrono::nanoseconds duration)
void	loop (loop_fun_t const &lambda)
	Enters the main loop of the agent. It also sets a signal handler for SIGNINT, which will set the running flag to false. The loop duration is set to the time step of the agent, as pased in the ini file with the time_step parameter (or the finer time_step_us). It it is zero or not set, it will run at max speed.
void	enable_high_res_loop (bool on=true, std::chrono::nanoseconds spin_margin=std::chrono::microseconds(200))
	Opt into (or out of) high-resolution loop pacing.
bool	high_res_loop () const
	Returns whether high-resolution loop pacing is enabled.
void	remote_control (std::string payload_str)
	Handles remote control commands.
void	set_agent_id (std::string id)
	Set the agent ID field.
std::string	get_agent_id ()
	Get the agent ID.
void	set_sub_endpoint (std::string endpoint)
	Set the subscribe endpoint URL.
std::string	sub_endpoint () const
	Get the subscribe endpoint URL.
void	set_pub_endpoint (std::string endpoint)
	Set the publish endpoint URL.
std::string	pub_endpoint () const
	Get the publish endpoint URL.
void	set_pub_topic (std::string topic)
	Sets the publish topic.
std::string	pub_topic () const
	Gets the publish topic.
void	set_sub_topic (std::vector< std::string > topics)
	Sets the subscribe topics.
std::vector< std::string >	sub_topic () const
	Gets the subscribe topics.
std::map< std::string, std::string >	status ()
	Returns the status of the system.
std::string	name ()
	Returns the name of the agent.
std::tuple< std::string, std::string >	last_message ()
	Returns the last received message by the agent.
std::tuple< std::string, nlohmann::json >	last_json ()
	Returns the last received message as a parsed JSON object.
std::string	last_topic ()
	Returns the topic of the last received message by the agent.
std::tuple< std::string, std::string, std::vector< unsigned char > >	last_blob ()
	Returns the last received blob by the agent.
std::tuple< std::string_view, std::string_view, std::span< const unsigned char > >	last_blob_view () const
	Zero-copy view of the last received blob.
size_t	dropped_messages () const
	Number of messages dropped because they were malformed or could not be decoded (bad part count, failed decompression, etc.).
bool	settings_are_local () const
	Detects if settings are local or loaded from URI.
bool	is_connected ()
	Detects if agent is connected.
int	settings_timeout ()
	Returns the value of timeout in loading settings from URI.
void	set_settings_timeout (int to)
	Sets the value of timeout in loading settings from URI. Set to for no timeout.
void	set_settings_timeout (std::chrono::milliseconds to)
	Sets the value of timeout in loading settings from URI. Set to for no timeout.
int	receive_timeout ()
	Returns the value of timeout in receiving messages.
void	set_receive_timeout (int to)
	Sets the value of timeout in receiving messages. Set to 0 for no timeout.
void	set_receive_timeout (std::chrono::milliseconds to)
	Sets the value of timeout in receiving messages. Set to 0 for no timeout.
bool	restart ()
	Returns wheter a restart has been requested.
std::shared_ptr< Mads::Runtime >	runtime () const
	The Runtime that owns this agent's run state.
bool	running () const
	True while this agent's loops should keep going.
void	set_runtime (std::shared_ptr< Mads::Runtime > runtime)
	Attach the agent to a different Runtime.
std::filesystem::path	attachment_path ()
	Returns the path to the attachment file.
bool	is_crypto ()
	Returns whether CURVE encryption is enabled.
void	setup_crypto (Mads::auth_verbose verbose=auth_verbose::off)
	Set the use of CURVE encryption and enables authentication.
std::unique_ptr< CurveAuth > *	curve_auth ()
	Returns a pointer to the CurveAuth object.
std::filesystem::path	key_dir ()
	Returns the path to the etc directory.
void	set_key_dir (const std::filesystem::path &path)
	Sets the path to the etc directory.
std::string	settings_uri ()
void	set_conflate (bool conflate)
bool	conflate ()
void	set_high_watermark (int i=1000)
	Set the high watermark (ZMQ receive queue bound).
int	high_watermark ()
void	set_delivery (Delivery d)
	Select subscriber delivery semantics.
Delivery	delivery () const
	Current delivery semantics.
void	set_wire_format (WireFormat fmt)
	Select the on-the-wire payload encoding used when publishing.
WireFormat	wire_format () const
	The wire format used for outgoing messages.
void	set_compression (Compression c)
	Select the payload compression policy for outgoing messages.
Compression	compression () const
	The compression policy used for outgoing messages.
Mads::ClockOffsetResult	measure_clock_offset (size_t samples=5, int timeout_ms=1000)
	Runs samples four-timestamp exchanges against the broker's settings endpoint and adopts the result as this agent's own clock-offset measurement (clock-offset source A – clock_offset.hpp §"Source A"). Never throws: an unreachable broker, or one too old to know the "clock" command, leaves any previous measurement in place and returns {valid=false}.
std::vector< Mads::ClockPeerObservation >	broadcast_clock_probe (std::chrono::milliseconds window=std::chrono::milliseconds(300))
	Broadcasts a clock-sync ping on CLOCKSYNC_TOPIC and collects every responder's pong that arrives within window (clock-offset source B – clock_offset.hpp §"Source B"). Used internally when [agents] clock_source is "peer", and by mads top --probe to build its fleet table.
Mads::ClockOffsetResult	clock_offset () const
	This agent's currently adopted clock offset: the clock domain's consensus winner (Mads::ClockConsensus, possibly another agent's measurement) if one is known and fresh, else this agent's own last measurement, else {valid=false}. Add offset_us to a wall-clock reading taken on this host to obtain broker-host time.
std::string	clock_domain () const
	This process's clock-domain identity (Mads::detail::clock_domain_id(): boot_id on Linux, hostname elsewhere). Agents that share this value share a clock and are expected to converge on one identical clock_offset().

Additional Inherited Members
Public Types inherited from Mads::Agent
using	loop_fun_t = std::function< std::chrono::nanoseconds()>
	Enters the main loop of the agent. It also sets a signal handler for SIGNINT, which will set the running flag to false.
Static Public Member Functions inherited from Mads::Agent
static void	install_signal_handlers ()
	Install SIGINT/SIGTERM handlers that request a clean shutdown.
Public Attributes inherited from Mads::Agent
double	timecode_fps = MADS_FPS
Mads::auth_verbose	auth_verbose = auth_verbose::off
std::string	server_key_name = "broker"
std::string	client_key_name = "client"
bool	dummy = false
Protected Member Functions inherited from Mads::Agent
bool	keep_running () const
	True while this agent's loops should keep going.
void	connect_pub (std::chrono::milliseconds delay=std::chrono::milliseconds(0))
	Connects the agent to the publish endpoint.
void	connect_sub ()
	Connects the agent to the subscribe endpoint and subscribes to the topics.
bool	receive_raw (zmq::multipart_t &message, bool dont_block=false)
	Internal use wrapping the receive step both for normal operations and for LastKnown Value (LKV) semantic, when high_watermark is 1.
bool	_topic_matches_subscription (const std::string &topic) const
	MQTT-style wildcard filter (P2): true if topic is accepted by at least one entry of _sub_topic – literal entries are matched exactly as the raw ZMQ SUBSCRIBE prefix already does today, wildcard entries (containing '+'/'#') via Mads::topic_match(). Only ever consulted when _wildcard_sub_topic is non-empty; see connect_sub()/receive_raw().
void	_apply_socket_options ()
	Resolves the plain libzmq transport-tuning knobs (ZMQ_DEVELOPMENT.md §1.4: TCP keepalive, SNDBUF/RCVBUF, ...) from settings and applies them to both sockets. Called once from init(), after settings are parsed and before connect(). A key left unset in both [agents] and the agent's own section makes no setsockopt() call at all, so an unedited mads.ini is unaffected.
void	_start_io_thread ()
	Brings up _io_thread if it is not already running. Called from whichever of connect_pub()/connect_sub() runs first, since a publish-only agent never reaches the latter and a subscribe-only agent never reaches the former.
void	_configure_clock_sync ()
	Parses the [agents]/[<name>] clock_* settings during init() and appends CLOCKSYNC_TOPIC to _sub_topic when this agent needs to participate (mirrors how enable_remote_control() appends "control").
bool	_clock_wants_sync () const
void	_start_clock_thread ()
	Brings up _clock_thread if it is not already running: announces this agent's adopted clock offset on CLOCKSYNC_TOPIC every clock_announce_ms, and – when clock_interval_ms > 0 and this agent is its clock domain's current winner (or no winner is known yet) – re-measures. A "peer"-source agent takes its very first measurement here too, since broadcast_clock_probe() needs a connected socket that does not exist yet during init(). Called from connect(); joined in disconnect()/shutdown() exactly like _io_thread.
void	_run_peer_measurement ()
void	_handle_clocksync_message (const nlohmann::json &msg)
void	_announce_clock_offset (const Mads::ClockOffsetResult &r)
Mads::ClockOffsetResult	_stamp_clock_result (Mads::ClockOffsetResult r, Mads::ClockSource source, uint8_t hops)
std::string	_clock_agent_identity () const
Static Protected Member Functions inherited from Mads::Agent
static std::tuple< std::string, std::string, std::string >	split_URL (const std::string &url)
Protected Attributes inherited from Mads::Agent
std::string	_hostname
std::string	_name
std::string	_settings_uri
std::string	_raw_settings
toml::table	_config
std::string	_pub_endpoint
std::string	_sub_endpoint
std::string	_pub_topic
std::string	_agent_id
std::vector< std::string >	_sub_topic
std::vector< std::string >	_wildcard_sub_topic
zmq::context_t	_context
zmq::socket_t	_publisher
zmq::socket_t	_subscriber
Mads::SocketMonitor	_pub_monitor
Mads::SocketMonitor	_sub_monitor
std::map< std::string, std::shared_ptr< LazyPayload > >	_status
std::tuple< std::string, std::shared_ptr< LazyPayload > >	_last_message
std::tuple< std::string, std::string, std::vector< unsigned char > >	_last_blob
std::mutex	_message_state_mutex
bool	_cross = false
bool	_connected = false
int	_receive_timeout = DEFAULT_RECEIVE_TIMEOUT_MS
int	_settings_timeout = 0
bool	_init_done = false
bool	_settings_fetched = false
bool	_restart = false
std::shared_ptr< Mads::Runtime >	_runtime = std::make_shared<Mads::Runtime>()
std::atomic< bool >	_stopping {false}
bool	_remote_controlled = false
std::chrono::nanoseconds	_time_step = std::chrono::nanoseconds(0)
bool	_high_res_loop = false
std::chrono::nanoseconds	_spin_margin = std::chrono::microseconds(200)
double	_timecode_offset = 0.0
std::filesystem::path	_attachment_path
bool	_crypto = false
bool	_conflate = false
std::unique_ptr< CurveAuth >	_curve_auth = nullptr
std::filesystem::path	_key_dir
bool	_last_value_only = false
bool	_shutdown_done = false
SharedLatest< zmq::multipart_t >	_latest_message
std::thread	_io_thread
std::atomic< bool >	_io_reads_subscriber {false}
std::thread	_watchdog_thread
std::thread	_startup_event_thread
std::mutex	_event_mtx
std::condition_variable	_event_cv
std::mutex	_publish_mutex
std::atomic< bool >	_watchdog_stop {false}
bool	_rc_owns_socket = false
WireFormat	_wire_format = WireFormat::Json
Compression	_compression = Compression::Auto
std::atomic< size_t >	_dropped_messages {0}
nlohmann::json	_settings_json
Mads::ClockSource	_clock_source = Mads::ClockSource::Broker
bool	_clock_sync_responder = true
int	_clock_interval_ms = 0
int	_clock_announce_ms = 5000
bool	_clock_correction = false
Mads::ClockConsensus	_clock_consensus {std::chrono::seconds(30)}
std::mutex	_clock_mtx
Mads::ClockOffsetResult	_own_clock_measurement
uint64_t	_clock_seq = 0
std::thread	_clock_thread
std::mutex	_clocksync_mtx
std::chrono::steady_clock::time_point	_clocksync_probe_sent_at {}
std::vector< Mads::ClockPeerObservation >	_clocksync_pongs
uint64_t	_clocksync_probe_seq = 0
std::map< std::string, std::chrono::steady_clock::time_point >	_clocksync_last_reply

Detailed Description

The Bridge class represents metadata for an agent.

This class inherits from the Agent class and provides additional functionality for loading settings and managing metadata.

Definition at line 34 of file bridge.hpp.

Constructor & Destructor Documentation

Bridge()

Mads::Bridge::Bridge ( std::string  name, std::string  settings_path  )

inline

Constructs a Bridge object with the given name and settings path.

Parameters

name	The name of the Bridge.
settings_path	The path to the settings file.

Definition at line 42 of file bridge.hpp.

Member Function Documentation

route()

void Mads::Bridge::route ( )

inline

Routes any line received via STDIN to the mads network.

Note

The input data must be a valid JSON string, one document per line, i.e. no newline within a single JSON document. Optionally, the input string may start with a topic name, followed by a colon and zero or more spaces, e.g.: "my_topic: {...}". In this case, this topic name will be used to publish the message, otherwise the default topic will be used. This allowas a single CLI client to publish to multiple topics.

This function is blocking.

Definition at line 59 of file bridge.hpp.

References Mads::Agent::_pub_topic, and Mads::Agent::publish().

---

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

• src/bridge.hpp