bag_processor.h File Reference

Data-plane-time sliding-window reorder buffer, a helper for BagPluginInterface plugins. More...

#include <cstdint>
#include <memory>
#include <mutex>
#include "../impl/types.h"

Include dependency graph for bag_processor.h:

Go to the source code of this file.

Classes
class	vlink::BagProcessor
	Time-sorted relay buffer keyed on the data-plane time, shared by read- and write-side plugins. More...
struct	vlink::BagProcessor::Config
	Tunables controlling the reorder buffer behaviour. More...

Namespaces
	vlink

Detailed Description

Data-plane-time sliding-window reorder buffer, a helper for BagPluginInterface plugins.

BagProcessor is the building block a BagPluginInterface plugin uses to reorder frames by their true data-plane time before re-emitting them – on the write side before a frame is persisted, on the read side before it is replayed. It is symmetric: the same engine serves both directions because both push frames in and forward the reordered output through do_callback().

Two distinct times are at play, which can disagree (out-of-order recording, async I/O, sender-side batching, transport requeue):

• Frame::timestamp – the canonical record / playback time the frame carries downstream.
• the data-plane time – the true event time carried inside the payload header, passed separately as the push() reorder key. BagProcessor emits frames in ascending data-plane-time order and remaps Frame::timestamp onto that sorted data-plane-time axis.

BagProcessor is a pure reorder buffer. The plugin extracts the data-plane time from the header (parsing or deserialising the payload as needed) and passes it to push(); BagProcessor never inspects the payload nor touches the serialisation layer. Any payload transform (compress / decompress, URL / serialisation-type rewrite) is likewise the plugin's job, applied before push() or inside the output callback before it forwards the frame.

The oldest cached frame is released only once the data-plane-time span between the oldest and newest cached frames reaches Config::min_cache_time, giving a late-but-earlier frame a chance to slot ahead of already-cached later frames. A wall-clock fallback drains the cache when a producer goes silent, so the stream can always make progress. Data-plane timestamps passed to push() are in microseconds; Config time tunables are expressed in milliseconds and converted internally.

Write-side example (reorder by header time, then persist)

class MyWritePlugin : public vlink::BagPluginInterface {
 public:
  MyWritePlugin() {
    processor_.register_output_callback([this](const vlink::Frame& f) { do_callback(f); });
  }
 
  vlink::BagPluginInterface::VersionInfo get_version_info() const override {
    return {"my-write", "1.0.0", __DATE__, "", ""};
  }
 
  void on_write(const vlink::Frame& frame) override {
    const int64_t data_timestamp = parse_header_time(frame.data);  // plugin extracts the data-plane time
    processor_.push(data_timestamp, frame);                        // buffered + reordered by data_timestamp
  }
 
  void flush() override { processor_.flush(); }                    // drain the buffered tail at teardown
 
 private:
  vlink::BagProcessor processor_;
};