reader.hpp

Go to the documentation of this file.

#pragma once
 
#include "intervaltree.hpp"
#include "read_job_queue.hpp"
#include "types.hpp"
#include "visibility.hpp"
#include <cstdio>
#include <fstream>
#include <map>
#include <memory>
#include <optional>
#include <string>
#include <unordered_map>
#include <unordered_set>
#include <vector>
 
namespace mcap {
 
enum struct ReadSummaryMethod {
  NoFallbackScan,
  AllowFallbackScan,
  ForceScan,
};
 
struct MCAP_PUBLIC IReadable {
  virtual ~IReadable() = default;
 
  virtual uint64_t size() const = 0;
  virtual uint64_t read(std::byte** output, uint64_t offset, uint64_t size) = 0;
};
 
class MCAP_PUBLIC FileReader final : public IReadable {
public:
  FileReader(std::FILE* file);
 
  uint64_t size() const override;
  uint64_t read(std::byte** output, uint64_t offset, uint64_t size) override;
 
private:
  std::FILE* file_;
  std::vector<std::byte> buffer_;
  uint64_t size_;
  uint64_t position_;
};
 
class MCAP_PUBLIC FileStreamReader final : public IReadable {
public:
  FileStreamReader(std::ifstream& stream);
 
  uint64_t size() const override;
  uint64_t read(std::byte** output, uint64_t offset, uint64_t size) override;
 
private:
  std::ifstream& stream_;
  std::vector<std::byte> buffer_;
  uint64_t size_;
  uint64_t position_;
};
 
class MCAP_PUBLIC ICompressedReader : public IReadable {
public:
  virtual ~ICompressedReader() override = default;
 
  virtual void reset(const std::byte* data, uint64_t size, uint64_t uncompressedSize) = 0;
  virtual Status status() const = 0;
};
 
class MCAP_PUBLIC BufferReader final : public ICompressedReader {
public:
  void reset(const std::byte* data, uint64_t size, uint64_t uncompressedSize) override;
  uint64_t read(std::byte** output, uint64_t offset, uint64_t size) override;
  uint64_t size() const override;
  Status status() const override;
 
  BufferReader() = default;
  BufferReader(const BufferReader&) = delete;
  BufferReader& operator=(const BufferReader&) = delete;
  BufferReader(BufferReader&&) = delete;
  BufferReader& operator=(BufferReader&&) = delete;
 
private:
  const std::byte* data_;
  uint64_t size_;
};
 
class MCAP_PUBLIC ZStdReader final : public ICompressedReader {
public:
  void reset(const std::byte* data, uint64_t size, uint64_t uncompressedSize) override;
  uint64_t read(std::byte** output, uint64_t offset, uint64_t size) override;
  uint64_t size() const override;
  Status status() const override;
 
  static Status DecompressAll(const std::byte* data, uint64_t compressedSize,
                              uint64_t uncompressedSize, ByteArray* output);
  ZStdReader() = default;
  ZStdReader(const ZStdReader&) = delete;
  ZStdReader& operator=(const ZStdReader&) = delete;
  ZStdReader(ZStdReader&&) = delete;
  ZStdReader& operator=(ZStdReader&&) = delete;
 
private:
  Status status_;
  ByteArray uncompressedData_;
};
 
class MCAP_PUBLIC LZ4Reader final : public ICompressedReader {
public:
  void reset(const std::byte* data, uint64_t size, uint64_t uncompressedSize) override;
  uint64_t read(std::byte** output, uint64_t offset, uint64_t size) override;
  uint64_t size() const override;
  Status status() const override;
 
  Status decompressAll(const std::byte* data, uint64_t size, uint64_t uncompressedSize,
                       ByteArray* output);
  LZ4Reader();
  LZ4Reader(const LZ4Reader&) = delete;
  LZ4Reader& operator=(const LZ4Reader&) = delete;
  LZ4Reader(LZ4Reader&&) = delete;
  LZ4Reader& operator=(LZ4Reader&&) = delete;
  ~LZ4Reader() override;
 
private:
  void* decompressionContext_ = nullptr;  // LZ4F_dctx*
  Status status_;
  const std::byte* compressedData_;
  ByteArray uncompressedData_;
  uint64_t compressedSize_;
  uint64_t uncompressedSize_;
};
 
struct LinearMessageView;
 
struct MCAP_PUBLIC ReadMessageOptions {
public:
  Timestamp startTime = 0;
  Timestamp endTime = MaxTime;
  std::function<bool(std::string_view)> topicFilter;
  enum struct ReadOrder { FileOrder, LogTimeOrder, ReverseLogTimeOrder };
  ReadOrder readOrder = ReadOrder::FileOrder;
 
