Asynchronous VLink message recorder built on top of MessageLoop. More...

#include <bag_writer.h>

Inheritance diagram for vlink::BagWriter:

Collaboration diagram for vlink::BagWriter:

Classes
struct	Config
	Recording behaviour, split policy and resource budgets. More...

Public Types
enum	CompressType : uint8_t { kCompressNone = 0 , kCompressAuto = 1 , kCompressZstd = 2 , kCompressLz4 = 3 , kCompressLzav = 4 }
	Compression codec selector understood by the writer backends. More...

using	SplitCallback = MoveFunction< void(int split_index, const std::string &split_filename)>
	Notification fired when the writer rotates to a new split file. More...

using	SchemaCallback = MoveFunction< SchemaData(const std::string &ser_type, SchemaType schema_type)>
	Schema resolver used by the writer when a previously unseen URL is recorded. More...

using	SystemClock = std::chrono::time_point< std::chrono::system_clock, std::chrono::milliseconds >
	System clock alias used when formatting timestamps into split file names. More...

Public Types inherited from vlink::MessageLoop
enum	Type : uint8_t { kNormalType = 0 , kLockfreeType = 1 , kPriorityType = 2 }
	Internal queue implementation type. More...

enum	Strategy : uint8_t { kOptimizationStrategy = 0 , kPopStrategy = 1 , kBlockStrategy = 2 }
	Back-pressure strategy applied when the bounded queue is at capacity. More...

enum	Priority : uint16_t { kNoPriority = 0 , kLowestPriority = 1 , kTimerPriority = 50 , kNormalPriority = 100 , kHighestPriority = std::numeric_limits<uint16_t>::max() }
	Built-in priority levels for `kPriorityType` loops; higher values dispatch first. More...

using	Callback = MoveFunction< void()>
	Callback type for tasks and event handlers. More...

Public Member Functions
	BagWriter (const std::string &path, const Config &config={})
	Constructs the base writer and opens the output file. More...

virtual	~BagWriter ()
	Halts the loop, flushes pending writes and closes the file. More...

virtual void	bind_plugin_interface (const std::shared_ptr< BagPluginInterface > &plugin_interface)
	Attaches a custom frame-forwarding plugin to this writer. More...

virtual void	clear_plugin_interface ()
	Detaches the currently bound plugin, if any. More...

virtual void	register_split_callback (SplitCallback &&callback, bool before)=0
	Installs a hook fired around split rotation. More...

virtual void	register_schema_callback (SchemaCallback &&callback)=0
	Installs the resolver invoked when an unseen serialisation type is recorded. More...

virtual bool	push_schema (const SchemaData &schema_data, bool immediate=false)=0
	Embeds a schema descriptor into the bag for downstream introspection. More...

int64_t	push (const Frame &frame, bool immediate=false)
	Records a single frame to the bag. More...

BagWriter &	operator<< (const Frame &frame)
	Streaming shorthand for `push(frame)`. More...

BagWriter &	operator<< (const SchemaData &schema_data)
	Streaming shorthand for `push_schema(schema_data)`. More...

bool	fail () const noexcept
	Returns whether a previous `operator<<` observed a write failure. More...

	operator bool () const noexcept
	Reports whether the streaming write state is still good (no latched failure). More...

void	clear () noexcept
	Clears a latched `fail()` state so streaming writes can resume being observed. More...

virtual bool	is_dumping () const =0
	Returns the backend-specific "dump in progress" flag. More...

virtual bool	is_split_mode () const =0
	Returns whether split mode is currently in effect. More...

virtual int	get_split_index () const =0
	Returns the zero-based index of the active split file. More...

virtual void	set_url_loss (const std::string &url, double loss)
	Records the expected loss ratio for `url` as bag metadata. More...

Public Member Functions inherited from vlink::MessageLoop
	MessageLoop ()
	Constructs a loop with the default `kNormalType` queue. More...

	MessageLoop (Type type)
	Constructs a loop with the given queue type. More...

virtual	~MessageLoop ()
	Destructor; requests quit and joins the dispatcher thread if needed. More...

void	set_name (const std::string &name)
	Sets a human-readable name visible to profiling tools. More...

const std::string &	get_name () const
	Returns the loop name set via `set_name`. More...

