#include <agent.hpp>

Inheritance diagram for Mads::Agent:

Public Types
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.

Public Member Functions
	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.

virtual void	load_settings ()
	Additional settings to be loaded. Virtual function to be implemented by the derived class.

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().

Static Public Member Functions
static void	install_signal_handlers ()
	Install SIGINT/SIGTERM handlers that request a clean shutdown.

Public Attributes
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
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
static std::tuple< std::string, std::string, std::string >	split_URL (const std::string &url)

Protected Attributes
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

Definition at line 190 of file agent.hpp.

Member Typedef Documentation

◆ loop_fun_t

using Mads::Agent::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.

The loop period is expressed in nanoseconds internally, so that any std::chrono::duration (milliseconds, microseconds, ...) can be passed or returned by the lambda: e.g. agent.loop(cb, 500us); opts into microsecond-resolution pacing, while existing millisecond-based code (agent.loop(cb, 100ms);, return 0ms;) keeps compiling and behaving as before (the conversion to nanoseconds is implicit and lossless).

By default, the wait between iterations uses this_thread::sleep_for, whose actual wake-up jitter is limited by the OS scheduler (typically within a fraction of a millisecond). For tighter, microsecond-accurate pacing, opt into enable_high_res_loop(), which busy-spins the tail of each interval instead of sleeping through it, trading CPU time for precision: the fraction of each interval spent spinning (and so the extra CPU load) scales with spin_margin relative to the loop period, so it is most noticeable on short periods and/or a large spin_margin.

Parameters

lambda	the function to be executed in the main loop; it can return the requested duration of the next loop: if this is 0, the default loop time is used, otherwise the value returned by the lambda
duration	the duration of the loop (default 0, max speed)

Exceptions

AgentError if not initialized

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 626 of file agent.hpp.

Constructor & Destructor Documentation

◆ Agent()

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.

Parameters

name	The name of the agent. If it is a path, it take the filename.
settings_path	The path or URI to the settings file for the agent.

◆ ~Agent()

virtual Mads::Agent::~Agent ( )

virtual

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Member Function Documentation

◆ _announce_clock_offset()

void Mads::Agent::_announce_clock_offset ( const Mads::ClockOffsetResult & r )

protected

Publishes r as a CLOCKSYNC_TOPIC announcement and records it into this agent's own local consensus view.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ _apply_socket_options()

void Mads::Agent::_apply_socket_options ( )

protected

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.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ _clock_agent_identity()

std::string Mads::Agent::_clock_agent_identity ( ) const

protected

Stable per-agent identity for clock-sync bookkeeping (ping/pong/ announce "who"), independent of the optional, user-facing –agent-id/_agent_id used for payload stamping: falls back to "<name>@<pid>" so consensus works even when –agent-id was never set.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ _clock_wants_sync()

bool Mads::Agent::_clock_wants_sync ( ) const

protected

True when this agent needs to hear/answer CLOCKSYNC_TOPIC (source != none, or the responder is enabled – true by default). Deliberately not reflected in _sub_topic/sub_topic() – see _configure_clock_sync().

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ _configure_clock_sync()

void Mads::Agent::_configure_clock_sync ( )

protected

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").

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ _handle_clocksync_message()

void Mads::Agent::_handle_clocksync_message ( const nlohmann::json & msg )

protected

Handles one decoded CLOCKSYNC_TOPIC message (ping/pong/announce). Called from receive()'s two data-frame branches and from the _io_thread control-analogous branch; never throws, never surfaces the message to the caller.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ _run_peer_measurement()

void Mads::Agent::_run_peer_measurement ( )

protected

One clock_source == "peer" measurement attempt: probes, picks an anchor per the chain-safety rules (bounded hops, non-stale, a real source), composes clock_offset(mine) = anchor.offset_us + theta, and records/announces the result. A no-op if no eligible peer responds.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ _stamp_clock_result()

Mads::ClockOffsetResult Mads::Agent::_stamp_clock_result	(	Mads::ClockOffsetResult	r,
		Mads::ClockSource	source,
		uint8_t	hops
	)

protected

Fills in the Agent-level bookkeeping estimate()/ClockOffsetEstimator do not know about: source, hops, this agent's identity, and the next measurement sequence number.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ _start_clock_thread()

void Mads::Agent::_start_clock_thread ( )

protected

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.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ _start_io_thread()

void Mads::Agent::_start_io_thread ( )

protected

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.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ _topic_matches_subscription()

bool Mads::Agent::_topic_matches_subscription ( const std::string & topic ) const

protected

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().

Parameters

topic The concrete topic of an already-received message.

Returns: true if the message should be delivered to receive()/callbacks.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ attachment_path()

std::filesystem::path Mads::Agent::attachment_path ( )

Returns the path to the attachment file.

This is the file that was sent by the broker when reading settings from URI. It is used to load additional plugins or resources.

Returns: The path to the attachment file.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ broadcast_clock_probe()

std::vector< Mads::ClockPeerObservation > Mads::Agent::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.

