multi_file_reader.h

Go to the documentation of this file.

 
#pragma once
#include <adtf_file/reader.h>
#include <unordered_map>
#include <vector>
#include <exception>
 
namespace adtfdat_processing
{

class MultiFileReader : public adtf_file::SeekableReader, public adtf_file::ExplicitReader
{
public:
    MultiFileReader();
 
    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;
 
    std::vector<adtf_file::Stream> getStreams() const override;
    adtf_file::FileItem getNextItem() override;
    std::optional<uint64_t> getItemCount() const override;
    std::optional<double> getProgress() const 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::ReaderFactories getFactories() const;

    std::shared_ptr<adtf_file::Reader> addFile(const std::string& file_name,
                                               std::shared_ptr<adtf_file::SampleFactory> sample_factory,
                                               std::shared_ptr<adtf_file::StreamTypeFactory> stream_type_factory);
    void build();
 
private:
    class ReaderSequence
    {
    public:
        ReaderSequence() = default;
        ReaderSequence(const ReaderSequence&) = delete;
        ReaderSequence(ReaderSequence&&) = default;
 
        void addReader(const std::shared_ptr<Reader>& reader,  std::optional<uint64_t> first_item_index)
        {
            if (!_elements.empty())
            {
                const auto get_streams = [](adtf_file::Reader& reader)
                {
                    std::unordered_map<uint16_t, std::string> streams;
                    for (const auto& stream : reader.getStreams())
                    {
                        streams[stream.stream_id] = stream.name;
                    }
                    return streams;
                };
 
                if (get_streams(*reader) != get_streams(*_elements.front().reader))
                {
                    throw std::runtime_error(
                        "Sequence files (split files) do not share the same streams with the same stream ids.");
                }
            }
 
            try
            {
                Element element{reader, reader->getNextItem(), first_item_index};
                _elements.insert(std::lower_bound(_elements.begin(), _elements.end(), element), element);
            }
            catch (const adtf_file::exceptions::EndOfFile&)
            {
                if (_elements.empty())
                {
                    _empty_streams_cache = reader->getStreams();
                }
                return;
            }
 
            _current = _elements.begin();
            _empty_streams_cache.clear();
        }
 
        adtf_file::FileItem getNext()
        {
            while (_current != _elements.end())
            {
                if (!_current->item_cache.empty())
                {
                    auto item = std::move(_current->item_cache.front());
                    _current->item_cache.pop_front();
                    return item;
                }
 
                try
                {
                    auto item = _current->reader->getNextItem();
                    adjustTimeStamp(item);
                    return item;
                }
                catch (const adtf_file::exceptions::EndOfFile&)
                {
                    ++_current;
                    if (_current != _elements.end())
                    {
                        if (const auto next = std::next(_current); next != _elements.end())
                        {
                            _current_timestamp_upper_bound = next->first_item_timestamp;
                        }
                        else
                        {
                            _current_timestamp_upper_bound.reset();
                        }
 
                        // Need additional synthetic items because this reader switch isn't visible to caller.
                        _current->item_cache.insert(_current->item_cache.begin(),
                                                    _current->initial_stream_types.cbegin(),
                                                    _current->initial_stream_types.cend());
                    }
                }
            }
 
            throw adtf_file::exceptions::EndOfFile();
        }
 
        uint64_t getItemIndexForTimeStamp(std::chrono::nanoseconds time_stamp)
        {
            std::exception_ptr last_exception;
            for (auto element = _elements.begin(); element != _elements.end(); ++element)
            {
                if (auto next = std::next(element); next != _elements.end())
                {
                    if (next->first_item_timestamp < time_stamp)
                    {
                        // Don't even query, this reader is ordered before another one that is also ordered before
                        // time_stamp.
                        continue;
                    }
                    else if (element->first_item_timestamp >= time_stamp)
                    {
                        // No need to query, we already know it's the first item of this reader.
                        _seek_cache.emplace_back(element, *element->first_item_index);
                        return _seek_cache.size() - 1;
                    }
                }
                const auto seekable_reader = std::dynamic_pointer_cast<adtf_file::SeekableReader>(element->reader);
                try
                {
                    const auto item_index = seekable_reader->getItemIndexForTimeStamp(time_stamp);
                    _seek_cache.emplace_back(element, item_index);
                    return _seek_cache.size() - 1;
                }
                catch (...)
                {
                    last_exception = std::current_exception();
                }
            }
            if (last_exception)
            {
                std::rethrow_exception(last_exception);
            }
            // Reachable when _elements is empty.
            throw adtf_file::exceptions::EndOfFile();
        }
 
        std::shared_ptr<const adtf_file::StreamType> getStreamTypeBefore(uint64_t item_index, uint16_t stream_id)
        {
            if (item_index >= _seek_cache.size())
            {
                throw std::logic_error("Invalid seek cache");
            }
 
            const auto cache = _seek_cache[item_index];
            const auto seekable_reader = std::dynamic_pointer_cast<adtf_file::SeekableReader>(cache.first->reader);
            return seekable_reader->getStreamTypeBefore(cache.second, stream_id);
        }
 