Type	get_type () const
	Returns the queue type this loop was constructed with. More...

Strategy	get_strategy () const
	Returns the active back-pressure strategy. More...

void	set_strategy (Strategy strategy)
	Replaces the back-pressure strategy. More...

void	register_begin_handler (Callback &&callback)
	Registers a callback fired once at loop thread startup. More...

void	register_end_handler (Callback &&callback)
	Registers a callback fired once when the loop thread exits. More...

void	register_idle_handler (Callback &&callback)
	Registers a callback fired every time the queue becomes empty. More...

bool	run ()
	Runs the loop on the calling thread until `quit` is requested. More...

bool	async_run ()
	Starts the loop on a new background thread. More...

bool	spin ()
	Alias of `run` blocking the calling thread. More...

bool	spin_once (bool block=true)
	Processes one batch of pending tasks and timers on the calling thread. More...

bool	quit (bool force=false)
	Requests the loop to exit. More...

bool	wait_for_quit (int ms=Timer::kInfinite, bool check=true)
	Waits until the loop has fully exited. More...

bool	post_task (Callback &&callback)
	Posts a task for execution on the loop thread. More...

TaskHandle	post_task_handle (Callback &&callback, const PostTaskOptions &options={})
	Tracked variant of `post_task` returning a `TaskHandle`. More...

bool	post_task_with_priority (Callback &&callback, uint16_t priority)
	Posts a task with an explicit priority on a `kPriorityType` loop. More...

TaskHandle	post_task_with_priority_handle (Callback &&callback, uint16_t priority, const PostTaskOptions &options={})
	Tracked variant of `post_task_with_priority` returning a `TaskHandle`. More...

template<typename CallbackT , typename = std::enable_if_t<!std::is_convertible_v<CallbackT, Schedule::RetCallback>>>
Schedule::Status	exec_task (const Schedule::Config &config, CallbackT &&callback)
	Schedules a `void-returning` callable and returns a chainable `Schedule::Status`. More...

template<typename CallbackT , typename = std::enable_if_t<std::is_convertible_v<CallbackT, Schedule::RetCallback>>>
Schedule::RetStatus	exec_task (const Schedule::Config &config, CallbackT &&callback)
	Schedules a `bool-returning` callable and returns a chainable `Schedule::RetStatus`. More...

bool	wakeup ()
	Wakes the loop thread if it is suspended in its idle wait. More...

void	reset_lockfree_capacity ()
	Recreates the lock-free queue, clearing all queued tasks and counters. More...

bool	is_running () const
	Reports whether the loop is currently running. More...

bool	is_ready_to_quit () const
	Reports whether `quit` has been requested and the loop is winding down. More...

bool	is_busy () const
	Reports whether the loop is currently executing a task. More...

size_t	get_task_count () const
	Returns the current pending task count. More...

virtual bool	wait_for_idle (int ms=Timer::kInfinite, bool check=true)
	Waits until the loop has drained its queue and is not executing a task. More...

virtual size_t	get_max_task_count () const
	Returns the maximum queue depth. More...

virtual size_t	get_max_timer_count () const
	Returns the maximum number of timers that can be attached. More...

virtual uint32_t	get_max_elapsed_time () const
	Returns the maximum allowed task execution time in milliseconds. More...

virtual bool	is_in_same_thread () const
	Reports whether the calling thread is owned by this loop. More...

std::shared_ptr< AliveState >	get_alive_state () const
	Returns the shared lifetime flag used by cross-thread bridges. More...

template<class FunctionT , class... ArgsT, typename ResultT = std::invoke_result_t<FunctionT, ArgsT...>>
std::future< ResultT >	invoke_task (FunctionT &&function, ArgsT &&... args)
	Dispatches a callable to the loop thread and returns a `std::future` for the result. More...

template<class FunctionT , class... ArgsT, typename ResultT = std::invoke_result_t<FunctionT, ArgsT...>>
std::future< ResultT >	invoke_task_with_priority (FunctionT &&function, uint16_t priority, ArgsT &&... args)
	Priority variant of `invoke_task`; requires a `kPriorityType` loop. More...

Static Public Member Functions
static std::shared_ptr< BagWriter >	create (const std::string &path, const Config &config={})
	Builds the concrete writer matching the extension of `path`. More...

