bag_plugin_interface.h

Go to the documentation of this file.

/*
 * Copyright (C) 2026 by Thun Lu. All rights reserved.
 * Author: Thun Lu <thun.lu@zohomail.cn>
 * Repo:   https://github.com/thun-res/vlink
 *  _    __   __      _           __
 * | |  / /  / /     (_) ____    / /__
 * | | / /  / /     / / / __ \  / //_/
 * | |/ /  / /___  / / / / / / / ,<
 * |___/  /_____/ /_/ /_/ /_/ /_/|_|
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
 
/**
 * @file bag_plugin_interface.h
 * @brief Unified plugin contract for rewriting bag traffic on both playback (read) and
 *        recording (write).
 *
 * @details
 * @c BagPluginInterface is a dynamic plugin loaded through the VLink @c Plugin framework and
 * attached either to a @c BagReader (via @c BagReader::bind_plugin_interface()) or to a
 * @c BagWriter (via @c BagWriter::bind_plugin_interface()).  A single class serves both
 * directions; an implementation discovers which side it is bound to through
 * @c get_direction() and overrides only the hooks for that side.
 *
 * The two directions are @b symmetric: each is a frame-forwarding pipeline.  The host supplies a
 * downstream sink at bind time and the plugin re-emits every @c Frame -- optionally transformed,
 * dropped, fanned out, or @e reordered -- through that sink.  Plugins never touch the raw sink
 * member directly; they emit by calling the single @c do_callback() helper.  Read and write share
 * one callable type (@c Callback, a @c const @c Frame& sink); a plugin overrides the @c on_read() or
 * @c on_write() hook for its bound direction.
 *
 * - @b Read (playback).  Frames originate inside the reader and flow @e out to the user.
 *   @c on_read() receives each frame and re-emits it via @c do_callback().  URL/type remapping is a
 *   separate, once-per-URL hook, @c convert_url_meta(), applied when the bag is opened.  On the read
 *   side @c Frame::ser_type / @c schema_type are empty (query @c BagReader::get_ser_type() if needed).
 *
 * - @b Write (recording).  Frames originate from the caller and flow @e into the bag.  @c on_write()
 *   receives each frame -- fully populated with @c ser_type / @c schema_type because recording
 *   persists them -- and re-emits it via @c do_callback().
 *
 * Both hooks may forward unchanged, transcode (e.g. record a raw image as compressed JPEG by emitting
 * new @c Frame::data plus a new @c ser_type / @c schema_type), drop a frame (by not emitting), fan a
 * frame out into several, or buffer frames and emit them @e reordered by their true data-plane time --
 * a sliding-window reorder, identical on both sides, typically built on a @c BagProcessor.
 *
 * Plugin contract:
 *
 * | Hook                | Dir   | Purpose                                                       |
 * | ------------------- | ----- | ------------------------------------------------------------- |
 * | get_version_info()  | both  | After load: return a populated VersionInfo                    |
 * | bind_direction()    | both  | At bind time: stored, observable via get_direction()          |
 * | register_callback() | both  | At bind time: store the forwarding sink                       |
 * | convert_url_meta()  | read  | Once per URL at open: true = keep, false = drop               |
 * | on_read()           | read  | Every replayed frame: re-emit via do_callback()               |
 * | on_write()          | write | Every frame before persist: re-emit via do_callback()         |
 * | do_callback()       | both  | Forward one frame to the sink (drop = not call)               |
 *
 * Lifecycle:
 *
 * @verbatim
 *   read  : load .so -> bind_direction(kRead) -> register_callback -> convert_url_meta (per URL)
 *                    -> on_read (per frame) -> do_callback -> callback_ -> user
 *   write : load .so -> bind_direction(kWrite) -> register_callback
 *                    -> on_write (per frame) -> do_callback -> callback_ -> writer persists
 * @endverbatim
 *
 * @par Read-side example (rename a topic on replay)
 * @code
 * class MyReadPlugin : public vlink::BagPluginInterface {
 *  public:
 *   VersionInfo get_version_info() const override { return {"my-read", "1.0.0", __DATE__, "", ""}; }
 *
 *   bool convert_url_meta(std::string& url, std::string& ser_type,
 *                         vlink::SchemaType& schema_type) override {
 *     if (url.rfind("dds://legacy/", 0) == 0) { url.replace(0, 13, "dds://v2/"); }
 *     (void)ser_type;
 *     (void)schema_type;
 *     return true;
 *   }
 *
 *   void on_read(const vlink::Frame& frame) override {
 *     do_callback(frame);  // forward downstream (drop by not calling)
 *   }
 * };
 * VLINK_PLUGIN_DECLARE(MyReadPlugin, 1, 0)
 * @endcode
 *
 * @par Write-side example (sliding-window reorder by true data-plane time before persist)
 * @code
 * class MyWritePlugin : public vlink::BagPluginInterface {
 *  public:
 *   MyWritePlugin() {
 *     processor_.register_output_callback([this](const vlink::Frame& frame) { do_callback(frame); });
 *   }
 *
 *   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);                        // reorder by data_timestamp; emit via
 * do_callback()
 *   }
 *
 *   void flush() override { processor_.flush(); }                    // drain the buffered tail at teardown
 *
 *  private:
 *   vlink::BagProcessor processor_;
 * };
 * VLINK_PLUGIN_DECLARE(MyWritePlugin, 1, 0)
 * @endcode
 */
 
