sorting_reader.h

Go to the documentation of this file.

 
#pragma once
 
#include <adtf_file/reader.h>
 
namespace adtfdat_processing
{
class SortingReader : public adtf_file::SeekableReader
{
public:
    void open(const std::vector<adtf_file::Stream>& streams,
              bool sample_timestamps,
              std::chrono::seconds sorting_window,
              bool trigger_for_each_sample);
 
    using adtf_file::Reader::open;
 
    adtf_file::FileItem getNextItem() final override;
 
    uint64_t getItemIndexForTimeStamp(std::chrono::nanoseconds time_stamp) override;
    uint64_t getItemIndexForStreamItemIndex(uint16_t stream_id, uint64_t stream_item_index) override;
    std::shared_ptr<const adtf_file::StreamType> getStreamTypeBefore(uint64_t item_index, uint16_t stream_id) override;
    void seekTo(uint64_t item_index) override;
 
protected:
    virtual adtf_file::FileItem getNextUnsortedItem() = 0;
 
    virtual uint64_t getUnsortedItemIndexForTimeStamp(std::chrono::nanoseconds time_stamp);
    virtual std::shared_ptr<const adtf_file::StreamType> getUnsortedStreamTypeBefore(uint64_t item_index,
                                                                                     uint16_t stream_id);
    virtual void unsortedSeekTo(uint64_t item_index);
 
private:
    void internalSeek(std::chrono::nanoseconds time_stamp);
    void fillQueue(std::chrono::nanoseconds sorting_window_duration);
 
    struct FileItemCompare
    {
        constexpr bool operator()(const adtf_file::FileItem& lhs, const adtf_file::FileItem& rhs) const
        {
            return lhs.time_stamp < rhs.time_stamp;
        }
    };
    using Queue = std::multimap<adtf_file::FileItem, std::shared_ptr<const adtf_file::StreamType>, FileItemCompare>;
    Queue _queue;
    bool _end_of_file = false;
    // Internal queue position caused by call to getStreamTypeBefore() or seekTo()
    std::optional<std::chrono::nanoseconds> _internal_seek_queue_position;
    // Queue position as set to explicit call to seekTo().
    // If _internal_seek_queue_position and _seek_queue_position are inequal,
    std::optional<std::chrono::nanoseconds> _seek_queue_position;
    std::chrono::seconds _sorting_window = std::chrono::seconds(10);
    std::unordered_map<uint16_t, std::chrono::nanoseconds> _highest_sample_item_timestamps;
    bool _sample_timestamps = true;
    bool _trigger_for_each_sample = false;
    std::vector<uint16_t> _stream_ids;
    std::unordered_map<uint16_t, std::shared_ptr<const adtf_file::StreamType>> _current_types;
    std::unordered_map<uint16_t, std::shared_ptr<const adtf_file::StreamType>> _last_published_types;
};

class SortingProxyReader : public SortingReader, public adtf_file::ExplicitReader
{
public:
    SortingProxyReader();
 
    std::string getReaderIdentifier() const override;
    void open(const std::string& file_name,
              std::shared_ptr<adtf_file::SampleFactory> sample_factory,
              std::shared_ptr<adtf_file::StreamTypeFactory> stream_type_factory) override;
 
    uint32_t getFileVersion() const override;
    std::string getDescription() const override;
    std::vector<adtf_file::Stream> getStreams() const override;
    std::vector<adtf_file::Extension> getExtensions() const override;
    adtf_file::FileItem getNextUnsortedItem() override;
    std::optional<uint64_t> getItemCount() const override;
    std::optional<double> getProgress() const override;
 
    uint64_t getUnsortedItemIndexForTimeStamp(std::chrono::nanoseconds time_stamp) override;
    std::shared_ptr<const adtf_file::StreamType> getUnsortedStreamTypeBefore(uint64_t item_index,
                                                                             uint16_t stream_id) override;
    void unsortedSeekTo(uint64_t item_index) override;
 
private:
    std::shared_ptr<Reader> _reader;
    std::vector<adtf_file::Stream> _streams;
};

using SortingReaderFactory = adtf_file::ReaderFactoryImplementation<SortingProxyReader>;
 
} // namespace adtfdat_processing