output.h

Go to the documentation of this file.

/*****************************************
 * This file is documented for Doxygen.
 * If you modify this file please update
 * the comments so that Doxygen will still
 * be able to work.
 ****************************************/
#ifndef OUTPUT_H
#define OUTPUT_H

#include <json.h>
#include <threadbaseclass.h>
#include <outputinterface.h>
#include <associatorinterface.h>
#include <cache.h>
#include <queue.h>
#include <threadpool.h>

#include <thread>
#include <mutex>
#include <future>
#include <string>
#include <sstream>
#include <iostream>
#include <fstream>
#include <vector>
#include <memory>

namespace glass3 {

namespace output {
class output : public glass3::util::iOutput,
        public glass3::util::ThreadBaseClass {
 public:
    output();

    ~output();

    bool setup(std::shared_ptr<const json::Object> config) override;

    void clear() override;

    void sendToOutput(std::shared_ptr<json::Object> message) override;

    bool start() override;

    bool healthCheck() override;

    void setSiteListRequestInterval(int delay);

    int getSiteListRequestInterval();

    void setReportInterval(int interval);

    int getReportInterval();

    void setAssociator(glass3::util::iAssociator* associator);

    glass3::util::iAssociator* getAssociator();

    void setPubOnExpiration(bool pub);

    bool getPubOnExpiration();

    void setImmediatePubThreshold(double threshold);

    double getImmediatePubThreshold();

    std::vector<int> getPubTimes();

    void addPubTime(int pubTime);

    void clearPubTimes();

    bool addTrackingData(std::shared_ptr<json::Object> data);

    std::shared_ptr<const json::Object> getTrackingData(std::string id);

    std::shared_ptr<const json::Object> getNextTrackingData();

    bool haveTrackingData(std::shared_ptr<json::Object> data);

    bool haveTrackingData(std::string ID);

    bool removeTrackingData(std::shared_ptr<const json::Object> data);

    bool removeTrackingData(std::string ID);

    void clearTrackingData();

    bool isDataReady(std::shared_ptr<const json::Object> data);

    bool isDataChanged(std::shared_ptr<const json::Object> data);

    bool isDataPublished(std::shared_ptr<const json::Object> data,
                            bool ignoreVersion = true);
    bool isDataFinished(std::shared_ptr<const json::Object> data);

    void writeOutput(std::shared_ptr<json::Object> data);

    glass3::util::WorkState work() override;

    // constants
    static constexpr int k_iMinimumPublicationTime = 10;

 protected:
    void checkEventsLoop();

    virtual void sendOutput(const std::string &type, const std::string &id,
                            const std::string &message) = 0;

    virtual void sendHeartbeat();

 private:
    std::vector<int> m_PublicationTimes;

    std::atomic<bool> m_bPubOnExpiration;

    std::atomic<int> m_iReportInterval;

    std::atomic<double> m_dImmediatePubThreshold;

    glass3::util::iAssociator* m_Associator;

    std::atomic<int> m_iSiteListRequestInterval;

    glass3::util::Cache * m_TrackingCache;

    std::mutex m_TrackingCacheMutex;

    glass3::util::Queue* m_OutputQueue;

    glass3::util::Queue* m_LookupQueue;

    int m_iMessageCounter;

    int m_iEventCounter;

    int m_iCancelCounter;

    int m_iExpireCounter;

    int m_iHypoCounter;

    int m_iLookupCounter;

    int m_iSiteListCounter;

    std::time_t tLastWorkReport;

    std::time_t m_tLastSiteRequest;

    glass3::util::ThreadPool *m_ThreadPool;

 private:
    std::mutex & getMutex();

    std::mutex m_Mutex;
};
}  // namespace output
}  // namespace glass3
#endif  // OUTPUT_H