  ReadMessageOptions(Timestamp start, Timestamp end)
      : startTime(start)
      , endTime(end) {}
 
  ReadMessageOptions() = default;
 
  Status validate() const;
};
 
class MCAP_PUBLIC McapReader final {
public:
  ~McapReader();
 
  Status open(IReadable& reader);
  Status open(std::string_view filename);
  Status open(std::ifstream& stream);
 
  void close();
 
  Status readSummary(
    ReadSummaryMethod method, const ProblemCallback& onProblem = [](const Status&) {});
 
  LinearMessageView readMessages(Timestamp startTime = 0, Timestamp endTime = MaxTime);
  LinearMessageView readMessages(const ProblemCallback& onProblem, Timestamp startTime = 0,
                                 Timestamp endTime = MaxTime);
 
  LinearMessageView readMessages(const ProblemCallback& onProblem,
                                 const ReadMessageOptions& options);
 
  std::pair<ByteOffset, ByteOffset> byteRange(Timestamp startTime,
                                              Timestamp endTime = MaxTime) const;
 
  IReadable* dataSource();
 
  const std::optional<Header>& header() const;
  const std::optional<Footer>& footer() const;
  const std::optional<Statistics>& statistics() const;
 
  const std::unordered_map<ChannelId, ChannelPtr> channels() const;
  const std::unordered_map<SchemaId, SchemaPtr> schemas() const;
 
  ChannelPtr channel(ChannelId channelId) const;
  SchemaPtr schema(SchemaId schemaId) const;
 
  const std::vector<ChunkIndex>& chunkIndexes() const;
 
  const std::multimap<std::string, MetadataIndex>& metadataIndexes() const;
 
  // The following static methods are used internally for parsing MCAP records
  // and do not need to be called directly unless you are implementing your own
  // reader functionality or tests.
 
  static Status ReadRecord(IReadable& reader, uint64_t offset, Record* record);
  static Status ReadFooter(IReadable& reader, uint64_t offset, Footer* footer);
 
  static Status ParseHeader(const Record& record, Header* header);
  static Status ParseFooter(const Record& record, Footer* footer);
  static Status ParseSchema(const Record& record, Schema* schema);
  static Status ParseChannel(const Record& record, Channel* channel);
  static Status ParseMessage(const Record& record, Message* message);
  static Status ParseChunk(const Record& record, Chunk* chunk);
  static Status ParseMessageIndex(const Record& record, MessageIndex* messageIndex);
  static Status ParseChunkIndex(const Record& record, ChunkIndex* chunkIndex);
  static Status ParseAttachment(const Record& record, Attachment* attachment);
  static Status ParseAttachmentIndex(const Record& record, AttachmentIndex* attachmentIndex);
  static Status ParseStatistics(const Record& record, Statistics* statistics);
  static Status ParseMetadata(const Record& record, Metadata* metadata);
  static Status ParseMetadataIndex(const Record& record, MetadataIndex* metadataIndex);
  static Status ParseSummaryOffset(const Record& record, SummaryOffset* summaryOffset);
  static Status ParseDataEnd(const Record& record, DataEnd* dataEnd);
 
  static std::optional<Compression> ParseCompression(const std::string_view compression);
 
private:
  using ChunkInterval = internal::Interval<ByteOffset, ChunkIndex>;
  friend LinearMessageView;
 