#pragma once
 
#include <cstdint>
#include <string>
#include <utility>
 
#include "../base/functional.h"
#include "../base/plugin.h"
#include "../impl/types.h"
 
namespace vlink {
 
/**
 * @class BagPluginInterface
 * @brief Abstract plugin base shared by bag playback and bag recording.
 *
 * @details
 * The host binds an instance through @c BagReader::bind_plugin_interface() or
 * @c BagWriter::bind_plugin_interface().  At bind time the host calls @c bind_direction() to
 * record which side the plugin serves and @c register_callback() to supply the forwarding sink.
 * Every hook carries a default implementation, so an implementation overrides only the hooks
 * relevant to its direction (plus the mandatory @c get_version_info()).  Implementations are
 * expected to be thread-compatible with the host's loop thread.
 */
class BagPluginInterface {
  VLINK_PLUGIN_REGISTER(BagPluginInterface)
 
 protected:
  BagPluginInterface() = default;
 
  virtual ~BagPluginInterface() = default;
 
 public:
  /**
   * @brief Identifies whether the plugin is bound to a reader or a writer.
   */
  enum Direction : uint8_t {
    kRead = 0,   ///< Bound to a @c BagReader; the plugin forwards replayed frames.
    kWrite = 1,  ///< Bound to a @c BagWriter; the plugin forwards frames before they are persisted.
  };
 
  /**
   * @struct VersionInfo
   * @brief Build identity returned by @c get_version_info().
   */
  struct VersionInfo final {
    std::string name;       ///< Human-readable plugin display name.
    std::string version;    ///< Semantic version (e.g. @c "1.2.3").
    std::string timestamp;  ///< Build timestamp string.
    std::string tag;        ///< Source-control tag or label.
    std::string commit_id;  ///< Source-control commit hash.
  };
 
  /**
   * @brief Forwarding sink used by @c do_callback() to re-emit a frame downstream.
   *
   * @details
   * Supplied by the host (@c BagReader or @c BagWriter) at bind time and stored internally.  A single
   * @c const @c Frame& sink serves both directions: read plugins re-emit toward playback, write plugins
   * toward persistence.
   */
  using Callback = FrameCallback;
 
  /**
   * @brief Returns the build identity used by the host for logging and diagnostics.
   *
   * @return Populated @c VersionInfo describing this plugin.
   */
  [[nodiscard]] virtual VersionInfo get_version_info() const = 0;
 
  /**
   * @brief Records the binding direction so the plugin can branch on read vs write.
   *
   * @details
   * Invoked by the host at attach time before any other hook.  The value is observable from
   * the hooks through @c get_direction().
   *
   * @param direction Side the plugin is being bound to.
   */
  void bind_direction(Direction direction);
 
  /**
   * @brief Returns the side this plugin is currently bound to.
   *
   * @return @c Direction::kRead when bound to a reader, @c Direction::kWrite when bound to a writer.
   */
  [[nodiscard]] Direction get_direction() const;
 
  /**
   * @brief Stores the forwarding sink used by @c do_callback().
   *
   * @details
   * Invoked by the host's @c bind_plugin_interface() at attach time.  The plugin keeps @p callback in
   * @c callback_ and calls it from @c do_callback() to deliver a frame downstream -- toward the user's
   * playback callback on the read side, or toward persistence on the write side.  Cleared (with an
   * empty callable) on rebind and at host teardown, so a plugin-owned worker thread cannot reach a
   * destroyed host.
   *
   * @param callback Sink that forwards a frame downstream.
   */
  void register_callback(Callback&& callback);
 
  /**
   * @brief Rewrites or filters a stored URL before playback begins (read side).
   *
   * @details
   * Called once per URL contained in the bag when the reader opens the file.  Implementations
   * may modify any of the three parameters in place to remap topics or override schema
   * metadata.  The default implementation keeps every URL unchanged.
   *
   * @param url         URL string; may be modified in place.
   * @param ser_type    Serialisation type; may be modified in place.
   * @param schema_type Coarse schema family; may be modified in place.
   * @return @c true to retain the URL in playback; @c false to exclude it.
   */
  virtual bool convert_url_meta(std::string& url, std::string& ser_type, SchemaType& schema_type);
 
