demultiplexer.h

Go to the documentation of this file.

 
#pragma once
 
#include <string>
#include <functional>
#include <map>
 
#include <adtf_file/reader.h>
#include "processor.h"
 
#include <adtf_file/adtf3/substream_utilities.h>

namespace adtfdat_processing
{

class Demultiplexer
{
public:
    Demultiplexer(std::shared_ptr<adtf_file::Reader> reader, const ProcessorFactories& processor_factories);

    void setTimeRange(const std::optional<std::chrono::nanoseconds>& time_start,
                      const std::optional<std::chrono::nanoseconds>& time_end);

    void addProcessor(const std::string& stream_name,
                      const std::string& processor_id,
                      const std::string& destination_url,
                      const adtf_file::Configuration& configuration);

    void addProcessor(const std::string& stream_name,
                      const std::string& substream_name,
                      const std::string& processor_id,
                      const std::string& destination_url,
                      const adtf_file::Configuration& configuration);

    void process(const std::function<bool(double)>& progress_handler = {});
 
private:    
    std::shared_ptr<adtf_file::Reader> _reader;
    ProcessorFactories _processor_factories;
    std::optional<std::chrono::nanoseconds> _time_start;
    std::optional<std::chrono::nanoseconds> _time_end;
 
    using SubstreamsInType = std::unordered_map<std::string, adtf_file::adtf3::SubstreamProperties>;
 
    struct SubstreamProcessor
    {
        std::string processor_id;
        adtf_file::Configuration processor_configuration;
        std::string destination_url;
        std::shared_ptr<Processor> processor;
 
        void open_or_change(const ProcessorFactories& processor_factories, const std::string& stream_name, const adtf_file::FileItem& item);
    private:
        void open(const ProcessorFactories& processor_factories, const adtf_file::Stream& stream);
    };
 
    std::unordered_map<uint16_t, std::unordered_map<std::string, SubstreamProcessor>> _processors;
 
    struct InputStream
    {
        std::string name;
        bool use_substream_id;
        std::shared_ptr<const adtf_file::StreamType> initial_type;
        std::unordered_map<uint32_t, std::shared_ptr<Processor>> processors;
    };
    std::unordered_map<uint16_t, InputStream> _input_streams;
 
    void remapProcessors(InputStream& input_stream, const adtf_file::FileItem& item);
};
}