recordingThread.hpp

Go to the documentation of this file.

 
#ifndef RTCTK_REUSABLECOMPONENT_TELEMETRYRECORDER_RECORDINGTHREAD_HPP
#define RTCTK_REUSABLECOMPONENT_TELEMETRYRECORDER_RECORDINGTHREAD_HPP
 
#include <mutex>
#include <thread>
 
#include <ipcq/reader.hpp>
#include <rtctk/telemetryRecorder/fitsRecorder.hpp>
 
namespace rtctk::telemetryRecorder {
 
enum ThreadState {
    STOPPED = 0,
    RUNNING = 1,
    PAUSED = 2,
};
 
template<typename Writer, typename Topic = typename Writer::TelemetryType, typename Reader = ipcq::Reader<Topic>>
class RecordingThread
{
    class InvalidStorageURLException : public ::elt::error::CiiBaseException {
            public:
                using ::elt::error::CiiBaseException::CiiBaseException;
                explicit InvalidStorageURLException(const std::string& uri) :
                    CiiBaseException("Invalid Telemetry Storage URL: " +uri) {}
    };
    class InvalidRecordingThreadOperation : public ::elt::error::CiiBaseException {
        public:
            using ::elt::error::CiiBaseException::CiiBaseException;
            explicit InvalidRecordingThreadOperation(const std::string& message) :
                CiiBaseException("Invalid Operation on Recording Thread: " + message) {}
    };
 
    class RecordingThreadIPCQError : public ::elt::error::CiiBaseException {
        public:
            using ::elt::error::CiiBaseException::CiiBaseException;
            explicit RecordingThreadIPCQError(const std::string& message) :
                CiiBaseException("Error in IPCQ: " + message) {}
    };
 
    class RecordingThreadInvalidStartSample : public ::elt::error::CiiBaseException {
        public:
            using ::elt::error::CiiBaseException::CiiBaseException;
            explicit RecordingThreadInvalidStartSample(const uint64_t sample_id) :
                CiiBaseException("Invalid Start SampleID " + std::to_string(sample_id)) {}
    };
 
    public:
        RecordingThread();
        RecordingThread(std::unique_ptr<Writer>&& writer, const std::string& topic);
        RecordingThread(std::unique_ptr<Writer>&& writer, std::unique_ptr<Reader>&& reader);
        RecordingThread(const RecordingThread&) = delete;
        RecordingThread(RecordingThread&&) = delete;
        ~RecordingThread();
 
        // Mutex would prevent these anyways, but let us be explicit
        RecordingThread& operator=(const RecordingThread&) = delete;
        RecordingThread& operator=(RecordingThread&&) = delete;
 
        void Open();
        void Close();
        void SetTelemetrySubSet(const std::string& mask);
        void SetSubSampleFactor(uint64_t rate);
        void SetOutputUri(const ::elt::mal::Uri& storage_uri);
        void SetTopic(const std::string& topic);
        void SetStartSampleId(const uint64_t sample_id);
 
        void Start();
        void Pause();
        void Unpause();
        void Stop();
 
    private:
        void Process();
        void ReadDataWrapper(const Topic& telemetry_data);
 
        std::mutex m_state_lock;
        enum ThreadState m_state = STOPPED;
        std::unique_ptr<Reader> m_reader;
        std::unique_ptr<Writer> m_writer;
        std::thread m_thread;
        uint64_t m_subsample_rate = 0;
        size_t m_sampling_counter = 0;
        std::string m_topic_name;
        std::string m_output_dir;
        uint64_t m_start_sample = 0;
        uint64_t m_last_observed_sample_id = 0;
};
 
}
 
#include <rtctk/telemetryRecorder/detail/recordingThreadDetail.hpp>
 
#endif //RTCTK_REUSABLECOMPONENT_TELEMETRYRECORDER_RECORDINGTHREAD_HPP