static std::shared_ptr< BagWriter >	filter_get (const std::string &path)
	Returns the cached writer for `path`, lazily creating and starting one. More...

static BagWriter *	global_get ()
	Returns the singleton writer driven by the `VLINK_BAG_PATH` environment variable. More...

Protected Member Functions
virtual int64_t	record (const Frame &frame, bool immediate)=0

virtual int64_t	get_record_timestamp () const =0

std::string	convert_recorded_url (const std::string &url) const

std::vector< std::string >	recorded_urls_for_origin (const std::string &url) const

std::string	recover_recorded_url (const std::string &url) const

void	get_url_meta (const std::string &url, const std::string &ser, int &url_index, int &ser_index) const

void	get_url_meta (int url_index, int ser_index, std::string &url, std::string &ser) const

std::mutex &	sample_mutex ()

std::unordered_map< std::string, double > &	url_loss_map_ref ()

std::unordered_map< std::string, double > &	total_url_loss_map_ref ()

void	flush_plugin ()

void	detach_plugin ()

Protected Member Functions inherited from vlink::MessageLoop
virtual void	on_begin ()
	Hook invoked once on the loop thread before the first task runs. More...

virtual void	on_end ()
	Hook invoked once on the loop thread after the last task runs. More...

virtual void	on_idle ()
	Hook invoked on the loop thread each time the queue becomes empty. More...

virtual void	on_task_changed (Callback &&callback, uint32_t start_time)
	Dispatches a ready task on the loop thread. More...

virtual void	on_task_timeout (Callback &&callback, uint32_t elapsed_time)
	Hook invoked when a task exceeds `get_max_elapsed_time()`. More...

Static Protected Member Functions
static const std::string &	get_default_tag_name ()

static const std::string &	get_default_app_name ()

static SchemaPluginInterface *	get_schema_interface ()

static int32_t	get_default_timezone_diff ()

static std::string_view	convert_action (ActionType type)

static std::string	get_format_date (SystemClock *current=nullptr, bool file_format=false)

Detailed Description

Asynchronous VLink message recorder built on top of MessageLoop.

Construct via create() (or directly) and call async_run() to start the recording thread, then push messages with push(). Concrete subclasses implement every virtual persistence operation; the base class owns the shared bookkeeping and the loop wiring.

Member Typedef Documentation

◆ SchemaCallback

using vlink::BagWriter::SchemaCallback = MoveFunction<SchemaData(const std::string& ser_type, SchemaType schema_type)>

Schema resolver used by the writer when a previously unseen URL is recorded.

The writer passes the requested serialisation type together with a coarse schema family hint so that families sharing a single type name (e.g. Protobuf vs Arrow) can still be disambiguated.

◆ SplitCallback

using vlink::BagWriter::SplitCallback = MoveFunction<void(int split_index, const std::string& split_filename)>

Notification fired when the writer rotates to a new split file.

Called with the zero-based split index and the new file path. The before flag of register_split_callback() chooses whether the hook runs before or after the rotation is committed.

◆ SystemClock

using vlink::BagWriter::SystemClock = std::chrono::time_point<std::chrono::system_clock, std::chrono::milliseconds>

System clock alias used when formatting timestamps into split file names.

Member Enumeration Documentation

◆ CompressType

enum vlink::BagWriter::CompressType : uint8_t

Compression codec selector understood by the writer backends.

Value	Algorithm	Notes
kCompressNone	none	Payloads stored as raw bytes
kCompressAuto	backend	Uses the backend default (LZAV for VDB, Zstd for MCAP)
kCompressZstd	Zstandard	Active for MCAP when Zstd support is available
kCompressLz4	LZ4	Reserved selector; not currently used by built-ins
kCompressLzav	LZAV	Active for SQLite-backed VDB recordings

Enumerator
kCompressNone	Store payloads uncompressed.
kCompressAuto	Defer codec choice to the active backend.
kCompressZstd	Force Zstandard codec where supported.
kCompressLz4	Reserved selector; no built-in writer emits LZ4 today.
kCompressLzav	Force LZAV codec where supported.

Constructor & Destructor Documentation

◆ BagWriter()

