replay: add support for reading from plain logs (#25053)

* read from plain log

* reduce memory copies

* cleanup

* fix test_replay
old-commit-hash: 91eb096ac98b46605c283c637ed1e6b15598dc52

tools/replay/logreader.cc

@@ -47,22 +47,22 @@ LogReader::~LogReader() {
 }
 
 bool LogReader::load(const std::string &url, std::atomic<bool> *abort, bool local_cache, int chunk_size, int retries) {
-  FileReader f(local_cache, chunk_size, retries);
-  std::string data = f.read(url, abort);
-  if (data.empty()) return false;
+  raw_ = FileReader(local_cache, chunk_size, retries).read(url, abort);
+  if (raw_.empty()) return false;
 
-  return load((std::byte*)data.data(), data.size(), abort);
+  if (url.find(".bz2") != std::string::npos) {
+    raw_ = decompressBZ2(raw_, abort);
+    if (raw_.empty()) return false;
+  }
+  return parse(abort);
 }
 
 bool LogReader::load(const std::byte *data, size_t size, std::atomic<bool> *abort) {
-  raw_ = decompressBZ2(data, size, abort);
-  if (raw_.empty()) {
-    if (!(abort && *abort)) {
-      rWarning("failed to decompress log");
-    }
-    return false;
-  }
+  raw_.assign((const char *)data, size);
+  return parse(abort);
+}
 
+bool LogReader::parse(std::atomic<bool> *abort) {
   try {
     kj::ArrayPtr<const capnp::word> words((const capnp::word *)raw_.data(), raw_.size() / sizeof(capnp::word));
     while (words.size() > 0 && !(abort && *abort)) {

tools/replay/logreader.h

@@ -52,10 +52,10 @@ public:
   ~LogReader();
   bool load(const std::string &url, std::atomic<bool> *abort = nullptr, bool local_cache = false, int chunk_size = -1, int retries = 0);
   bool load(const std::byte *data, size_t size, std::atomic<bool> *abort = nullptr);
-
   std::vector<Event*> events;
 
 private:
+  bool parse(std::atomic<bool> *abort);
   std::string raw_;
 #ifdef HAS_MEMORY_RESOURCE
   std::pmr::monotonic_buffer_resource *mbr_ = nullptr;

tools/replay/route.cc

@@ -82,9 +82,9 @@ void Route::addFileToSegment(int n, const QString &file) {
   const int pos = name.lastIndexOf("--");
   name = pos != -1 ? name.mid(pos + 2) : name;
 
-  if (name == "rlog.bz2") {
+  if (name == "rlog.bz2" || name == "rlog") {
     segments_[n].rlog = file;
-  } else if (name == "qlog.bz2") {
+  } else if (name == "qlog.bz2" || name == "qlog") {
     segments_[n].qlog = file;
   } else if (name == "fcamera.hevc") {
     segments_[n].road_cam = file;

tools/replay/tests/test_replay.cc

@@ -71,6 +71,7 @@ TEST_CASE("LogReader") {
     FileReader reader(true);
     std::string corrupt_content = reader.read(TEST_RLOG_URL);
     corrupt_content.resize(corrupt_content.length() / 2);
+    corrupt_content = decompressBZ2(corrupt_content);
     LogReader log;
     REQUIRE(log.load((std::byte *)corrupt_content.data(), corrupt_content.size()));
     REQUIRE(log.events.size() > 0);