Blocking: drives this agent's own receive() calls for the duration of window, so nothing else on the calling thread progresses meanwhile, and it is not safe to call concurrently with another in-flight broadcast_clock_probe() on the same Agent (single-flight; the background clock thread already respects this by construction). Returns an empty vector immediately if this agent has no publisher connected (no pub_topic) – it cannot ping without one.

Parameters

window how long to wait for responders.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ clock_domain()

std::string Mads::Agent::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().

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ clock_offset()

Mads::ClockOffsetResult Mads::Agent::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.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ compression()

Compression Mads::Agent::compression ( ) const

The compression policy used for outgoing messages.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ conflate()

bool Mads::Agent::conflate ( )

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ connect()

void Mads::Agent::connect ( std::chrono::milliseconds delay = std::chrono::milliseconds(250) )

Connects the agent to the publish and subscribe endpoints.

Optionally, this function accepts a delay parameter, which bounds how long connecting may take before publishing is safe. This is needed to ensure that the agent is connected before sending messages. The default delay is 250 milliseconds.

delay is an upper bound, not a fixed cost: the publisher waits for its ZMTP handshake and then for SUBSCRIPTION_SETTLE_DELAY of slow-joiner grace, and returns as soon as both are done. Pass 0 only when the agent does not publish immediately – a first message sent with no grace at all is dropped by the PUB socket, unsent and unreported.

Parameters

type	The type of connection to be established.
delay	Upper bound, in milliseconds, on the connect-and-settle wait.

Exceptions

AgentError if not initialized

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Referenced by Mads::Dealer::connect(), and Mads::Worker::connect().

◆ connect_pub()

void Mads::Agent::connect_pub ( std::chrono::milliseconds delay = std::chrono::milliseconds(0) )

protected

Connects the agent to the publish endpoint.

Parameters

delay The delay in milliseconds after connecting.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ connect_sub()

void Mads::Agent::connect_sub ( )

protected

Connects the agent to the subscribe endpoint and subscribes to the topics.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ curve_auth()

std::unique_ptr< CurveAuth > * Mads::Agent::curve_auth ( )

Returns a pointer to the CurveAuth object.

Returns: unique_ptr<CurveAuth>*

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ delivery()

Delivery Mads::Agent::delivery ( ) const

Current delivery semantics.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ disconnect()

void Mads::Agent::disconnect ( )

Disconnects the agent from the publish and subscribe endpoints.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ dropped_messages()

size_t Mads::Agent::dropped_messages ( ) const

Number of messages dropped because they were malformed or could not be decoded (bad part count, failed decompression, etc.).

Returns: The cumulative count since startup.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ enable_high_res_loop()

void Mads::Agent::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.

When enabled, the wait between loop iterations sleeps for most of the remaining interval and then busy-spins on a steady_clock check for the last spin_margin, trading CPU usage for microsecond-accurate wake-up timing: the extra CPU cost is roughly proportional to spin_margin divided by the loop period (e.g. a 200us margin on a 500us period keeps a core busy on the order of ~20-40% of the time; a 200us margin on a 100ms period costs a negligible fraction). Leave disabled (the default) for millisecond-granular loops, where plain sleep_for is sufficient and cheaper.

Parameters

on	whether to enable high-resolution pacing (default true).
spin_margin	the portion of each interval to busy-spin instead of sleep (default 200us); only relevant when `on` is true.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ enable_remote_control()

void Mads::Agent::enable_remote_control ( bool threaded = false )

Enables remote control for the agent.

This function subscribes the agent to the "control" topic. If the agent is not a subscriber starts a thread to handle remote control commands.

If the agent is also a subscriber, then the remote control is handled in the main loop by calling the remote_control() function.

Parameters

threaded If true, the agent is NOT supposed to receive data (it is a pure sink) so messages are read on a searate thread

Exceptions

AgentError	if not initialized
AgentError	if already connected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Referenced by enable_threaded_remote_control().

◆ enable_threaded_remote_control()

void Mads::Agent::enable_threaded_remote_control ( )

inline

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 477 of file agent.hpp.

References enable_remote_control().

◆ fetch_settings()

void Mads::Agent::fetch_settings ( bool crypto = false )

Acquires settings (and any broker-served attachment) without binding the agent to a settings section.

This performs the settings acquisition portion of init() only: CURVE setup, resolving the local-file/broker-URI settings source, and (if settings come from a broker) saving any served attachment and renaming it per the section's attachment_ext. It does not validate or bind the _name section, install the loop watchdog, or call load_settings().

Idempotent: a second call (including the one made internally by init(), if not already fetched) is a no-op. Useful for callers that need to inspect settings or a served attachment (e.g. to select a plugin file) before committing to init()'s section binding.

Parameters

crypto Whether to use CURVE encryption (default false).

Exceptions

AgentError if timed out in reading settings from broker.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ get_agent_id()

std::string Mads::Agent::get_agent_id ( )

Get the agent ID.

Returns: The agent ID

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ get_settings()

nlohmann::json Mads::Agent::get_settings ( )

Get all settings as JSON.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ high_res_loop()

bool Mads::Agent::high_res_loop ( ) const

Returns whether high-resolution loop pacing is enabled.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ high_watermark()