vlink::BagWriter::BagWriter	(	const std::string &	path,
		const Config &	config = `{}`
	)

explicit

Constructs the base writer and opens the output file.

The recording loop is not yet running; call async_run() before any push().

Parameters

path	Output file path.
config	Recording configuration.

◆ ~BagWriter()

virtual vlink::BagWriter::~BagWriter ( )

virtual

Halts the loop, flushes pending writes and closes the file.

Member Function Documentation

◆ bind_plugin_interface()

virtual void vlink::BagWriter::bind_plugin_interface ( const std::shared_ptr< BagPluginInterface > & plugin_interface )

virtual

Attaches a custom frame-forwarding plugin to this writer.

The plugin's on_write() hook runs for every frame before it is persisted; it re-emits each frame through do_callback(), and may transcode, drop, fan out, or buffer and reorder frames by their true data-plane time (a sliding-window reorder) before they reach the bag. The writer supplies the record sink via BagPluginInterface::register_callback() and binds the plugin with BagPluginInterface::Direction::kWrite. Passing nullptr detaches and clears the previous plugin's sink.

Parameters

plugin_interface Plugin instance, or nullptr to detach the current binding.

See also: clear_plugin_interface() for the named equivalent of passing nullptr.

◆ clear()

void vlink::BagWriter::clear ( )

noexcept

Clears a latched fail() state so streaming writes can resume being observed.

◆ clear_plugin_interface()

virtual void vlink::BagWriter::clear_plugin_interface ( )

virtual

Detaches the currently bound plugin, if any.

Convenience wrapper equivalent to bind_plugin_interface(nullptr): flushes the bound plugin's pending frames, clears its record sink and drops the binding. Safe to call when no plugin is bound (in which case it is a no-op).

◆ convert_action()

static std::string_view vlink::BagWriter::convert_action ( ActionType type )

staticprotected

◆ convert_recorded_url()

std::string vlink::BagWriter::convert_recorded_url ( const std::string & url ) const

protected

◆ create()

static std::shared_ptr<BagWriter> vlink::BagWriter::create	(	const std::string &	path,
		const Config &	config = `{}`
	)

static

Builds the concrete writer matching the extension of path.

Suffix dispatch: .vdb / .vdbx select VDBWriter, .vcap / .vcapx select VCAPWriter; other suffixes return nullptr. The returned writer is open but idle until async_run() starts its loop.

Parameters

path	Output file path.
config	Recording configuration.

Returns: Shared pointer to the new writer, or nullptr on unsupported suffix.

◆ detach_plugin()

void vlink::BagWriter::detach_plugin ( )

protected

◆ fail()

bool vlink::BagWriter::fail ( ) const

noexcept

Returns whether a previous operator<< observed a write failure.

Latches when operator<<(const Frame&) sees a negative push() result or operator<<(const SchemaData&) sees a false push_schema() result. Plain push() / push_schema() never alter this flag – callers of those keep using their return values. Cleared by clear().

◆ filter_get()

static std::shared_ptr<BagWriter> vlink::BagWriter::filter_get ( const std::string & path )

static

Returns the cached writer for path, lazily creating and starting one.

Looks up the process-wide writer registry. When no entry exists, a writer is built by create(), its loop is started with async_run(), and it is registered for reuse. The registry releases the entry automatically when the last shared owner goes away. Unsupported suffixes return nullptr and are not registered.

Parameters

path	Output file path.

Returns: Shared pointer to a started writer, or nullptr on unsupported suffix.

◆ flush_plugin()

void vlink::BagWriter::flush_plugin ( )

protected

◆ get_default_app_name()

static const std::string& vlink::BagWriter::get_default_app_name ( )

staticprotected

◆ get_default_tag_name()

static const std::string& vlink::BagWriter::get_default_tag_name ( )

staticprotected

◆ get_default_timezone_diff()

static int32_t vlink::BagWriter::get_default_timezone_diff ( )

staticprotected

◆ get_format_date()

static std::string vlink::BagWriter::get_format_date	(	SystemClock *	current = `nullptr`,
		bool	file_format = `false`
	)

staticprotected

◆ get_record_timestamp()

virtual int64_t vlink::BagWriter::get_record_timestamp ( ) const

protectedpure virtual

