DBServer_PostgreSQL.h

/*
 * Copyright (c) 2015 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 DBServer_PostgreSQL_H_
#define DBServer_PostgreSQL_H_
// --------------------------------------------------------------------------
#include <string_view>
#include <unordered_map>
#include <queue>
#include "UniSetTypes.h"
#include "PostgreSQLInterface.h"
#include "DBServer.h"
#include "SharedMemory.h"
#include "USingleProcess.h"
// -------------------------------------------------------------------------
namespace uniset
{
    //------------------------------------------------------------------------------------------
    class DBServer_PostgreSQL:
        private USingleProcess,
        public DBServer
    {
        public:
            DBServer_PostgreSQL( uniset::ObjectId id, xmlNode* cnode, const std::string& prefix );
            DBServer_PostgreSQL();
            virtual ~DBServer_PostgreSQL();
 
            static std::shared_ptr<DBServer_PostgreSQL> init_dbserver( int argc, const char* const* argv, const std::shared_ptr<uniset::SharedMemory>& ic = nullptr, const std::string& prefix = "pgsql" );
 
            static void help_print( int argc, const char* const* argv );
 
            inline std::shared_ptr<LogAgregator> logAggregator()
            {
                return loga;
            }
            inline std::shared_ptr<DebugStream> log()
            {
                return dblog;
            }
 
            bool isConnectOk() const;
 
            static constexpr std::string_view tblcols = { "date,time,time_usec,sensor_id,value,node" };
 
        protected:
            typedef std::unordered_map<int, std::string> DBTableMap;
 
            virtual void initDBServer() override;
            virtual void onReconnect( std::unique_ptr<PostgreSQLInterface>& db ) {};
 
            virtual void timerInfo( const uniset::TimerMessage* tm ) override;
            virtual void sysCommand( const uniset::SystemMessage* sm ) override;
            virtual void sensorInfo( const uniset::SensorMessage* sm ) override;
            virtual void confirmInfo( const uniset::ConfirmMessage* cmsg ) override;
            virtual void onTextMessage( const uniset::TextMessage* msg ) override;
            virtual bool deactivateObject() override;
            virtual std::string getMonitInfo( const std::string& params ) override;
 
            bool writeToBase( const std::string& query );
 
            inline std::string tblName(int key)
            {
                return tblMap[key];
            }
 
            enum Timers
            {
                PingTimer,        
                ReconnectTimer,   
                FlushInsertBuffer, 
                lastNumberOfTimer
            };
 
            std::unique_ptr<PostgreSQLInterface> db;
            typedef std::queue<std::string> QueryBuffer;
 
            void flushBuffer();
 
            // writeBuffer
 
            typedef std::vector<PostgreSQLInterface::Record> InsertBuffer;
            void flushInsertBuffer();
            virtual void addRecord( const PostgreSQLInterface::Record&& rec );
            virtual bool writeInsertBufferToDB( const std::string& table
                                                , std::string_view colname
                                                , const InsertBuffer& ibuf );
 
        private:
            DBTableMap tblMap;
 
            int PingTime = { 15000 };
            int ReconnectTime = { 30000 };
 
            bool connect_ok = { false }; 
            QueryBuffer qbuf;
            size_t qbufSize = { 200 }; // размер буфера сообщений.
            bool lastRemove = { false };
            std::mutex mqbuf;
 
            InsertBuffer ibuf;
            size_t ibufSize = { 0 };
            size_t ibufMaxSize = { 2000 };
            timeout_t ibufSyncTimeout = { 15000 };
            float ibufOverflowCleanFactor = { 0.5 }; // коэффициент {0...1} чистки буфера при переполнении
    };
    // ----------------------------------------------------------------------------------
} // end of namespace uniset
//------------------------------------------------------------------------------------------
#endif