int Mads::Agent::high_watermark ( )

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ info()

virtual void Mads::Agent::info ( std::ostream & out = std::cout )

virtual

Prints information about the agent.

This function prints the agent's name, settings file path, publish endpoint, publish topic, subscribe endpoint, and subscribe topics.

Parameters

out	The output stream to be used (default is cout).

Exceptions

AgentError if not initialized

Reimplemented in Mads::Dealer, Mads::Dummy, Mads::Image, Mads::Logger, Mads::Metadata, and Mads::Worker.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Referenced by Mads::Dealer::info(), Mads::Dummy::info(), Mads::Image::info(), Mads::Metadata::info(), and Mads::Worker::info().

◆ init() [1/2]

void Mads::Agent::init	(	bool	crypto = `false`,
		bool	install_watchdog = `true`
	)

Initializes the agent.

This function loads the settings file and sets the member variables of the agent.

Parameters

crypto Whether to use CURVE encryption (default false).

Exceptions

AgentError if timed out in reading settings from broker.

◆ init() [2/2]

void Mads::Agent::init	(	std::string	name,
		std::string	settings_uri,
		bool	crypto = `false`,
		std::filesystem::path const &	key_dir = `""`,
		bool	install_watchdog = `true`
	)

Initializes the agent.

This function loads the settings file and sets the member variables of the agent.

If the settings_uri is "none", the agent is initialized with default settings and does not attempt to load any settings file. This is useful for testing purposes.

Parameters

name	The name of the agent (as it is).
settings_uri	The path or URI to the settings file for the agent.
crypto	Whether to use CURVE encryption (default false).
key_dir	The directory where the CURVE keys are stored.

Exceptions

AgentError if timed out in reading settings from broker.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ install_loop_watchdog()

void Mads::Agent::install_loop_watchdog ( uint8_t max_count = 3 )

Install a watch thread to ensure exit from loops.

This starts a low-frequency thread that forces an exit when the agent keeps looping for more than 3 seconds after it was asked to stop (its Runtime was stopped or shutdown/disconnect was requested)

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ install_signal_handlers()

static void Mads::Agent::install_signal_handlers ( )

static

Install SIGINT/SIGTERM handlers that request a clean shutdown.

Idempotent and process-global: only the first call installs handlers, so calling loop() repeatedly (or running several agents in one process) does not clobber handler state.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ is_connected()

bool Mads::Agent::is_connected ( )

Detects if agent is connected.

Returns: true or false.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ is_crypto()

bool Mads::Agent::is_crypto ( )

Returns whether CURVE encryption is enabled.

Returns: true if CURVE encryption is enabled, false otherwise.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ keep_running()

bool Mads::Agent::keep_running ( ) const

inlineprotected

True while this agent's loops should keep going.

Combines the Runtime state (group + process level) with the per-agent stop request raised by shutdown()/disconnect().

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1170 of file agent.hpp.

References _runtime, and _stopping.

Referenced by running().

◆ key_dir()

std::filesystem::path Mads::Agent::key_dir ( )

Returns the path to the etc directory.

Returns: filesystem::path

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ last_blob()

std::tuple< std::string, std::string, std::vector< unsigned char > > Mads::Agent::last_blob ( )

Returns the last received blob by the agent.

Returns: A tuple containing the topic, format, and payload of the last received blob.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ last_blob_view()

std::tuple< std::string_view, std::string_view, std::span< const unsigned char > > Mads::Agent::last_blob_view ( ) const

Zero-copy view of the last received blob.

Returns non-owning views (topic, format, bytes) into the agent's internally stored blob, avoiding the full copy performed by last_blob(). The returned views remain valid only until the next call to receive() and must not be used concurrently with a threaded receive/drain thread.

Returns: A tuple of {topic, format, bytes} as views.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ last_json()

std::tuple< std::string, nlohmann::json > Mads::Agent::last_json ( )

Returns the last received message as a parsed JSON object.

Fast path for consumers that want a nlohmann::json rather than its text: for MsgPack frames this returns the already-decoded object (no re-dump and no re-parse); for JSON frames it parses once and caches. Avoids the round-trip that last_message() + json::parse() would otherwise incur.

Returns: A tuple of {topic, parsed payload}. The payload is a copy.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ last_message()

std::tuple< std::string, std::string > Mads::Agent::last_message ( )

Returns the last received message by the agent.

Returns: A tuple containing the topic and payload of the last received message.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ last_topic()

std::string Mads::Agent::last_topic ( )

Returns the topic of the last received message by the agent.

Returns: The topic of the last received message.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ link_state()

Mads::LinkState Mads::Agent::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.

Reported from the subscriber socket, since that is the one over which an agent would otherwise silently stop hearing from a departed broker; publish-only agents (no sub_topic, so no subscriber connection) fall back to the publisher's. Both connect to the same broker, so in practice they rise and fall together.

A set_cross() agent binds rather than connects, and gets one ZMQ_EVENT_DISCONNECTED per departing peer with no matching handshake event on arrival – events that would add up to "down" as soon as any one peer left. There is no broker link to describe in that topology, so this reports Mads::LinkStatus::Unknown throughout.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ load_settings()

