BackendClickHouse.h

/*
 * Copyright (c) 2020 Pavel Vainerman.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as
 * published by the Free Software Foundation, version 2.1.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 * Lesser General Lesser Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */
// -----------------------------------------------------------------------------
#ifndef _BackendClickHouse_H_
#define _BackendClickHouse_H_
// -----------------------------------------------------------------------------
#include <memory>
#include <unordered_map>
#include <chrono>
#include "UObject_SK.h"
#include "SMInterface.h"
#include "SharedMemory.h"
#include "extensions/Extensions.h"
#include "ClickHouseInterface.h"
#include "ClickHouseTagsConfig.h"
#include "USingleProcess.h"
// --------------------------------------------------------------------------
namespace uniset
{
    // -----------------------------------------------------------------------------
    //------------------------------------------------------------------------------------------
    class BackendClickHouse:
        private USingleProcess,
        public UObject_SK
    {
        public:
            BackendClickHouse( uniset::ObjectId objId, xmlNode* cnode, uniset::ObjectId shmID, const std::shared_ptr<SharedMemory>& ic = nullptr,
                               const std::string& prefix = "clickhouse" );
            virtual ~BackendClickHouse();
 
            static std::shared_ptr<BackendClickHouse> init_clickhouse( int argc, const char* const* argv,
                    uniset::ObjectId shmID, const std::shared_ptr<SharedMemory>& ic = nullptr,
                    const std::string& prefix = "clickhouse" );
 
            static void help_print( int argc, const char* const* argv );
 
            inline std::shared_ptr<LogAgregator> getLogAggregator()
            {
                return loga;
            }
            inline std::shared_ptr<DebugStream> log()
            {
                return mylog;
            }
 
            enum Timers
            {
                tmFlushBuffer,
                tmReconnect,
                tmLastNumberOfTimer
            };
 
        protected:
            BackendClickHouse();
 
            // переопределяем callback, чтобы оптимизировать
            // обработку большого количества сообщений
            // и убрать не нужную в данном процессе обработку (включая sleep_msec)
            virtual void callback() noexcept override;
 
            virtual void askSensors( UniversalIO::UIOCommand cmd ) override;
            virtual void sensorInfo( const uniset::SensorMessage* sm ) override;
            virtual void timerInfo( const uniset::TimerMessage* tm ) override;
            virtual void sysCommand( const uniset::SystemMessage* sm ) override;
            virtual std::string getMonitInfo() const override;
 
            void init( xmlNode* cnode );
            bool flushBuffer();
            bool reconnect();
 
            std::shared_ptr<SMInterface> shm;
 
            using Tag = std::pair<std::string, std::string>;
            using TagList = std::vector<Tag>;
 
            struct ParamInfo
            {
                const std::string name;
                TagList tags;
 
                ParamInfo( const std::string& _name, const TagList& _tags ):
                    name(_name), tags(_tags) {}
            };
 
            std::unordered_map<uniset::ObjectId, ParamInfo> clickhouseParams;
            TagList globalTags;
 
            timeout_t bufSyncTime = { 5000 };
            size_t bufSize = { 5000 };
            size_t bufMaxSize = { 100000 }; // drop messages
            bool timerIsOn = { false };
            timeout_t reconnectTime = { 5000 };
            std::string lastError;
 
            // работа с ClickHouse
            std::shared_ptr<clickhouse::ColumnDateTime64> colTimeStamp;
            std::shared_ptr<clickhouse::ColumnFloat64> colValue;
            std::shared_ptr<clickhouse::ColumnString> colName;
            std::shared_ptr<clickhouse::ColumnString> colNodeName;
            std::shared_ptr<clickhouse::ColumnString> colProducer;
            std::shared_ptr<clickhouse::ColumnArray> arrTagKeys;
            std::shared_ptr<clickhouse::ColumnArray> arrTagValues;
 
            void createColumns();
            void clearData();
            static TagList parseTags( const std::string& tags );
 
            std::unique_ptr<uniset::ClickHouseTagsConfig> dyntags;
 
            std::string fullTableName;
            std::unique_ptr<ClickHouseInterface> db;
 
            std::string dbhost;
            int dbport;
            std::string dbuser;
            std::string dbpass;
            std::string dbname;
 
        private:
            std::string prefix;
            bool connect_ok = { false };
    };
    // --------------------------------------------------------------------------
} // end of namespace uniset
// -----------------------------------------------------------------------------
#endif // _BackendClickHouse_H_
// -----------------------------------------------------------------------------