adtf_file_writer.h

Go to the documentation of this file.

 
#ifndef ADTF_FILE_FILE_WRITER_INCLUDED
#define ADTF_FILE_FILE_WRITER_INCLUDED
 
#include <memory>
#include <string>
#include <chrono>
#include <unordered_map>
#include <queue>
#include <list>
 
#include <ifhd/ifhd.h>
#include "sample.h"
#include "stream_type.h"
#include "object.h"
 
namespace adtf_file
{

class OutputStream
{
    public:
        virtual void write(const void* data, size_t data_size) = 0;

        template <typename T>
        OutputStream& operator<<(const T& value)
        {
            static_assert(std::is_trivially_copyable<T>::value, "Only trivially copyable types and std::string are allowed. Alternatively, provide a specialization for adtf_file::OutputStream& operator<<(T& value).");
            write(&value, sizeof(T));
            return *this;
        }
};

template <>
OutputStream& OutputStream::operator<<(const std::string& value);

class StreamTypeSerializer: public Object
{
    public:
        virtual std::string getMetaType() const = 0;
        virtual void serialize(const StreamType& stream_type, OutputStream& stream) const = 0;
};

class ADTF2StreamTypeSerializer: public StreamTypeSerializer
{
    public:
        virtual std::string getTypeId() const = 0;
};

class StreamTypeSerializers: private std::unordered_map<std::string, std::shared_ptr<StreamTypeSerializer>>
{
    public:
        void add(const std::shared_ptr<StreamTypeSerializer>& serializer)
        {
            emplace(serializer->getMetaType(), serializer);
        }

        void serialize(const StreamType& stream_type, OutputStream& stream) const
        {
            auto& property_type = dynamic_cast<const PropertyStreamType&>(stream_type);
            auto serializer = find(property_type.getMetaType());
            if (serializer == end())
            {
                // try to find a default serializer
                serializer = find("");
                if (serializer == end())
                {
                    throw std::runtime_error("No serializer for stream type " + property_type.getMetaType());
                }
            }
 
            serializer->second->serialize(stream_type, stream);
        }

        std::string getAdtf2TypeId(const StreamType& stream_type)
        {
            auto& property_type = dynamic_cast<const PropertyStreamType&>(stream_type);
            auto serializer = find(property_type.getMetaType());
            if (serializer != end())
            {
                auto adtf2_serializer = std::dynamic_pointer_cast<ADTF2StreamTypeSerializer>(serializer->second);
                if (adtf2_serializer)
                {
                    return adtf2_serializer->getTypeId();
                }
            }
 
            return "";
        }
};

class SampleSerializer
{
    public:
        virtual std::string getId() const = 0;
        virtual void setStreamType(const StreamType& stream_type) = 0;
        virtual void serialize(const WriteSample& sample, OutputStream& stream) = 0;
};

class SampleSerializerFactory: public Object
{
    public:
        virtual std::string getId() const = 0;
        virtual std::shared_ptr<SampleSerializer> build() const = 0;
};

template <typename SERIALIZER>
class sample_serializer_factory: public SampleSerializerFactory
{
    public:
        std::string getId() const override
        {
            return SERIALIZER::id;
        }
 
        std::shared_ptr<SampleSerializer> build() const override
        {
            return std::make_shared<SERIALIZER>();
        }
};

class SampleSerializerFactories: private std::unordered_map<std::string, std::shared_ptr<SampleSerializerFactory>>
{
    public:
        void add(const std::shared_ptr<SampleSerializerFactory>& serializer_factory)
        {
            emplace(serializer_factory->getId(), serializer_factory);
        }

        std::shared_ptr<SampleSerializer> build(const std::string& id) const
        {
            try
            {
                return at(id)->build();
            }
            catch (...)
            {
                throw std::runtime_error("no sample serializer factory with id '" + id + "' available.");
            }
        }
};

class ADTFDatFileWriter: private ifhd::v400::IndexedFileWriter::ChunkDroppedCallback
{
    private:
        class Buffer: public std::vector<uint8_t>, public OutputStream
        {
 
            public:
                void write(const void* data, size_t data_size) override
                {
                    this->insert(end(), static_cast<const uint8_t*>(data), static_cast<const uint8_t*>(data) + data_size);
                }
        };
 
        class Chunk: public Buffer
        {
            public:
                size_t stream_id;
                std::chrono::nanoseconds time_stamp;
                uint32_t flags;
 
            public:
                void reset(size_t stream_id_to_set,
                           std::chrono::nanoseconds time_stamp_to_set,
                           uint32_t flags_to_set)
                {
                    this->stream_id = stream_id_to_set;
                    this->time_stamp = time_stamp_to_set;
                    this->flags = flags_to_set;
                    clear();
                }
        };
 
    public:
        enum TargetADTFVersion
        {
            adtf2 = 2,
            adtf3 = 3,
            adtf3ns = 4
        };
 
    public:
        ADTFDatFileWriter() = delete;
        ADTFDatFileWriter(const std::string& file_name,
               std::chrono::nanoseconds history_duration,
               StreamTypeSerializers type_serializers,
               TargetADTFVersion target_adtf_version = adtf3ns,
               uint32_t writer_flags = 0);
        ~ADTFDatFileWriter();
 
        ADTFDatFileWriter(const ADTFDatFileWriter&) = delete;
        ADTFDatFileWriter(ADTFDatFileWriter&&) = default;

        size_t createStream(const std::string& name,
                            const StreamType& initial_type,
                            const std::shared_ptr<SampleSerializer>& serializer);
        void write(size_t stream_id, std::chrono::nanoseconds time_stamp, const StreamType& type);
        void write(size_t stream_id, std::chrono::nanoseconds time_stamp, const WriteSample& sample);
        void writeTrigger(size_t stream_id, std::chrono::nanoseconds time_stamp);

        void quitHistory();

        std::shared_ptr<OutputStream> getExtensionStream(const std::string& name,
                                                         uint32_t user_id,
                                                         uint32_t type_id,
                                                         uint32_t file_version_id);
        void setFileDescription(const std::string& description);
    private:
        void onChunkDropped(uint64_t index, uint16_t stream_id, uint16_t flags, timestamp_t time_stamp) override;
 
        const Chunk& serialize(size_t stream_id, std::chrono::nanoseconds time_stamp, const StreamType& type);
        const Chunk& serialize(size_t stream_id, std::chrono::nanoseconds time_stamp, const WriteSample& sample);
        const Chunk& serializeTrigger(size_t stream_id, std::chrono::nanoseconds time_stamp);
 
        void write(const Chunk& chunk);
 
        void closeAdtf2();
        void closeAdtf3();
 
    private:
        std::unique_ptr<ifhd::v500::IndexedFileWriter> _file;
        StreamTypeSerializers _type_serializers;
        TargetADTFVersion _target_adtf_version;
        size_t _stream_id_counter = 0;
        bool _history_active;
 
        struct Stream
        {
            std::string name;
            Buffer initial_type;
            std::string adtf2_initial_type_id;
            std::shared_ptr<SampleSerializer> sample_serializer;
            std::list<Buffer> type_queue;
            Chunk chunk_buffer;
            bool has_samples = false;
        };
 
        std::vector<Stream> _streams;
};
 
}
 
#endif