virtual void Mads::Agent::load_settings ( )

virtual

Additional settings to be loaded. Virtual function to be implemented by the derived class.

This function is automatically called by init() to load additional settings.

Reimplemented in Mads::Dummy.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ loop() [1/2]

void Mads::Agent::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.

Parameters

lambda the function to be executed in the main loop; it can return the requested duration of the next loop: if this is 0, the default loop time is used, otherwise the value returned by the lambda

Exceptions

AgentError if not initialized

◆ loop() [2/2]

void Mads::Agent::loop	(	loop_fun_t const &	lambda,
		std::chrono::nanoseconds	duration
	)

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ measure_clock_offset()

Mads::ClockOffsetResult Mads::Agent::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}.

Called automatically at the end of init() when [agents] clock_source is "broker" (the default) and settings come from a broker; called again periodically by the background clock thread when clock_interval_ms > 0 and this agent is (or would become) its clock domain's adopted source. Safe to call directly at any other time too.

Parameters

samples	number of round-trips; the min-delay one is kept.
timeout_ms	per-request send/receive timeout.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ name()

std::string Mads::Agent::name ( )

Returns the name of the agent.

Returns: The name of the agent.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ pub_endpoint()

std::string Mads::Agent::pub_endpoint ( ) const

inline

Get the publish endpoint URL.

Returns: The publish endpoint URL

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 734 of file agent.hpp.

References _pub_endpoint.

◆ pub_topic()

std::string Mads::Agent::pub_topic ( ) const

inline

Gets the publish topic.

Returns: The publish topic.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 748 of file agent.hpp.

References _pub_topic.

◆ publish() [1/3]

void Mads::Agent::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.

Parameters

payload	The binary blob payload of the message.
format	The format of the blob (a string, default "raw").
topic	The topic of the message.

Exceptions

AgentError if not initialized

◆ publish() [2/3]

void Mads::Agent::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.

Parameters

payload	The binary blob payload of the message.
metadata	The metadata for the blob.
topic	The topic of the message.

Exceptions

AgentError if not initialized

◆ publish() [3/3]

void Mads::Agent::publish	(	nlohmann::json	payload,
		std::string	topic = `""`
	)

Publishes a message with the given JSON payload.

Parameters

payload The JSON payload of the message.

Exceptions

AgentError if not initialized

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Referenced by Mads::Dummy::publish(), Mads::Image::publish_change(), and Mads::Bridge::route().

◆ publish_raw_message()

void Mads::Agent::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).

Sends topic followed by parts, byte for byte, exactly as given. Intended for mads-play, to republish a recorded frame unchanged – pairs with receive_raw_message() and the parts BagReader hands back, so a JSON message or a blob's meta+bytes parts are never re-parsed, re-dumped, or re-copied through the JSON path.

Parameters

topic	The topic frame.
parts	The frames to send after the topic, in order.

Exceptions

AgentError if not initialized.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ receive()

message_type Mads::Agent::receive ( bool dont_block = false )

Receives a message from the subscribe socket.

This function receives a message from the subscribe socket and updates the agent's status and last received message.

Exceptions

AgentError	if the received message has only one part or more than two parts.
AgentError	if not initialized

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ receive_raw()

bool Mads::Agent::receive_raw	(	zmq::multipart_t &	message,
		bool	dont_block = `false`
	)

protected

Internal use wrapping the receive step both for normal operations and for LastKnown Value (LKV) semantic, when high_watermark is 1.

Parameters

message
dont_block

Returns: true new message; false no new message (when non blocking or timeout)

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ receive_raw_message()

bool Mads::Agent::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).