  /**
   * @brief Intercepts a replayed frame on its way to the user (read side).
   *
   * @details
   * Called for every replayed frame after timing pacing.  Forward it downstream by calling
   * @c do_callback(); transforming the payload, dropping the frame (by not emitting), fanning it out
   * (emitting several), or buffering and re-emitting frames @e reordered by data-plane time (e.g. via
   * @c BagProcessor) is permitted.  The default implementation forwards the frame unchanged.
   *
   * @note @p frame::ser_type / @c schema_type are empty on the read side; query
   *       @c BagReader::get_ser_type() / @c get_schema_type() (or stash them from @c convert_url_meta())
   *       if needed.  The payload is a shallow view valid for the duration of the call; copy it before
   *       buffering for asynchronous emit.
   *
   * @param frame Replayed frame.
   */
  virtual void on_read(const Frame& frame);
 
  /**
   * @brief Intercepts a frame before it is persisted (write side).
   *
   * @details
   * Called for every frame handed to the writer, before it is recorded.  Re-emit it by calling
   * @c do_callback().  Transcoding (rewrite @c frame.data plus @c frame.ser_type / @c schema_type, e.g.
   * raw image to JPEG), dropping the frame (by not emitting), fanning it out, or buffering and
   * re-emitting frames @e reordered by data-plane time (a sliding-window reorder, e.g. via
   * @c BagProcessor) is permitted.  The default implementation forwards the frame unchanged.
   *
   * @note The payload is a view valid for the duration of the call; a plugin that emits
   *       asynchronously must copy it before buffering.
   *
   * @note When a plugin renames the URL, the recorder learns the source-to-recorded mapping only for
   *       @e synchronous emits (within this @c on_write() call) and only when the rewrite is one-to-one,
   *       so URL-level metadata such as loss stays correctly attributed.  A plugin that both renames
   *       and emits asynchronously is responsible for any loss attribution itself.
   *
   * @note @c BagWriter::push() resolves a negative @c Frame::timestamp to the writer clock @e before
   *       calling this hook; that auto-assignment does @b not re-run on the frames a plugin emits.  A
   *       re-emitted frame is persisted with its own @c Frame::timestamp verbatim, so a plugin that
   *       constructs a fresh frame must set a resolved (non-negative) timestamp on it.
   *
   * @param frame Frame to persist.
   */
  virtual void on_write(const Frame& frame);
 
  /**
   * @brief Drains any internally-buffered frames downstream before the host unbinds and tears down.
   *
   * @details
   * Called by the host on its own thread, while its sink is still valid, right before it detaches this
   * plugin (read side: at reader teardown; write side: at writer teardown, before the recording loop is
   * stopped).  A plugin that buffers frames for asynchronous re-emit (e.g. a @c BagProcessor reorder
   * buffer) must override this to flush those frames synchronously -- typically @c processor_.flush() --
   * so a buffered tail is recorded / replayed instead of dropped.  The default implementation is a no-op
   * (a synchronous plugin holds nothing back).  After @c flush() returns the host stops delivering this
   * plugin's emitted frames, so any frame produced afterwards is ignored.
   */
  virtual void flush();
 
  /**
   * @brief Forwards one frame downstream through the registered sink.
   *
   * @details
   * The emit helper a plugin calls -- typically from @c on_read() / @c on_write() or from a reorder
   * buffer's output callback -- to deliver a frame downstream without touching @c callback_ directly.
   * Invokes @c callback_ when one is registered and is otherwise a no-op.
   *
   * @param frame Frame to forward to the sink.
   */
  void do_callback(const Frame& frame);
 
 protected:
  Direction direction_{Direction::kRead};
 
 private:
  Callback callback_;
 
  VLINK_DISALLOW_COPY_AND_ASSIGN(BagPluginInterface)
};
 
////////////////////////////////////////////////////////////////
/// Details
////////////////////////////////////////////////////////////////
 
inline void BagPluginInterface::bind_direction(Direction direction) { direction_ = direction; }
 
inline BagPluginInterface::Direction BagPluginInterface::get_direction() const { return direction_; }
 
inline void BagPluginInterface::register_callback(Callback&& callback) { callback_ = std::move(callback); }
 
inline bool BagPluginInterface::convert_url_meta(std::string& url, std::string& ser_type,
                                                 SchemaType& schema_type) {  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
  (void)url;
  (void)ser_type;
  (void)schema_type;
 
  return true;  // LCOV_EXCL_LINE GCOVR_EXCL_LINE
}
 
inline void BagPluginInterface::on_read(const Frame& frame) { do_callback(frame); }
 
inline void BagPluginInterface::on_write(const Frame& frame) { do_callback(frame); }
 
inline void BagPluginInterface::flush() {}
 
inline void BagPluginInterface::do_callback(const Frame& frame) {
  if (callback_) {
    callback_(frame);
  }
}
 
}  // namespace vlink