Concrete SQLite-backed bag file recorder with transactional write caching. More...

#include <database_writer.h>

Inheritance diagram for vlink::DatabaseWriter:

Collaboration diagram for vlink::DatabaseWriter:

Public Member Functions
	DatabaseWriter (const std::string &path, const Config &config={})
	Constructs a `DatabaseWriter` for the given `path`.
	~DatabaseWriter () override
	Destructor – commits remaining cached writes and closes the SQLite database.
void	register_split_callback (SplitCallback &&callback, bool before) override
	Registers a callback invoked when a file split occurs.
void	register_schema_callback (SchemaCallback &&callback) override
	Registers a callback that resolves serialisation type strings to `SchemaData`.
bool	push_schema (const SchemaData &schema_data, bool immediate=false) override
	Embeds a `SchemaData` into the SQLite bag for offline introspection.
int64_t	push (const std::string &url, const std::string &ser_type, SchemaType schema_type, ActionType action_type, const Bytes &data, int64_t *microseconds_timestamp=nullptr, bool immediate=false) override
	Records one message to the SQLite bag file.
bool	is_dumping () const override
	Returns `true` if the writer is actively recording to disk.
bool	is_split_mode () const override
	Returns `true` if split-file mode is active.
int	get_split_index () const override
	Returns the zero-based index of the current split file.
void	set_url_loss (const std::string &url, double loss) override
	Sets the expected message loss ratio for a given URL.
Public Member Functions inherited from vlink::BagWriter
	BagWriter (const std::string &path, const Config &config={})
	Constructs a `BagWriter` for `path` with the given `config`.
virtual	~BagWriter ()
	Destructor – stops the recording loop and flushes pending writes.
Public Member Functions inherited from vlink::MessageLoop
	MessageLoop ()
	Constructs a `MessageLoop` with `kNormalType` queue.
	MessageLoop (Type type)
	Constructs a `MessageLoop` with the specified queue type.
virtual	~MessageLoop ()
	Destructor. Calls `quit(true)` and waits for the background thread (if any).
void	set_name (const std::string &name)
	Sets a human-readable name for this loop (visible in profiling tools).
const std::string &	get_name () const
	Returns the name set via `set_name()`.
Type	get_type () const
	Returns the queue type this loop was constructed with.
Strategy	get_strategy () const
	Returns the current idle dispatch strategy.
void	set_strategy (Strategy strategy)
	Changes the idle dispatch strategy.
void	register_begin_handler (Callback &&callback)
	Registers a callback invoked once when the loop thread starts.
void	register_end_handler (Callback &&callback)
	Registers a callback invoked once when the loop thread exits.
void	register_idle_handler (Callback &&callback)
	Registers a callback invoked each time the task queue becomes empty.
bool	run ()
	Runs the message loop on the calling thread (blocking).
bool	async_run ()
	Starts the message loop on a new background thread (non-blocking).
bool	spin ()
	Runs the loop continuously in a spin mode (blocking; no background thread).
bool	spin_once (bool block=true)
	Processes one batch of pending tasks and timers.
bool	quit (bool force=false)
	Requests the loop to exit cleanly.
bool	wait_for_idle (int ms=Timer::kInfinite, bool check=true)
	Waits until the task queue is drained.
bool	wait_for_quit (int ms=Timer::kInfinite, bool check=true)
	Waits until the loop has fully exited (after `quit()` was called).
bool	post_task (Callback &&callback)
	Posts a task to the queue for execution on the loop thread.
bool	post_task_with_priority (Callback &&callback, uint16_t priority)
	Posts a task with an explicit priority (requires `kPriorityType` loop).
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)
	Posts a scheduled task and returns a `Schedule::Status` for chaining callbacks.
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)
	Posts a scheduled task and returns a `Schedule::RetStatus` for chaining callbacks.
bool	wakeup ()
	Wakes the loop thread if it is sleeping (e.g., in `kBlockStrategy`).
void	reset_lockfree_capacity ()
	Resets the lock-free queue to its initial capacity.
bool	is_running () const
	Returns `true` if the loop is currently running (started and not quit).
bool	is_ready_to_quit () const
	Returns `true` if `quit()` has been called and the loop is winding down.
bool	is_busy () const
	Returns `true` if the loop is currently executing a task.
size_t	get_task_count () const
	Returns the number of tasks currently in the queue.
virtual size_t	get_max_timer_count () const
	Returns the maximum number of timers that can be attached to this loop.
virtual uint32_t	get_max_elapsed_time () const
	Returns the maximum allowed task execution time in milliseconds.
virtual bool	is_in_same_thread () const
	Returns `true` if the calling thread is the same as the loop thread.
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.
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)
	Dispatches a callable with an explicit priority and returns a `std::future`.

Protected Member Functions
size_t	get_max_task_count () const override
	Returns the maximum queue depth.
void	on_begin () override
	Called from the loop thread just before the first task is processed.
void	on_end () override
	Called from the loop thread just after the last task has been processed.
Protected Member Functions inherited from vlink::BagWriter
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
Protected Member Functions inherited from vlink::MessageLoop
virtual void	on_idle ()
	Called from the loop thread each time the queue becomes empty.
virtual void	on_task_changed (Callback &&callback, uint32_t start_time)
	Called before each task is executed.
virtual void	on_task_timeout (Callback &&callback, uint32_t elapsed_time)
	Called when a task's execution time exceeds `get_max_elapsed_time()`.

Additional Inherited Members
Public Types inherited from vlink::BagWriter
enum	CompressType : uint8_t { kCompressNone = 0 , kCompressAuto = 1 , kCompressZstd = 2 , kCompressLz4 = 3 , kCompressLzav = 4 }
	Compression algorithm applied to each recorded payload. More...
using	SplitCallback = std::function<void(int split_index, const std::string& split_filename)>
	Callback fired when a split occurs.
using	SchemaCallback = std::function<SchemaData(const std::string& ser_type, SchemaType schema_type)>
	Callback that resolves a serialisation type string to a `SchemaData`.
using	SystemClock = std::chrono::time_point<std::chrono::system_clock, std::chrono::milliseconds>
	System clock type used for file-name timestamp generation.
Public Types inherited from vlink::MessageLoop
enum	Type : uint8_t { kNormalType = 0 , kLockfreeType = 1 , kPriorityType = 2 }
	Queue implementation type. More...
enum	Strategy : uint8_t { kOptimizationStrategy = 0 , kPopStrategy = 1 , kBlockStrategy = 2 }
	Idle strategy controlling CPU and latency trade-offs. More...
enum	Priority : uint16_t { kNoPriority = 0 , kLowestPriority = 1 , kTimerPriority = 50 , kNormalPriority = 100 , kHighestPriority = std::numeric_limits<uint16_t>::max() }
	Pre-defined task priority levels for `kPriorityType` loops. More...
using	Callback = std::function<void()>
	Callback type for tasks and event handlers.
Static Public Member Functions inherited from vlink::BagWriter
static std::shared_ptr< BagWriter >	create (const std::string &path, const Config &config={})
	Creates a concrete `BagWriter` instance for `path`.
static std::shared_ptr< BagWriter >	filter_get (const std::string &path)
	Returns an existing writer for `path`, or creates and starts a new one.
static BagWriter *	global_get ()
	Returns the process-global `BagWriter` activated by the `VLINK_BAG_PATH` environment variable.
Static Protected Member Functions inherited from vlink::BagWriter
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

Concrete SQLite-backed bag file recorder with transactional write caching.

All virtual methods from BagWriter are implemented. Prefer using BagWriter::create() for format-agnostic construction.

Constructor & Destructor Documentation

◆ DatabaseWriter()

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

explicit

Constructs a DatabaseWriter for the given path.

Parameters

path	Path to the output .vdb file. Created if it does not exist.
config	Recording configuration.

Here is the caller graph for this function:

◆ ~DatabaseWriter()

