fix: use static lib for mcap

Author: nomanfound

examples/recorder_and_replayer/dds_replayer.cpp

@@ -0,0 +1,387 @@
+/*
+ * Copyright(c) 2025 ZettaScale Technology and others
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
+ */
+
+#include <cassert>
+#include <cstdint>
+#include <cstring>
+#include <chrono>
+#include <iostream>
+#include <optional>
+#include <string>
+#include <thread>
+#include <unordered_map>
+#include <vector>
+
+#include "dds/dds.h"
+#include "dds/ddsrt/iovec.h"
+#include "dds/ddsi/ddsi_serdata.h"
+
+#include "mcap/reader.hpp"
+
+#include "dds_topic_descriptor_serde.h"
+
+namespace {
+
+// CLI config
+enum class Mode { Fast, Realtime, Period };
+
+struct Config {
+  std::string mcapPath;
+  Mode mode = Mode::Realtime;               // default to realtime
+  uint32_t periodMs = 0;                    // required when mode == Period
+  bool loop = false;
+  uint32_t waitSec = 5;                     // default 5
+};
+
+void printUsage(const char* argv0) {
+  std::cerr
+      << "Usage:\n"
+      << "  " << argv0 << " <file.mcap> [--mode fast|realtime|period] [--period-ms N]\n"
+      << "                              [--loop] [--wait-sec N]\n"
+      << "\n"
+      << "Notes:\n"
+      << "  - --topic filters by literal topic name; if omitted, all topics are replayed.\n"
+      << "  - Period mode requires --period-ms N (milliseconds). It is only valid with --mode period.\n"
+      << "  - Realtime mode uses global relative timing based on the first message logTime.\n"
+      << "  - Fast mode publishes as quickly as possible (no sleeps).\n"
+      << "  - --wait-sec defaults to 5 (wait before starting to publish).\n"
+      << "  - Error out if channel messageEncoding != \"cdr\" or if embedded topic_descriptor metadata is missing.\n";
+}
+
+bool parseArgs(int argc, char** argv, Config& cfg) {
+  if (argc < 2) {
+    printUsage(argv[0]);
+    return false;
+  }
+  cfg.mcapPath = argv[1];
+
+  for (int i = 2; i < argc; ++i) {
+    std::string arg(argv[i]);
+    if (arg == "--mode" && i + 1 < argc) {
+      std::string val(argv[++i]);
+      if (val == "fast") cfg.mode = Mode::Fast;
+      else if (val == "realtime") cfg.mode = Mode::Realtime;
+      else if (val == "period") cfg.mode = Mode::Period;
+      else {
+        std::cerr << "Unknown mode: " << val << "\n";
+        return false;
+      }
+    } else if (arg == "--period-ms" && i + 1 < argc) {
+      cfg.periodMs = static_cast<uint32_t>(std::stoul(argv[++i]));
+    } else if (arg == "--loop") {
+      cfg.loop = true;
+    } else if (arg == "--wait-sec" && i + 1 < argc) {
+      cfg.waitSec = static_cast<uint32_t>(std::stoul(argv[++i]));
+    } else {
+      std::cerr << "Unknown or incomplete argument: " << arg << "\n";
+      printUsage(argv[0]);
+      return false;
+    }
+  }
+
+  if (cfg.mode == Mode::Period) {
+    if (cfg.periodMs == 0) {
+      std::cerr << "Error: --period-ms N is required and must be > 0 when --mode period\n";
+      return false;
+    }
+  } else {
+    if (cfg.periodMs != 0) {
+      std::cerr << "Error: --period-ms is only valid with --mode period\n";
+      return false;
+    }
+  }
+
+  return true;
+}
+
+// Per-channel DDS context
+struct ChannelCtx {
+  dds_entity_t topic = DDS_ENTITY_NIL;
+  dds_entity_t writer = DDS_ENTITY_NIL;
+  const struct ddsi_sertype* sertype = nullptr;
+};
+
+// Build DDS writer(s) from MCAP channel metadata (topic_descriptor)
+bool buildWritersFromChannels(
+    dds_entity_t participant,
+    const std::unordered_map<mcap::ChannelId, mcap::ChannelPtr>& channels,
+    const std::optional<std::string>& topicFilter,
+    std::unordered_map<mcap::ChannelId, ChannelCtx>& out) {
+
+  out.clear();
+
+  // First, validate channels and select the subset we’ll replay
+  std::vector<mcap::ChannelPtr> selected;
+  selected.reserve(channels.size());
+  for (const auto& kv : channels) {
+    const auto& ch = kv.second;
+    if (topicFilter.has_value()) {
+      if (ch->topic != *topicFilter) continue;
+    }
+    selected.push_back(ch);
+  }
+
+  if (topicFilter.has_value() && selected.empty()) {
+    std::cerr << "Error: no channel found with topic '" << *topicFilter << "'\n";
+    return false;
+  }
+  if (selected.empty()) {
+    // Not a hard error if no channels exist at all; but nothing to replay
+    std::cerr << "Error: MCAP file contains no channels to replay\n";
+    return false;
+  }
+
+  // Validate and build each selected channel
+  for (const auto& ch : selected) {
+    if (ch->messageEncoding != "cdr") {
+      std::cerr << "Error: channel '" << ch->topic << "' has encoding '" << ch->messageEncoding
+                << "', expected 'cdr'\n";
+      return false;
+    }
+    auto it = ch->metadata.find("topic_descriptor");
+    if (it == ch->metadata.end()) {
+      std::cerr << "Error: channel '" << ch->topic
+                << "' missing required metadata 'topic_descriptor'\n";
+      return false;
+    }
+
+    // The metadata value is a std::string that contains raw bytes of the serialized topic descriptor
+    const std::string& descBytes = it->second;
+    if (descBytes.empty()) {
+      std::cerr << "Error: channel '" << ch->topic << "' has empty topic_descriptor\n";
+      return false;
+    }
+
+    // Deserialize topic descriptor
+    dds_topic_descriptor_t* desc =
+        dds_topic_descriptor_deserialize(descBytes.data(), descBytes.size());
+    if (!desc) {
+      std::cerr << "Error: failed to deserialize topic descriptor for topic '" << ch->topic << "'\n";
+      return false;
+    }
+
+    // Create topic & writer
+    dds_entity_t topic = dds_create_topic(participant, desc, ch->topic.c_str(), nullptr, nullptr);
+    dds_delete_topic_descriptor(desc);
+    if (topic < 0) {
+      std::cerr << "Error: dds_create_topic failed for topic '" << ch->topic
+                << "': " << dds_strretcode(topic) << "\n";
+      return false;
+    }
+
+    dds_entity_t writer = dds_create_writer(participant, topic, nullptr, nullptr);
+    if (writer < 0) {
+      std::cerr << "Error: dds_create_writer failed for topic '" << ch->topic
+                << "': " << dds_strretcode(writer) << "\n";
+      dds_delete(topic);
+      return false;
+    }
+
+    const struct ddsi_sertype* st = nullptr;
+    dds_return_t rc = dds_get_entity_sertype(writer, &st);
+    if (rc != DDS_RETCODE_OK || st == nullptr) {
+      std::cerr << "Error: dds_get_entity_sertype failed for topic '" << ch->topic
+                << "': " << dds_strretcode(rc) << "\n";
+      dds_delete(writer);
+      dds_delete(topic);
+      return false;
+    }
+
+    ChannelCtx ctx;
+    ctx.topic = topic;
+    ctx.writer = writer;
+    ctx.sertype = st;
+
+    out.emplace(ch->id, std::move(ctx));
+  }
+
+  return true;
+}
+
+void sleepUntilSteady(std::chrono::steady_clock::time_point target) {
+  using namespace std::chrono;
+  while (true) {
+    auto now = steady_clock::now();
+    if (now >= target) break;
+    auto remaining = target - now;
+
+    // Sleep in chunks to be resilient against spurious wakeups
+    if (remaining > milliseconds(1)) {
+      std::this_thread::sleep_for(milliseconds(1));
+    } else {
+      std::this_thread::sleep_for(remaining);
+    }
+  }
+}
+
+bool publishCdr(dds_entity_t writer, const struct ddsi_sertype* sertype,
+                const std::byte* data, size_t size) {
+  // Create serdata from the raw CDR with encapsulation header included
+  ddsrt_iovec_t iov{};
+  iov.iov_base = const_cast<std::byte*>(data);
+  iov.iov_len = static_cast<ddsrt_iov_len_t>(size);
+
+  struct ddsi_serdata* sd = ddsi_serdata_from_ser_iov(sertype, SDK_DATA, 1, &iov, size);
+  if (!sd) {
+    std::cerr << "Error: ddsi_serdata_from_ser_iov returned null\n";
+    return false;
+  }
+
+  dds_return_t rc = dds_writecdr(writer, sd);
+  // The writer keeps/makes its own ref/copy; we now drop our ref
+  ddsi_serdata_unref(sd);
+
+  if (rc != DDS_RETCODE_OK) {
+    std::cerr << "Error: dds_writecdr failed: " << dds_strretcode(rc) << "\n";
+    return false;
+  }
+  return true;
+}
+
+bool replayOnce(mcap::McapReader& reader,
+                const Config& cfg,
+                const std::unordered_map<mcap::ChannelId, ChannelCtx>& channelMap) {
+  using namespace std::chrono;
+
+  const auto onProblem = [](const mcap::Status& s) {
+    std::cerr << "MCAP read warning/error: [" << static_cast<int>(s.code) << "] " << s.message << "\n";
+  };
+
+  // iterate in global logTime order
+  // Note: If the file lacks summary, the reader handles ordering as best as possible.
+  auto messages = reader.readMessages(onProblem);
+
+  bool firstPublished = false;
+  mcap::Timestamp firstLogTime = 0;
+  steady_clock::time_point t0_steady;
+
+  for (const auto& view : messages) {
+    const auto chId = view.channel->id;
+
+    // Filter to selected channels (by topic or all)
+    auto it = channelMap.find(chId);
+    if (it == channelMap.end())
+      continue;
+
+    // Determine sleep based on mode
+    if (cfg.mode == Mode::Realtime) {
+      if (!firstPublished) {
+        firstPublished = true;
+        firstLogTime = view.message.logTime;
+        t0_steady = steady_clock::now();
+      } else {
+        // relative to first logTime, global
+        const uint64_t dt_ns = (view.message.logTime >= firstLogTime)
+                                    ? (view.message.logTime - firstLogTime)
+                                    : 0;
+        auto target = t0_steady + nanoseconds(dt_ns);
+        sleepUntilSteady(target);
+      }
+    } else if (cfg.mode == Mode::Period) {
+      if (!firstPublished) {
+        firstPublished = true;
+      } else {
+        std::this_thread::sleep_for(milliseconds(cfg.periodMs));
+      }
+    } else {
+      // Mode::Fast -> no sleep
+    }
+
+    // Publish
+    const auto& msg = view.message;
+    const auto& ctx = it->second;
+    if (!publishCdr(ctx.writer, ctx.sertype, msg.data, static_cast<size_t>(msg.dataSize))) {
+      return false;
+    }
+  }
+
+  return true;
+}
+
+} // namespace
+
+int main(int argc, char** argv) {
+  Config cfg;
+  if (!parseArgs(argc, argv, cfg)) {
+    return 2;
+  }
+
+  // Open MCAP
+  mcap::McapReader reader;
+  auto st = reader.open(cfg.mcapPath);
+  if (!st.ok()) {
+    std::cerr << "Failed to open MCAP '" << cfg.mcapPath << "': " << st.message << "\n";
+    return 2;
+  }
+
+  // Parse summary (to collect channels, schemas etc.)
+  st = reader.readSummary(mcap::ReadSummaryMethod::AllowFallbackFileScan);
+  if (!st.ok()) {
+    std::cerr << "Failed to read MCAP summary: " << st.message
+              << " (continuing; reader may still iterate messages)\n";
+    // not fatal; but channels() might be incomplete until iteration progresses
+  }
+
+  // Create participant (default domain, no QoS)
+  dds_entity_t participant = dds_create_participant(DDS_DOMAIN_DEFAULT, nullptr, nullptr);
+  if (participant < 0) {
+    std::cerr << "dds_create_participant failed: " << dds_strretcode(participant) << "\n";
+    reader.close();
+    return 2;
+  }
+
+  // Build writers from channels (+validate)
+  std::unordered_map<mcap::ChannelId, ChannelCtx> channelMap;
+  if (!buildWritersFromChannels(participant, reader.channels(), cfg.topicFilter, channelMap)) {
+    dds_delete(participant);
+    reader.close();
+    return 2;
+  }
+
+  // Wait before start to allow subscribers to match
+  if (cfg.waitSec > 0) {
+    std::cout << "Waiting " << cfg.waitSec << "s before start...\n";
+    std::this_thread::sleep_for(std::chrono::seconds(cfg.waitSec));
+  }
+
+  // Replay (possibly loop)
+  bool ok = true;
+  do {
+    // Re-open for a fresh pass if looping (messages' data pointers are ephemeral)
+    // For the first iteration we already opened the reader above.
+    if (!ok || (&reader == nullptr)) {
+      // unreachable; kept for clarity
+    }
+
+    ok = replayOnce(reader, cfg, channelMap);
+    if (!ok) break;
+
+    if (cfg.loop) {
+      // Close and reopen to reset iteration state cleanly
+      reader.close();
+      auto rst = reader.open(cfg.mcapPath);
+      if (!rst.ok()) {
+        std::cerr << "Failed to reopen MCAP for loop: " << rst.message << "\n";
+        ok = false;
+        break;
+      }
+      rst = reader.readSummary(mcap::ReadSummaryMethod::AllowFallbackFileScan);
+      if (!rst.ok()) {
+        std::cerr << "Failed to re-read MCAP summary on loop: " << rst.message << "\n";
+      }
+    }
+  } while (cfg.loop);
+
+  // Cleanup
+  for (auto& kv : channelMap) {
+    if (kv.second.writer != DDS_ENTITY_NIL) dds_delete(kv.second.writer);
+    if (kv.second.topic  != DDS_ENTITY_NIL) dds_delete(kv.second.topic);
+  }
+  dds_delete(participant);
+  reader.close();
+
+  return ok ? 0 : 1;
+}