  IReadable* input_ = nullptr;
  std::FILE* file_ = nullptr;
  std::unique_ptr<FileReader> fileInput_;
  std::unique_ptr<FileStreamReader> fileStreamInput_;
  std::optional<Header> header_;
  std::optional<Footer> footer_;
  std::optional<Statistics> statistics_;
  std::vector<ChunkIndex> chunkIndexes_;
  internal::IntervalTree<ByteOffset, ChunkIndex> chunkRanges_;
  std::multimap<std::string, AttachmentIndex> attachmentIndexes_;
  std::multimap<std::string, MetadataIndex> metadataIndexes_;
  std::unordered_map<SchemaId, SchemaPtr> schemas_;
  std::unordered_map<ChannelId, ChannelPtr> channels_;
  ByteOffset dataStart_ = 0;
  ByteOffset dataEnd_ = EndOffset;
  Timestamp startTime_ = 0;
  Timestamp endTime_ = 0;
  bool parsedSummary_ = false;
 
  void reset_();
  Status readSummarySection_(IReadable& reader);
  Status readSummaryFromScan_(IReadable& reader);
};
 
struct MCAP_PUBLIC RecordReader {
  ByteOffset offset;
  ByteOffset endOffset;
 
  RecordReader(IReadable& dataSource, ByteOffset startOffset, ByteOffset endOffset = EndOffset);
 
  void reset(IReadable& dataSource, ByteOffset startOffset, ByteOffset endOffset);
 
  std::optional<Record> next();
 
  const Status& status() const;
 
  ByteOffset curRecordOffset() const;
 
private:
  IReadable* dataSource_ = nullptr;
  Status status_;
  Record curRecord_;
};
 
struct MCAP_PUBLIC TypedChunkReader {
  std::function<void(const SchemaPtr, ByteOffset)> onSchema;
  std::function<void(const ChannelPtr, ByteOffset)> onChannel;
  std::function<void(const Message&, ByteOffset)> onMessage;
  std::function<void(const Record&, ByteOffset)> onUnknownRecord;
 
  TypedChunkReader();
  TypedChunkReader(const TypedChunkReader&) = delete;
  TypedChunkReader& operator=(const TypedChunkReader&) = delete;
  TypedChunkReader(TypedChunkReader&&) = delete;
  TypedChunkReader& operator=(TypedChunkReader&&) = delete;
 
  void reset(const Chunk& chunk, Compression compression);
 
  bool next();
 
  ByteOffset offset() const;
 
  const Status& status() const;
 
private:
  RecordReader reader_;
  Status status_;
  BufferReader uncompressedReader_;
  LZ4Reader lz4Reader_;
  ZStdReader zstdReader_;
};
 
struct MCAP_PUBLIC TypedRecordReader {
  std::function<void(const Header&, ByteOffset)> onHeader;
  std::function<void(const Footer&, ByteOffset)> onFooter;
  std::function<void(const SchemaPtr, ByteOffset, std::optional<ByteOffset>)> onSchema;
  std::function<void(const ChannelPtr, ByteOffset, std::optional<ByteOffset>)> onChannel;
  std::function<void(const Message&, ByteOffset, std::optional<ByteOffset>)> onMessage;
  std::function<void(const Chunk&, ByteOffset)> onChunk;
  std::function<void(const MessageIndex&, ByteOffset)> onMessageIndex;
  std::function<void(const ChunkIndex&, ByteOffset)> onChunkIndex;
  std::function<void(const Attachment&, ByteOffset)> onAttachment;
  std::function<void(const AttachmentIndex&, ByteOffset)> onAttachmentIndex;
  std::function<void(const Statistics&, ByteOffset)> onStatistics;
  std::function<void(const Metadata&, ByteOffset)> onMetadata;
  std::function<void(const MetadataIndex&, ByteOffset)> onMetadataIndex;
  std::function<void(const SummaryOffset&, ByteOffset)> onSummaryOffset;
  std::function<void(const DataEnd&, ByteOffset)> onDataEnd;
  std::function<void(const Record&, ByteOffset, std::optional<ByteOffset>)> onUnknownRecord;
  std::function<void(ByteOffset)> onChunkEnd;
 
  TypedRecordReader(IReadable& dataSource, ByteOffset startOffset,
                    ByteOffset endOffset = EndOffset);
 
  TypedRecordReader(const TypedRecordReader&) = delete;
  TypedRecordReader& operator=(const TypedRecordReader&) = delete;
  TypedRecordReader(TypedRecordReader&&) = delete;
  TypedRecordReader& operator=(TypedRecordReader&&) = delete;
 