Exposes the wire frame exactly as received: the topic (part 0) and every subsequent part verbatim as bytes (embedded NUL bytes included), with no frame-header parsing, no decompression, and no JSON decode – whatever was published (JSON text, a MsgPack frame, or a blob's meta+bytes parts) comes back unchanged. Intended for mads-record, so a bag file gets exactly what was on the wire without round-tripping through nlohmann::json or re-copying blob bytes.

Goes through the same receive_raw() path as receive() (wildcard sub_topic filtering (P2) and Last-Known-Value delivery both still apply); this is purely a difference in how the received frame is exposed to the caller, not in what gets delivered.

Parameters

topic	Out param: the topic frame.
parts	Out param: every frame after the topic, in wire order. Cleared and repopulated on a successful receive; left untouched if nothing was received.
dont_block	Whether to return immediately if no message is pending.

Returns: true if a message was received, false if nothing arrived (dont_block, or the receive timeout elapsed).

Exceptions

AgentError if not initialized, or while threaded remote control owns the subscriber socket.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ receive_timeout()

int Mads::Agent::receive_timeout ( )

Returns the value of timeout in receiving messages.

Returns: the timeout in ms (default DEFAULT_RECEIVE_TIMEOUT_MS = 500).

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Referenced by Mads::Worker::pull().

◆ register_event()

void Mads::Agent::register_event	(	const event_type	event = `event_type::marker`,
		const nlohmann::json &	info = `nlohmann::json()`,
		const std::string &	info_name = `"info"`
	)

Registers an event.

This function registers an event with the broker. The event is sent after 500 milliseconds.

Parameters

event The event to be registered.

Exceptions

AgentError if not initialized

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ remote_control()

void Mads::Agent::remote_control ( std::string payload_str )

Handles remote control commands.

This function is called by the main loop to handle remote control commands. It checks if the last received message is a control message and acts accordingly.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ restart()

bool Mads::Agent::restart ( )

Returns wheter a restart has been requested.

Returns: the restart flag.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ running()

bool Mads::Agent::running ( ) const

inline

True while this agent's loops should keep going.

Combines the Runtime state (group and process level) with the per-agent stop request raised by shutdown()/disconnect(). This is the condition loop() checks between iterations, exposed for application main loops.

Returns: Whether the agent should keep running.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 942 of file agent.hpp.

References keep_running().

◆ runtime()

std::shared_ptr< Mads::Runtime > Mads::Agent::runtime ( ) const

inline

The Runtime that owns this agent's run state.

loop(), the delivery drain thread, and the threaded remote control all keep going while runtime()->running() is true and the agent has not been individually stopped (shutdown()/disconnect()). Stopping the Runtime stops every agent that shares it.

Returns: The agent's Runtime (never null).

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 931 of file agent.hpp.

References _runtime.

◆ save_settings()

void Mads::Agent::save_settings ( const std::string path = SETTINGS_PATH )

Save settings read from broker to file.

Parameters

path

Exceptions

AgentError if not initialized or settings are local.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ set_agent_id()

void Mads::Agent::set_agent_id ( std::string id )

Set the agent ID field.

Parameters

id	The agent ID

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ set_compression()

void Mads::Agent::set_compression ( Compression c )

Select the payload compression policy for outgoing messages.

Default is Compression::Auto (compress only frames >= COMPRESSION_AUTO_THRESHOLD bytes). Compression::Snappy reproduces the historical always-compress behaviour; Compression::None disables it.

Parameters

c	The compression policy to use.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ set_conflate()

void Mads::Agent::set_conflate ( bool conflate )

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ set_cross()

void Mads::Agent::set_cross ( bool cross )

Sets the cross flag.

If the cross flag is set, the agent will bind to the publish endpoint and connect to the subscribe endpoint. This is useful for testing purposes.

Parameters

cross The value of the cross flag.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ set_delivery()

void Mads::Agent::set_delivery ( Delivery d )

Select subscriber delivery semantics.

Parameters

d	Delivery::Queued keeps the receive queue; Delivery::LastKnownValue keeps only the latest message, draining the rest on a background thread.

Exceptions

AgentError if already connected.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ set_high_watermark()

void Mads::Agent::set_high_watermark ( int i = 1000 )

Set the high watermark (ZMQ receive queue bound).

Parameters

i	The queue size. A value of 1 selects Last-Known-Value (LKV) delivery for backward compatibility (equivalent to set_delivery(Delivery::LastKnownValue)); prefer set_delivery() for clarity. A value of 0 means "unlimited" (ZMQ semantics) and Queued delivery.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ set_key_dir()

void Mads::Agent::set_key_dir ( const std::filesystem::path & path )

Sets the path to the etc directory.

Parameters

path

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ set_pub_endpoint()

void Mads::Agent::set_pub_endpoint ( std::string endpoint )

inline

Set the publish endpoint URL.

Parameters

endpoint in the form tcp://hostname:port

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 727 of file agent.hpp.

References _pub_endpoint.

◆ set_pub_topic()

void Mads::Agent::set_pub_topic ( std::string topic )

Sets the publish topic.

Parameters

topic The topic to be used for publishing messages.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ set_receive_timeout() [1/2]

void Mads::Agent::set_receive_timeout ( int to )

Sets the value of timeout in receiving messages. Set to 0 for no timeout.

Parameters

to	the timeout in ms.

Exceptions

AgentError if already initialized

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ set_receive_timeout() [2/2]

void Mads::Agent::set_receive_timeout ( std::chrono::milliseconds to )

Sets the value of timeout in receiving messages. Set to 0 for no timeout.

Parameters

to	the timeout.

Exceptions

AgentError if already initialized

◆ set_runtime()

void Mads::Agent::set_runtime ( std::shared_ptr< Mads::Runtime > runtime )

Attach the agent to a different Runtime.

Each agent owns its own Runtime by default; attach several agents to a shared Runtime (e.g. another agent's runtime()) to stop them together without affecting the rest of the process.

Parameters

runtime The Runtime to attach to.

Exceptions

AgentError if runtime is null or the agent is connected.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ set_settings_timeout() [1/2]

void Mads::Agent::set_settings_timeout ( int to )

Sets the value of timeout in loading settings from URI. Set to for no timeout.

Parameters

to	the timeout in ms.

Exceptions

AgentError if already initialized

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ set_settings_timeout() [2/2]

void Mads::Agent::set_settings_timeout ( std::chrono::milliseconds to )

Sets the value of timeout in loading settings from URI. Set to for no timeout.

Parameters

to	the timeout in milliseconds.

Exceptions

AgentError if already initialized

◆ set_sub_endpoint()

void Mads::Agent::set_sub_endpoint ( std::string endpoint )

inline

Set the subscribe endpoint URL.

Parameters

endpoint in the form tcp://hostname:port

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 713 of file agent.hpp.

References _sub_endpoint.

◆ set_sub_topic()

void Mads::Agent::set_sub_topic ( std::vector< std::string > topics )

inline

Sets the subscribe topics.

Parameters

topics The topics to be used for subscribing to messages.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 755 of file agent.hpp.

References _sub_topic.

◆ set_wire_format()

void Mads::Agent::set_wire_format ( WireFormat fmt )

Select the on-the-wire payload encoding used when publishing.

Default is WireFormat::Json (legacy, header-less, fully backward compatible). WireFormat::MsgPack emits a self-describing frame header and MessagePack-encoded payloads. Receivers accept both transparently.

Parameters

fmt	The wire format to use for outgoing messages.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ settings_are_local()

bool Mads::Agent::settings_are_local ( ) const

Detects if settings are local or loaded from URI.

Returns: true or false.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ settings_timeout()

int Mads::Agent::settings_timeout ( )

Returns the value of timeout in loading settings from URI.

Returns: the timeout in ms (default to 0, no timeout).

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ settings_uri()

std::string Mads::Agent::settings_uri ( )

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ setup_crypto()

void Mads::Agent::setup_crypto ( Mads::auth_verbose verbose = auth_verbose::off )

Set the use of CURVE encryption and enables authentication.

Parameters

verbose

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ shutdown()

void Mads::Agent::shutdown ( )

Performs a coordinated shutdown of the agent.

This ensures all background threads (drain, remote control) are properly joined before closing sockets and terminating the ZMQ context. Called automatically by the destructor.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ split_URL()

static std::tuple< std::string, std::string, std::string > Mads::Agent::split_URL ( const std::string & url )

staticprotected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ status()

std::map< std::string, std::string > Mads::Agent::status ( )

Returns the status of the system.

Returns: A map containing the last messages for each subscribed topic.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ sub_endpoint()

std::string Mads::Agent::sub_endpoint ( ) const

inline

Get the subscribe endpoint URL.

Returns: The subscribe endpoint URL

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 720 of file agent.hpp.

References _sub_endpoint.

◆ sub_topic()

std::vector< std::string > Mads::Agent::sub_topic ( ) const

inline

Gets the subscribe topics.

Returns: The subscribe topics.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 762 of file agent.hpp.

References _sub_topic.

◆ wait_for_connection()

bool Mads::Agent::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).

Warning: This is a transport-liveness signal, not a licence to publish. ZMQ_EVENT_CONNECTED fires at the TCP level, before the broker's XSUB frontend has forwarded the fleet's subscriptions back to this publisher, and a PUB socket silently discards anything sent while no subscription matches it. connect() therefore does not use this: it waits on the ZMTP handshake and then adds SUBSCRIPTION_SETTLE_DELAY of slow-joiner grace. Callers waiting by hand before a one-shot publish need that same grace.

Parameters

timeout upper bound on how long to wait.

Returns: true if the connection was observed within the timeout.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

◆ wire_format()

WireFormat Mads::Agent::wire_format ( ) const

The wire format used for outgoing messages.

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Member Data Documentation

◆ _agent_id

std::string Mads::Agent::_agent_id

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1292 of file agent.hpp.

◆ _attachment_path

std::filesystem::path Mads::Agent::_attachment_path

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1329 of file agent.hpp.

◆ _clock_announce_ms

int Mads::Agent::_clock_announce_ms = 5000

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1370 of file agent.hpp.

◆ _clock_consensus

Mads::ClockConsensus Mads::Agent::_clock_consensus {std::chrono::seconds(30)}

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1376 of file agent.hpp.

◆ _clock_correction

bool Mads::Agent::_clock_correction = false

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1371 of file agent.hpp.

◆ _clock_interval_ms

int Mads::Agent::_clock_interval_ms = 0

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1369 of file agent.hpp.

◆ _clock_mtx

std::mutex Mads::Agent::_clock_mtx

mutableprotected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1381 of file agent.hpp.

◆ _clock_seq

uint64_t Mads::Agent::_clock_seq = 0

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1383 of file agent.hpp.

◆ _clock_source

Mads::ClockSource Mads::Agent::_clock_source = Mads::ClockSource::Broker

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1367 of file agent.hpp.

◆ _clock_sync_responder

bool Mads::Agent::_clock_sync_responder = true

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1368 of file agent.hpp.

◆ _clock_thread

std::thread Mads::Agent::_clock_thread

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1384 of file agent.hpp.

◆ _clocksync_last_reply

std::map<std::string, std::chrono::steady_clock::time_point> Mads::Agent::_clocksync_last_reply

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1395 of file agent.hpp.

◆ _clocksync_mtx

std::mutex Mads::Agent::_clocksync_mtx

mutableprotected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1388 of file agent.hpp.

◆ _clocksync_pongs

std::vector<Mads::ClockPeerObservation> Mads::Agent::_clocksync_pongs

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1390 of file agent.hpp.

◆ _clocksync_probe_sent_at

std::chrono::steady_clock::time_point Mads::Agent::_clocksync_probe_sent_at {}

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1389 of file agent.hpp.

◆ _clocksync_probe_seq

uint64_t Mads::Agent::_clocksync_probe_seq = 0

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1391 of file agent.hpp.

◆ _compression

Compression Mads::Agent::_compression = Compression::Auto

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1362 of file agent.hpp.

◆ _config

toml::table Mads::Agent::_config

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1289 of file agent.hpp.

Referenced by Mads::Dummy::load_settings().

◆ _conflate

bool Mads::Agent::_conflate = false

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1331 of file agent.hpp.

◆ _connected

bool Mads::Agent::_connected = false

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1313 of file agent.hpp.

◆ _context

zmq::context_t Mads::Agent::_context

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1299 of file agent.hpp.

◆ _cross

bool Mads::Agent::_cross = false

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1312 of file agent.hpp.

◆ _crypto

bool Mads::Agent::_crypto = false

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1330 of file agent.hpp.

◆ _curve_auth

std::unique_ptr<CurveAuth> Mads::Agent::_curve_auth = nullptr

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1332 of file agent.hpp.

◆ _dropped_messages

std::atomic<size_t> Mads::Agent::_dropped_messages {0}

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1363 of file agent.hpp.

◆ _event_cv

std::condition_variable Mads::Agent::_event_cv

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1355 of file agent.hpp.

◆ _event_mtx

std::mutex Mads::Agent::_event_mtx

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1354 of file agent.hpp.

◆ _high_res_loop

bool Mads::Agent::_high_res_loop = false

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1326 of file agent.hpp.

◆ _hostname

std::string Mads::Agent::_hostname

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1285 of file agent.hpp.

◆ _init_done

bool Mads::Agent::_init_done = false

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1316 of file agent.hpp.

◆ _io_reads_subscriber

std::atomic<bool> Mads::Agent::_io_reads_subscriber {false}

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1349 of file agent.hpp.

◆ _io_thread

std::thread Mads::Agent::_io_thread

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1345 of file agent.hpp.

◆ _key_dir

std::filesystem::path Mads::Agent::_key_dir

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1333 of file agent.hpp.

◆ _last_blob

std::tuple<std::string, std::string, std::vector<unsigned char> > Mads::Agent::_last_blob

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1310 of file agent.hpp.

◆ _last_message

std::tuple<std::string, std::shared_ptr<LazyPayload> > Mads::Agent::_last_message

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1309 of file agent.hpp.

◆ _last_value_only

bool Mads::Agent::_last_value_only = false

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1334 of file agent.hpp.

◆ _latest_message

SharedLatest<zmq::multipart_t> Mads::Agent::_latest_message

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1336 of file agent.hpp.

◆ _message_state_mutex

std::mutex Mads::Agent::_message_state_mutex

mutableprotected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1311 of file agent.hpp.

◆ _name

std::string Mads::Agent::_name

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1286 of file agent.hpp.

Referenced by Mads::Dummy::load_settings().

◆ _own_clock_measurement

Mads::ClockOffsetResult Mads::Agent::_own_clock_measurement

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1382 of file agent.hpp.

◆ _pub_endpoint

std::string Mads::Agent::_pub_endpoint

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1290 of file agent.hpp.

Referenced by pub_endpoint(), and set_pub_endpoint().

◆ _pub_monitor

Mads::SocketMonitor Mads::Agent::_pub_monitor

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1304 of file agent.hpp.

◆ _pub_topic

std::string Mads::Agent::_pub_topic

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1291 of file agent.hpp.

Referenced by pub_topic(), and Mads::Bridge::route().

◆ _publish_mutex

std::mutex Mads::Agent::_publish_mutex

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1358 of file agent.hpp.

◆ _publisher

zmq::socket_t Mads::Agent::_publisher

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1300 of file agent.hpp.

◆ _raw_settings

std::string Mads::Agent::_raw_settings

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1288 of file agent.hpp.

◆ _rc_owns_socket

bool Mads::Agent::_rc_owns_socket = false

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1360 of file agent.hpp.

◆ _receive_timeout

int Mads::Agent::_receive_timeout = DEFAULT_RECEIVE_TIMEOUT_MS

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1314 of file agent.hpp.

◆ _remote_controlled

bool Mads::Agent::_remote_controlled = false

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1324 of file agent.hpp.

◆ _restart

bool Mads::Agent::_restart = false

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1318 of file agent.hpp.

◆ _runtime

std::shared_ptr<Mads::Runtime> Mads::Agent::_runtime = std::make_shared<Mads::Runtime>()

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1319 of file agent.hpp.

Referenced by keep_running(), and runtime().

◆ _settings_fetched

bool Mads::Agent::_settings_fetched = false

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1317 of file agent.hpp.

◆ _settings_json

nlohmann::json Mads::Agent::_settings_json

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1364 of file agent.hpp.

◆ _settings_timeout

int Mads::Agent::_settings_timeout = 0

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1315 of file agent.hpp.

◆ _settings_uri

std::string Mads::Agent::_settings_uri

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1287 of file agent.hpp.

◆ _shutdown_done

bool Mads::Agent::_shutdown_done = false

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1335 of file agent.hpp.

◆ _spin_margin

std::chrono::nanoseconds Mads::Agent::_spin_margin = std::chrono::microseconds(200)

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1327 of file agent.hpp.

◆ _startup_event_thread

std::thread Mads::Agent::_startup_event_thread

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1353 of file agent.hpp.

◆ _status

std::map<std::string, std::shared_ptr<LazyPayload> > Mads::Agent::_status

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1308 of file agent.hpp.

◆ _stopping

std::atomic<bool> Mads::Agent::_stopping {false}

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1323 of file agent.hpp.

Referenced by keep_running().

◆ _sub_endpoint

std::string Mads::Agent::_sub_endpoint

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1290 of file agent.hpp.

Referenced by set_sub_endpoint(), and sub_endpoint().

◆ _sub_monitor

Mads::SocketMonitor Mads::Agent::_sub_monitor

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1305 of file agent.hpp.

◆ _sub_topic

std::vector<std::string> Mads::Agent::_sub_topic

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1293 of file agent.hpp.

Referenced by set_sub_topic(), and sub_topic().

◆ _subscriber

zmq::socket_t Mads::Agent::_subscriber

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1301 of file agent.hpp.

◆ _time_step

std::chrono::nanoseconds Mads::Agent::_time_step = std::chrono::nanoseconds(0)

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1325 of file agent.hpp.

◆ _timecode_offset

double Mads::Agent::_timecode_offset = 0.0

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1328 of file agent.hpp.

◆ _watchdog_stop

std::atomic<bool> Mads::Agent::_watchdog_stop {false}

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1359 of file agent.hpp.

◆ _watchdog_thread

std::thread Mads::Agent::_watchdog_thread

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1350 of file agent.hpp.

◆ _wildcard_sub_topic

std::vector<std::string> Mads::Agent::_wildcard_sub_topic

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1298 of file agent.hpp.

◆ _wire_format

WireFormat Mads::Agent::_wire_format = WireFormat::Json

protected

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1361 of file agent.hpp.

◆ auth_verbose

Mads::auth_verbose Mads::Agent::auth_verbose = auth_verbose::off

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1150 of file agent.hpp.

◆ client_key_name

std::string Mads::Agent::client_key_name = "client"

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1152 of file agent.hpp.

◆ dummy

bool Mads::Agent::dummy = false

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1398 of file agent.hpp.

◆ server_key_name

std::string Mads::Agent::server_key_name = "broker"

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1151 of file agent.hpp.

◆ timecode_fps

double Mads::Agent::timecode_fps = MADS_FPS

Examples: /home/runner/work/MADS/MADS/src/agent.hpp.

Definition at line 1149 of file agent.hpp.

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

src/agent.hpp

Public Types

Public Member Functions

Static Public Member Functions

Public Attributes

Protected Member Functions

Static Protected Member Functions

Protected Attributes

Detailed Description

Member Typedef Documentation

◆ loop_fun_t

Constructor & Destructor Documentation

◆ Agent()

◆ ~Agent()

Member Function Documentation

◆ _announce_clock_offset()

◆ _apply_socket_options()

◆ _clock_agent_identity()

◆ _clock_wants_sync()

◆ _configure_clock_sync()

◆ _handle_clocksync_message()

◆ _run_peer_measurement()

◆ _stamp_clock_result()

◆ _start_clock_thread()

◆ _start_io_thread()

◆ _topic_matches_subscription()

◆ attachment_path()

◆ broadcast_clock_probe()

◆ clock_domain()

◆ clock_offset()

◆ compression()

◆ conflate()

◆ connect()

◆ connect_pub()

◆ connect_sub()

◆ curve_auth()

◆ delivery()

◆ disconnect()

◆ dropped_messages()

◆ enable_high_res_loop()

◆ enable_remote_control()

◆ enable_threaded_remote_control()

◆ fetch_settings()

◆ get_agent_id()

◆ get_settings()

◆ high_res_loop()

◆ high_watermark()

◆ info()

◆ init() [1/2]

◆ init() [2/2]

◆ install_loop_watchdog()

◆ install_signal_handlers()

◆ is_connected()

◆ is_crypto()

◆ keep_running()

◆ key_dir()

◆ last_blob()

◆ last_blob_view()

◆ last_json()

◆ last_message()

◆ last_topic()

◆ link_state()

◆ load_settings()

◆ loop() [1/2]

◆ loop() [2/2]

◆ measure_clock_offset()

◆ name()

◆ pub_endpoint()

◆ pub_topic()

◆ publish() [1/3]

◆ publish() [2/3]

◆ publish() [3/3]

◆ publish_raw_message()

◆ receive()

◆ receive_raw()

◆ receive_raw_message()

◆ receive_timeout()

◆ register_event()

◆ remote_control()

◆ restart()

◆ running()