VLink  2.1.0
A high-performance communication middleware
vlink::BagWriter类 参考abstract

Asynchronous VLink message recorder built on top of MessageLoop. 更多...

#include <bag_writer.h>

类 vlink::BagWriter 继承关系图:
vlink::BagWriter 的协作图:

struct  Config
 Recording behaviour, split policy and resource budgets. 更多...
 

Public 类型

enum  CompressType : uint8_t {
  kCompressNone = 0 , kCompressAuto = 1 , kCompressZstd = 2 , kCompressLz4 = 3 ,
  kCompressLzav = 4
}
 Compression codec selector understood by the writer backends. 更多...
 
using SplitCallback = MoveFunction< void(int split_index, const std::string &split_filename)>
 Notification fired when the writer rotates to a new split file. 更多...
 
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. 更多...
 
using SystemClock = std::chrono::time_point< std::chrono::system_clock, std::chrono::milliseconds >
 System clock alias used when formatting timestamps into split file names. 更多...
 
 Internal queue implementation type. 更多...
 Back-pressure strategy applied when the bounded queue is at capacity. 更多...
 Built-in priority levels for kPriorityType loops; higher values dispatch first. 更多...

Public 成员函数

 BagWriter (const std::string &path, const Config &config={})
 Constructs the base writer and opens the output file. 更多...
 
virtual ~BagWriter ()
 Halts the loop, flushes pending writes and closes the file. 更多...
 
virtual void bind_bag_interface (const std::shared_ptr< BagPluginInterface > &bag_interface)
 Attaches a custom frame-forwarding plugin to this writer. 更多...
 
virtual void clear_bag_interface ()
 Detaches the currently bound plugin, if any. 更多...
 
virtual void register_split_callback (SplitCallback &&callback, bool before)=0
 Installs a hook fired around split rotation. 更多...
 
virtual void register_schema_callback (SchemaCallback &&callback)=0
 Installs the resolver invoked when an unseen serialisation type is recorded. 更多...
 
virtual bool push_schema (const SchemaData &schema_data)=0
 Embeds a schema descriptor into the bag for downstream introspection. 更多...
 
int64_t push (const Frame &frame)
 Records a single frame to the bag. 更多...
 
BagWriteroperator<< (const Frame &frame)
 Streaming shorthand for push(frame). 更多...
 
BagWriteroperator<< (const SchemaData &schema_data)
 Streaming shorthand for push_schema(schema_data). 更多...
 
bool fail () const noexcept
 Returns whether a stream operation, deferred backend write or finalisation has failed. 更多...
 
 operator bool () const noexcept
 Reports whether the streaming write state is still good (no latched failure). 更多...
 
void clear () noexcept
 Clears a latched fail() state so streaming writes can resume being observed. 更多...
 
virtual bool is_dumping () const =0
 Returns the backend-specific "dump in progress" flag. 更多...
 
virtual bool is_split_mode () const =0
 Returns whether split mode is currently in effect. 更多...
 
virtual int get_split_index () const =0
 Returns the zero-based index of the active split file. 更多...
 
virtual void set_url_loss (const std::string &url, double loss)
 Records the expected loss ratio for url as bag metadata. 更多...
 
virtual void close ()
 Finalizes the backend file (final commit, metadata, footer) and latches any failure. 更多...
 

静态 Public 成员函数

static std::shared_ptr< BagWritercreate (const std::string &path, const Config &config={})
 Builds the concrete writer matching the extension of path. 更多...
 
static std::shared_ptr< BagWriterfilter_get (const std::string &path)
 Returns the cached writer for path, lazily creating and starting one. 更多...
 
static BagWriterglobal_get ()
 Returns the singleton writer driven by the VLINK_BAG_PATH environment variable. 更多...
 
static std::string get_format_date (SystemClock *current=nullptr, bool file_format=false)
 Formats a wall-clock timestamp with millisecond precision. 更多...
 

Protected 成员函数

virtual int64_t record (const Frame &frame, int64_t timestamp)=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 ()
 
bool post_persistent_task (Callback &&callback)
 
void set_fail () noexcept
 Latches the writer failure state from a concrete backend. 更多...
 

静态 Protected 成员函数

static const std::string & get_default_tag_name ()
 
static const std::string & get_default_app_name ()
 
static SchemaPluginInterfaceget_schema_interface ()
 
static int32_t get_default_timezone_diff ()
 
static std::string_view convert_action (ActionType type)
 

详细描述

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.

成员类型定义说明

◆ 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.

成员枚举类型说明

◆ CompressType

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
枚举值
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.

构造及析构函数说明

◆ 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() when Config::sync_mode is false. A synchronous writer performs frame, schema and plugin-output writes on the calling thread and does not require a recording-loop thread.

参数
pathOutput file path.
configRecording configuration.

◆ ~BagWriter()

virtual vlink::BagWriter::~BagWriter ( )
virtual

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

成员函数说明

◆ bind_bag_interface()