        void seekTo(uint64_t item_index)
        {
            if (item_index >= _seek_cache.size())
            {
                throw std::logic_error("Invalid seek cache");
            }
 
            const auto cache = _seek_cache[item_index];
 
            {
                const auto seekable_reader = std::dynamic_pointer_cast<adtf_file::SeekableReader>(cache.first->reader);
                seekable_reader->seekTo(cache.second);
                cache.first->item_cache.clear();
            }
 
            _current = cache.first;
 
            if (const auto next = std::next(_current); next != _elements.end())
            {
                _current_timestamp_upper_bound = next->first_item_timestamp;
            }
            else
            {
                _current_timestamp_upper_bound.reset();
            }
 
            for (auto file = std::next(cache.first); file != _elements.end(); ++file)
            {
                const auto seekable_reader = std::dynamic_pointer_cast<adtf_file::SeekableReader>(file->reader);
                seekable_reader->seekTo(*file->first_item_index);
                file->item_cache.clear();
            }
        }
 
        std::vector<adtf_file::Stream> getStreams() const
        {
            if (_elements.empty())
            {
                return _empty_streams_cache;
            }
 
            auto streams = _elements.front().reader->getStreams();
 
            for (auto element = ++_elements.begin(); element != _elements.end(); ++element)
            {
                const auto next_parts = element->reader->getStreams();
                size_t stream_index = 0;
                for (const auto& stream : next_parts)
                {
                    // All items from original reader plus one synthentic stream type item per additional reader and
                    // stream.
                    streams[stream_index].item_count += stream.item_count + streams.size();
                    streams[stream_index].timestamp_of_last_item = stream.timestamp_of_last_item;
                    ++stream_index;
                }
            }
 
            return streams;
        }
 
        std::optional<uint64_t> getItemCount() const
        {
            uint64_t item_count = 0;
            for (const auto& element : _elements)
            {
                if (const auto reader_item_count = element.reader->getItemCount())
                {
                    item_count += *reader_item_count + element.initial_stream_types.size();
                }
                else
                {
                    return {};
                }
            }
            if (!_elements.empty())
            {
                // Those items are never visible.
                item_count -= _elements.front().initial_stream_types.size();
            }
            return item_count;
        }
 
        std::optional<double> getProgress() const
        {
            if (_current != _elements.end())
            {
                double progress =
                    100.0 * ::std::distance<Sequence::const_iterator>(_elements.begin(), _current) / _elements.size();
 
                if (const auto file_progress = _current->reader->getProgress())
                {
                    progress += std::min(std::max(*file_progress, 0.0), 100.0) / _elements.size();
                }
                return progress;
            }
            else
            {
                return 100.0;
            }
        }
 
    private:
        void adjustTimeStamp(adtf_file::FileItem& item)
        {
            if (_current_timestamp_upper_bound && item.time_stamp > *_current_timestamp_upper_bound)
            {
                item.time_stamp = *_current_timestamp_upper_bound;
            }
        }
 
        struct Element
        {
            Element(std::shared_ptr<Reader> reader,
                    adtf_file::FileItem first_item,
                    std::optional<uint64_t> first_item_index):
                reader(std::move(reader)),
                first_item_timestamp(first_item.time_stamp),
                first_item_index(first_item_index),
                item_cache({std::move(first_item)})
            {
                for (const auto& stream : this->reader->getStreams())
                {
                    initial_stream_types.push_back(
                        adtf_file::FileItem{stream.stream_id, first_item_timestamp, stream.initial_type});
                }
            }
 
            std::shared_ptr<Reader> reader;
            // Item of first timestamp for timestamp correction in file sequence
            std::chrono::nanoseconds first_item_timestamp;
            // First item index if the reader is seekable
            std::optional<uint64_t> first_item_index;
 
            // Already read and/or synthetic items
            std::deque<adtf_file::FileItem> item_cache;
 
            // Synthetic items for initial stream types on roll over
            std::vector<adtf_file::FileItem> initial_stream_types;
 
            operator const std::chrono::nanoseconds&() const
            {
                return first_item_timestamp;
            }
 
            bool operator<(const Element& rhs) const
            {
                return first_item_timestamp < rhs.first_item_timestamp;
            }
        };
 
        using Sequence = std::vector<Element>;
        Sequence _elements;
        Sequence::iterator _current = _elements.end();
        std::vector<adtf_file::Stream> _empty_streams_cache;
        std::optional<std::chrono::nanoseconds> _current_timestamp_upper_bound;
 
        std::vector<std::pair<Sequence::iterator, uint64_t>> _seek_cache;
    };
 
    struct StreamGroup
    {
        ReaderSequence sequence;
        std::unordered_map<uint16_t, uint16_t> stream_ids;
 
        adtf_file::FileItem getNext()
        {
            auto item = sequence.getNext();
            item.stream_id = stream_ids[item.stream_id];
            return item;
        }
    };
 
    std::vector<std::pair<std::shared_ptr<Reader>, std::optional<uint64_t>>> _readers;
    std::vector<StreamGroup> _stream_groups;
    std::vector<adtf_file::Stream> _streams;
    struct GroupStream
    {
        size_t stream_group_index;
        uint16_t stream_id;
    };
    std::vector<GroupStream> _group_streams;
 
    struct QueueItem
    {
        StreamGroup* file;
        adtf_file::FileItem file_item;
    };
    std::multimap<std::chrono::nanoseconds, QueueItem> _queue;
 
    using stream_group_indices = std::vector<std::optional<uint64_t>>;
    using stream_group_cache_item =
        std::pair<std::chrono::nanoseconds, stream_group_indices>;
    std::vector<stream_group_cache_item> _seek_cache;
 
    bool _all_readers_seekable = true;
};

using MultiFileReaderFactory = adtf_file::ReaderFactoryImplementation<MultiFileReader>;
 
} // namespace adtfdat_processing