Implemented in vlink::VDBWriter, and vlink::VCAPWriter.

◆ get_schema_interface()

static SchemaPluginInterface* vlink::BagWriter::get_schema_interface ( )

staticprotected

◆ get_split_index()

virtual int vlink::BagWriter::get_split_index ( ) const

pure virtual

Returns the zero-based index of the active split file.

Returns: Active split index, or 0 outside split mode.

Implemented in vlink::VDBWriter, and vlink::VCAPWriter.

◆ get_url_meta() [1/2]

void vlink::BagWriter::get_url_meta	(	const std::string &	url,
		const std::string &	ser,
		int &	url_index,
		int &	ser_index
	)		const

protected

◆ get_url_meta() [2/2]

void vlink::BagWriter::get_url_meta	(	int	url_index,
		int	ser_index,
		std::string &	url,
		std::string &	ser
	)		const

protected

◆ global_get()

static BagWriter* vlink::BagWriter::global_get ( )

static

Returns the singleton writer driven by the VLINK_BAG_PATH environment variable.

On first call, the writer is created from VLINK_BAG_PATH and started. Returns nullptr when the environment variable is absent or carries an unsupported suffix.

Returns: Raw pointer to the global writer, or nullptr.

◆ is_dumping()

virtual bool vlink::BagWriter::is_dumping ( ) const

pure virtual

Returns the backend-specific "dump in progress" flag.

Implemented in vlink::VDBWriter, and vlink::VCAPWriter.

◆ is_split_mode()

virtual bool vlink::BagWriter::is_split_mode ( ) const

pure virtual

Returns whether split mode is currently in effect.

Returns: true when the bag uses a splittable multi-file container (e.g. a .vdbx / .vcapx suffix), in which case split_by_size / split_by_time control the rotation timing; false otherwise, regardless of the split_by_* values.

Implemented in vlink::VDBWriter, and vlink::VCAPWriter.

◆ operator bool()

vlink::BagWriter::operator bool ( ) const

explicitnoexcept

Reports whether the streaming write state is still good (no latched failure).

Returns true while no write failure has been latched, so if (*writer << frame) tests the post-write state.

◆ operator<<() [1/2]

BagWriter& vlink::BagWriter::operator<< ( const Frame & frame )

Streaming shorthand for push(frame).

Enqueues frame on the recording loop exactly like push(frame, false) and returns the writer so calls can be chained, e.g. *writer << frame_a << frame_b. The asynchronous (non-immediate) record path is always used; reach for push(frame, true) when a synchronous write is required. The per-frame timestamp that push() returns is not surfaced here; instead, a negative push() result (e.g. empty URL, or a synchronous record failure forwarded by a bound plugin) latches the fail() state so failures are observable without inspecting every return value.

Parameters

frame Frame to record; url must not be empty, timestamp < 0 requests auto-assign.

Returns: Reference to *this for chaining.

◆ operator<<() [2/2]

BagWriter& vlink::BagWriter::operator<< ( const SchemaData & schema_data )

Streaming shorthand for push_schema(schema_data).

Embeds schema_data through the asynchronous push_schema(schema_data, false) path and returns the writer for chaining, e.g. *writer << schema << frame. A false result – the merge task could not be enqueued, or a bound backend rejected it – latches the fail() state.

Parameters

schema_data Schema descriptor to persist.

Returns: Reference to *this for chaining.

◆ push()

int64_t vlink::BagWriter::push	(	const Frame &	frame,
		bool	immediate = `false`
	)

Records a single frame to the bag.

Enqueues a write task onto the recording loop; the actual disk write happens on the loop thread. When frame.timestamp is negative the writer assigns a recording-relative timestamp from its elapsed clock; a non-negative frame.timestamp (including 0) is recorded verbatim.

When a plugin is bound via bind_plugin_interface(), the frame is handed to the plugin's on_write() hook, which re-emits it (possibly transcoded, dropped, fanned out or reordered) through the writer's record sink into the concrete record() implementation. Because the plugin may emit asynchronously, the return value is then the assigned timestamp rather than a per-frame record result; a frame the plugin drops simply never reaches record().

Parameters