virtual void vlink::BagWriter::bind_bag_interface ( const std::shared_ptr< BagPluginInterface > &  bag_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.

参数
bag_interfacePlugin interface instance, or nullptr to detach the current binding.
参见
clear_bag_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_bag_interface()

virtual void vlink::BagWriter::clear_bag_interface ( )
virtual

Detaches the currently bound plugin, if any.

Convenience wrapper equivalent to bind_bag_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).

◆ close()

virtual void vlink::BagWriter::close ( )
virtual

Finalizes the backend file (final commit, metadata, footer) and latches any failure.

Idempotent; invoked automatically at destruction. Callers that must verify the close-time writes call it explicitly and then query fail() while the writer is still alive. It is not synchronised against the recording loop. After producers have stopped, detach any bound write plugin with clear_bag_interface() so its buffered tail is emitted while the loop still accepts tasks; then call wait_for_idle(), quit() and wait_for_quit() before calling close() from another thread.

注解
Declared after the original virtual interface to preserve its vtable slot ordering.

vlink::VDBWriter , 以及 vlink::VCAPWriter 重载.

◆ 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 immediately; asynchronous writers need async_run(), while synchronous writers do not.

参数
pathOutput file path.
configRecording configuration.
返回
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 stream operation, deferred backend write or finalisation has failed.

Latches when a stream insertion is rejected, a concrete backend cannot persist an accepted asynchronous frame or schema, or close() cannot finalise the bag. Callers may wait for the queue to become idle and then query this method; call close() first when close-time metadata, footer or manifest failures must also be observed. Synchronous callers continue to use the return value from push() or push_schema(). 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.

参数
pathOutput file path.
返回
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 
)
static

Formats a wall-clock timestamp with millisecond precision.

参数
currentTime point to format; nullptr formats the current system time.
file_formatWhen true, produces the file-name-safe form YYYY-MM-DD_hh-mm-ss-mmm shared by generated bag names; otherwise the log form YYYY/MM/DD hh:mm:ss:mmm.
返回
Formatted timestamp string.

◆ get_record_timestamp()

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

vlink::VDBWriter , 以及 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.

返回
Active split index, or 0 outside split mode.

vlink::VDBWriter , 以及 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.

返回
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.

vlink::VDBWriter , 以及 vlink::VCAPWriter 内被实现.

◆ is_split_mode()

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

Returns whether split mode is currently in effect.

返回
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.

vlink::VDBWriter , 以及 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).

Records frame according to Config::sync_mode and returns the writer so calls can be chained, e.g. *writer << frame_a << frame_b. The per-frame timestamp that push() returns is not surfaced; instead, a negative push() result (e.g. an empty URL, a queue or memory-limit rejection, or a synchronous record failure forwarded by a bound plugin) latches the fail() state so failures are observable without inspecting every return value.

参数
frameFrame to record; url must not be empty, timestamp < 0 requests auto-assign.
返回
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 according to Config::sync_mode 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.

参数
schema_dataSchema descriptor to persist.
返回
Reference to *this for chaining.

◆ post_persistent_task()

bool vlink::BagWriter::post_persistent_task ( Callback &&  callback)
protected

◆ push()

int64_t vlink::BagWriter::push ( const Frame frame)

Records a single frame to the bag.

The write follows the mode fixed at construction: Config::sync_mode writes on the caller's thread; otherwise a task is enqueued on the recording loop. Once accepted, an asynchronous frame is not evicted to admit a later frame. 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_bag_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().

参数
frameFrame to record. url must not be empty; timestamp < 0 requests auto-assign.
返回
Assigned timestamp in microseconds, or a negative value on validation/write failure or when an asynchronous write cannot be queued (for example, because a task or memory limit was reached).

◆ push_schema()

virtual bool vlink::BagWriter::push_schema ( const SchemaData schema_data)
pure virtual

Embeds a schema descriptor into the bag for downstream introspection.

The operation follows Config::sync_mode: synchronous writers merge on the caller's thread; asynchronous writers enqueue the merge on the recording loop.

参数
schema_dataSchema descriptor to persist.
返回
true on success; false when a synchronous merge fails or an asynchronous merge task cannot be enqueued.

vlink::VDBWriter , 以及 vlink::VCAPWriter 内被实现.

◆ record()

virtual int64_t vlink::BagWriter::record ( const Frame frame,
int64_t  timestamp 
)
protectedpure virtual

vlink::VDBWriter , 以及 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.

参数
callbackFunction mapping (ser_type, schema_type) to SchemaData.

vlink::VDBWriter , 以及 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.

参数
callbackReceives the new split index and the new file path.
beforeWhen true, the hook fires before the new file is opened; otherwise after.

vlink::VDBWriter , 以及 vlink::VCAPWriter 内被实现.

◆ sample_mutex()

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

◆ set_fail()

void vlink::BagWriter::set_fail ( )
protectednoexcept

Latches the writer failure state from a concrete backend.

◆ 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.

参数
urlTopic URL.
lossLoss 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

该类的文档由以下文件生成: