The Logger class is responsible for logging messages to a MongoDB instance. More...

#include <logger.hpp>

Inheritance diagram for Mads::Logger:

Public Member Functions
	Logger (std::string name, std::string settings_path)
	Constructs a Logger object with the specified name and settings path.

	~Logger ()

	Logger (const Logger &)=delete

Logger &	operator= (const Logger &)=delete

void	set_mongo (bool enabled=true, string uri="")
	Sets whether logging to MongoDB is enabled or disabled.

void	set_file (string filename, bool array=false)
	Sets the file name for logging.

void	set_file (bool enabled=false)
	Sets whether logging should be done to a file.

void	open_db ()
	Starts the logger.

void	close_db ()
	Stops the logger.

void	info (ostream &out=cout) override
	Overrides the info method from the Agent class.

void	register_event (event_type event)
	Registers the startup/shutdown of the logger.

void	log (tuple< string, string > *message=nullptr)
	Logs the given message to the MongoDB instance and to the log file.

void	log (message_type type)
	Logs the last message to the MongoDB instance and to the log file.

string	truncated_message (const string &message)
	Truncates the message to the maximum length.

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.

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 bool	has_mongo_support () noexcept
	Reports whether this build can log to MongoDB.

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

Public Attributes
bool	paused = false

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

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.

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 Logger class is responsible for logging messages to a MongoDB instance.

It inherits from the Agent class and provides methods for logging messages and connecting to the database.

Note: The MongoDB driver objects backing this class are hidden behind a pointer to implementation, so this header pulls in no bsoncxx/mongocxx headers and sizeof(Logger) does not depend on the driver ABI.; MongoDB support is optional at build time. When MADS is built with MADS_ENABLE_MONGOCXX=OFF the class still works as a file logger, and has_mongo_support() returns false. See has_mongo_support.

See also: Metadata for example usage. This class also has the log method.

Definition at line 42 of file logger.hpp.

Constructor & Destructor Documentation

◆ Logger() [1/2]

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

Constructs a Logger object with the specified name and settings path.

Parameters

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

◆ ~Logger()

Mads::Logger::~Logger ( )

◆ Logger() [2/2]

Mads::Logger::Logger ( const Logger & )

delete

Member Function Documentation

◆ close_db()

void Mads::Logger::close_db ( )

Stops the logger.

This method closes the logfile (if enabled) and disconnects from the MongoDB instance (if enabled)

◆ has_mongo_support()

static bool Mads::Logger::has_mongo_support ( )

staticnoexcept

Reports whether this build can log to MongoDB.

Returns: true when MADS was built with MADS_ENABLE_MONGOCXX=ON. When it returns false, logging to MongoDB is unavailable: set_mongo(true) warns and has no effect, and only file logging is performed.

◆ info()

void Mads::Logger::info ( ostream & out = cout )

overridevirtual

Overrides the info method from the Agent class.

This method provides information about the logger.

Reimplemented from Mads::Agent.

◆ log() [1/2]

void Mads::Logger::log ( message_type type )

Logs the last message to the MongoDB instance and to the log file.

Parameters

type	the type of message: json or blob.

◆ log() [2/2]

void Mads::Logger::log ( tuple< string, string > * message = nullptr )

Logs the given message to the MongoDB instance and to the log file.

Parameters

message A touple containing the topic and the message to be logged.

◆ open_db()

void Mads::Logger::open_db ( )

Starts the logger.

This method opens the logfile (if enabled) and connects to the MongoDB instance (if enabled)

◆ operator=()

Logger & Mads::Logger::operator= ( const Logger & )

delete

◆ register_event()

void Mads::Logger::register_event ( event_type event )

Registers the startup/shutdown of the logger.

This method registers the startup of the logger to the MongoDB instance.

Parameters

event The event to be registered.

◆ set_file() [1/2]

void Mads::Logger::set_file ( bool enabled = false )

Sets whether logging should be done to a file.

Parameters

enabled Whether logging to a file should be enabled or disabled. Default is false.

◆ set_file() [2/2]

void Mads::Logger::set_file	(	string	filename,
		bool	array = `false`
	)

Sets the file name for logging.

Parameters

filename	The name of the file to log to.
array	(optional) Indicates whether the log should be stored as an array. If not is is stored as a JSON object per line. Default is false.

◆ set_mongo()

void Mads::Logger::set_mongo	(	bool	enabled = `true`,
		string	uri = `""`
	)

Sets whether logging to MongoDB is enabled or disabled.

Parameters

enabled	Whether logging to MongoDB is enabled or disabled. Default is true.
uri	The URI of the MongoDB instance. If not empty, overrides the one in the settings.

◆ truncated_message()

string Mads::Logger::truncated_message ( const string & message )

Truncates the message to the maximum length.

Parameters

message The message to be truncated.

Returns: string The truncated message.

Member Data Documentation

◆ paused

bool Mads::Logger::paused = false

Definition at line 149 of file logger.hpp.

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

src/logger.hpp

Public Member Functions

Static Public Member Functions

Public Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ Logger() [1/2]

◆ ~Logger()

◆ Logger() [2/2]

Member Function Documentation

◆ close_db()

◆ has_mongo_support()

◆ info()

◆ log() [1/2]

◆ log() [2/2]

◆ open_db()

◆ operator=()

◆ register_event()

◆ set_file() [1/2]

◆ set_file() [2/2]

◆ set_mongo()

◆ truncated_message()

Member Data Documentation

◆ paused