frame	Frame to record. `url` must not be empty; `timestamp` < 0 requests auto-assign.
immediate	When true, bypasses the queue and writes synchronously (honoured for frames a plugin emits synchronously from `on_write()`).

Returns: Assigned timestamp in microseconds, or a negative value on immediate validation/write failure.

◆ push_schema()

virtual bool vlink::BagWriter::push_schema	(	const SchemaData &	schema_data,
		bool	immediate = `false`
	)

pure virtual

Embeds a schema descriptor into the bag for downstream introspection.

Parameters

schema_data	Schema descriptor to persist.
immediate	When true, performs a synchronous merge on the caller's thread.

Returns: true on success; false when an immediate merge fails or a queued merge task cannot be enqueued.

Implemented in vlink::VDBWriter, and vlink::VCAPWriter.

◆ record()

virtual int64_t vlink::BagWriter::record	(	const Frame &	frame,
		bool	immediate
	)

protectedpure virtual

Implemented in vlink::VDBWriter, and vlink::VCAPWriter.

◆ recorded_urls_for_origin()

std::vector<std::string> vlink::BagWriter::recorded_urls_for_origin ( const std::string & url ) const

protected

◆ recover_recorded_url()

std::string vlink::BagWriter::recover_recorded_url ( const std::string & url ) const

protected

◆ register_schema_callback()

virtual void vlink::BagWriter::register_schema_callback ( SchemaCallback && callback )

pure virtual

Installs the resolver invoked when an unseen serialisation type is recorded.

Parameters

callback Function mapping (ser_type, schema_type) to SchemaData.

Implemented in vlink::VDBWriter, and vlink::VCAPWriter.

◆ register_split_callback()

virtual void vlink::BagWriter::register_split_callback	(	SplitCallback &&	callback,
		bool	before
	)

pure virtual

Installs a hook fired around split rotation.

Parameters

callback	Receives the new split index and the new file path.
before	When true, the hook fires before the new file is opened; otherwise after.

Implemented in vlink::VDBWriter, and vlink::VCAPWriter.

◆ sample_mutex()

std::mutex& vlink::BagWriter::sample_mutex ( )

protected

◆ set_url_loss()

virtual void vlink::BagWriter::set_url_loss	(	const std::string &	url,
		double	loss
	)

virtual

Records the expected loss ratio for url as bag metadata.

Loss values feed offline diagnostics so that intentional drops can be distinguished from unexpected loss.

Parameters

url	Topic URL.
loss	Loss ratio; values greater than 1.0 are normalised to -1.

◆ total_url_loss_map_ref()

std::unordered_map<std::string, double>& vlink::BagWriter::total_url_loss_map_ref ( )

protected

◆ url_loss_map_ref()

std::unordered_map<std::string, double>& vlink::BagWriter::url_loss_map_ref ( )

protected

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

include/vlink/extension/bag_writer.h

Classes

Public Types

Public Member Functions

Static Public Member Functions

Protected Member Functions

Static Protected Member Functions

Detailed Description

Member Typedef Documentation

◆ SchemaCallback

◆ SplitCallback

◆ SystemClock

Member Enumeration Documentation

◆ CompressType

Constructor & Destructor Documentation

◆ BagWriter()

◆ ~BagWriter()

Member Function Documentation

◆ bind_plugin_interface()

◆ clear()

◆ clear_plugin_interface()

◆ convert_action()

◆ convert_recorded_url()

◆ create()

◆ detach_plugin()

◆ fail()

◆ filter_get()

◆ flush_plugin()

◆ get_default_app_name()

◆ get_default_tag_name()

◆ get_default_timezone_diff()

◆ get_format_date()

◆ get_record_timestamp()

◆ get_schema_interface()

◆ get_split_index()

◆ get_url_meta() [1/2]

◆ get_url_meta() [2/2]

◆ global_get()

◆ is_dumping()

◆ is_split_mode()

◆ operator bool()

◆ operator<<() [1/2]

◆ operator<<() [2/2]

◆ push()

◆ push_schema()

◆ record()

◆ recorded_urls_for_origin()

◆ recover_recorded_url()

◆ register_schema_callback()

◆ register_split_callback()

◆ sample_mutex()

◆ set_url_loss()

◆ total_url_loss_map_ref()

◆ url_loss_map_ref()