vlink::DatabaseWriter::~DatabaseWriter ( )

override

Destructor – commits remaining cached writes and closes the SQLite database.

Member Function Documentation

◆ get_max_task_count()

size_t vlink::DatabaseWriter::get_max_task_count ( ) const

overrideprotectedvirtual

Returns the maximum queue depth.

Returns: kMaxTaskSize (10000) by default.

Reimplemented from vlink::MessageLoop.

◆ get_split_index()

int vlink::DatabaseWriter::get_split_index ( ) const

nodiscardoverridevirtual

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

Implements vlink::BagWriter.

◆ is_dumping()

bool vlink::DatabaseWriter::is_dumping ( ) const

nodiscardoverridevirtual

Returns true if the writer is actively recording to disk.

Implements vlink::BagWriter.

◆ is_split_mode()

bool vlink::DatabaseWriter::is_split_mode ( ) const

nodiscardoverridevirtual

Returns true if split-file mode is active.

Implements vlink::BagWriter.

◆ on_begin()

void vlink::DatabaseWriter::on_begin ( )

overrideprotectedvirtual

Called from the loop thread just before the first task is processed.

Override in subclasses to perform per-thread initialisation.

Reimplemented from vlink::MessageLoop.

◆ on_end()

void vlink::DatabaseWriter::on_end ( )

overrideprotectedvirtual

Called from the loop thread just after the last task has been processed.

Override in subclasses to perform per-thread cleanup.

Reimplemented from vlink::MessageLoop.

Here is the call graph for this function:

◆ push()

int64_t vlink::DatabaseWriter::push	(	const std::string &	url,
		const std::string &	ser_type,
		SchemaType	schema_type,
		ActionType	action_type,
		const Bytes &	data,
		int64_t *	microseconds_timestamp = nullptr,
		bool	immediate = false )

overridevirtual

Records one message to the SQLite bag file.

Parameters

url	VLink URL of the topic.
ser_type	Serialisation type string.
schema_type	Coarse schema family for the payload.
action_type	Action type (`kPublish`, `kRequest`, etc.).
data	Serialized payload bytes.
microseconds_timestamp	Optional custom timestamp in microseconds. `nullptr` means use the current system time.
immediate	If `true`, writes synchronously bypassing the queue.

Returns: Sequence number of the recorded message, or a negative value on error.

Implements vlink::BagWriter.

◆ push_schema()

bool vlink::DatabaseWriter::push_schema	(	const SchemaData &	schema_data,
		bool	immediate = false )

overridevirtual

Embeds a SchemaData into the SQLite bag for offline introspection.

Parameters

schema_data	Schema descriptor to store.
immediate	If `true`, merges synchronously; otherwise enqueues.

Returns: false only when the immediate merge fails.

Implements vlink::BagWriter.

◆ register_schema_callback()

void vlink::DatabaseWriter::register_schema_callback ( SchemaCallback && callback )

overridevirtual

Registers a callback that resolves serialisation type strings to SchemaData.

Parameters

callback Function mapping (ser_type, schema_type) to SchemaData.

Implements vlink::BagWriter.

◆ register_split_callback()

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

overridevirtual

Registers a callback invoked when a file split occurs.

Parameters

callback	Called with (split_index, new_filename) on each split.
before	If `true`, fires before the new file is opened; otherwise after.

Implements vlink::BagWriter.

◆ set_url_loss()

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

overridevirtual

Sets the expected message loss ratio for a given URL.

Parameters

url	Topic URL.
loss	Loss ratio in the range [0.0, 1.0].

Implements vlink::BagWriter.

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

include/vlink/extension/database_writer.h

Public Member Functions

Protected Member Functions

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ DatabaseWriter()

◆ ~DatabaseWriter()

Member Function Documentation

◆ get_max_task_count()

◆ get_split_index()

◆ is_dumping()

◆ is_split_mode()

◆ on_begin()

◆ on_end()

◆ push()

◆ push_schema()

◆ register_schema_callback()

◆ register_split_callback()

◆ set_url_loss()