  bool next();
 
  ByteOffset offset() const;
 
  const Status& status() const;
 
private:
  RecordReader reader_;
  TypedChunkReader chunkReader_;
  Status status_;
  bool parsingChunk_;
};
 
struct MCAP_PUBLIC IndexedMessageReader {
public:
  IndexedMessageReader(McapReader& reader, const ReadMessageOptions& options,
                       const std::function<void(const Message&, RecordOffset)> onMessage);
 
  bool next();
 
  Status status() const;
 
private:
  struct ChunkSlot {
    ByteArray decompressedChunk;
    ByteOffset chunkStartOffset;
    int unreadMessages = 0;
  };
  size_t findFreeChunkSlot();
  void decompressChunk(const Chunk& chunk, ChunkSlot& slot);
  Status status_;
  McapReader& mcapReader_;
  RecordReader recordReader_;
  LZ4Reader lz4Reader_;
  ReadMessageOptions options_;
  std::unordered_set<ChannelId> selectedChannels_;
  std::function<void(const Message&, RecordOffset)> onMessage_;
  internal::ReadJobQueue queue_;
  std::vector<ChunkSlot> chunkSlots_;
};
 
struct MCAP_PUBLIC LinearMessageView {
  struct MCAP_PUBLIC Iterator {
    using iterator_category = std::input_iterator_tag;
    using difference_type = int64_t;
    using value_type = MessageView;
    using pointer = const MessageView*;
    using reference = const MessageView&;
 
    reference operator*() const;
    pointer operator->() const;
    Iterator& operator++();
    void operator++(int);
    MCAP_PUBLIC friend bool operator==(const Iterator& a, const Iterator& b);
    MCAP_PUBLIC friend bool operator!=(const Iterator& a, const Iterator& b);
 
  private:
    friend LinearMessageView;
 
    Iterator() = default;
    Iterator(McapReader& mcapReader, ByteOffset dataStart, ByteOffset dataEnd,
             const ReadMessageOptions& options, const ProblemCallback& onProblem);
 
    class Impl {
    public:
      Impl(McapReader& mcapReader, ByteOffset dataStart, ByteOffset dataEnd,
           const ReadMessageOptions& options, const ProblemCallback& onProblem);
 
      Impl(const Impl&) = delete;
      Impl& operator=(const Impl&) = delete;
      Impl(Impl&&) = delete;
      Impl& operator=(Impl&&) = delete;
 
      void increment();
      reference dereference() const;
      bool has_value() const;
 
      McapReader& mcapReader_;
      std::optional<TypedRecordReader> recordReader_;
      std::optional<IndexedMessageReader> indexedMessageReader_;
      ReadMessageOptions readMessageOptions_;
      const ProblemCallback& onProblem_;
      Message curMessage_;
      std::optional<MessageView> curMessageView_;
 
    private:
      void onMessage(const Message& message, RecordOffset offset);
    };
 
    std::unique_ptr<Impl> impl_;
  };
 
  LinearMessageView(McapReader& mcapReader, const ProblemCallback& onProblem);
  LinearMessageView(McapReader& mcapReader, ByteOffset dataStart, ByteOffset dataEnd,
                    Timestamp startTime, Timestamp endTime, const ProblemCallback& onProblem);
  LinearMessageView(McapReader& mcapReader, const ReadMessageOptions& options, ByteOffset dataStart,
                    ByteOffset dataEnd, const ProblemCallback& onProblem);
 
  LinearMessageView(const LinearMessageView&) = delete;
  LinearMessageView& operator=(const LinearMessageView&) = delete;
  LinearMessageView(LinearMessageView&&) = default;
  LinearMessageView& operator=(LinearMessageView&&) = delete;
 
  Iterator begin();
  Iterator end();
 
private:
  McapReader& mcapReader_;
  ByteOffset dataStart_;
  ByteOffset dataEnd_;
  ReadMessageOptions readMessageOptions_;
  const ProblemCallback onProblem_;
};
 
}  // namespace mcap
 
#ifdef MCAP_IMPLEMENTATION
#  include "reader.inl"
#endif