tsa_pg_database.h

/*  ===================================================================
*
*                 T R A D I N G   S Y S T E M   A P I ™
*    Copyright © 1999 - 2014 by Peter Ritter ( TradingSystemAPI.com )
*                A L L   R I G H T S   R E S E R V E D
*
*     Consult your license regarding permissions and restrictions.
*             You may obtain a copy of the License at:
*     http://www.TradingSystemAPI.com/licenses/LICENSE-TSAPI-3.0.html
*
*  ====================================================================
*/

#ifndef TSA_POSTGRES_CONN__INCLUDED
#define TSA_POSTGRES_CONN__INCLUDED

#include <string>
#include <queue>
#include <vector>
#include <thread>

#include "tsa.h"

#include "libpq-fe.h"
#include "pqxx/pqxx"


namespace tsa {

    
    namespace postgres {

        struct pg_type {
            static const pqxx::oid int_8byte = 20;    //INT8OID
            static const pqxx::oid int_4byte = 23;    //INT4OID
            static const pqxx::oid float_8byte = 701; //FLOAT8OID
            static const pqxx::oid float_4byte = 700; //FLOAT4OID
            static const pqxx::oid boolean = 16;      //BOOLOID
            static const pqxx::oid datetime = 1114;   //TIMESTAMPOID
            static const pqxx::oid date = 1082;       //DATEOID
            static const pqxx::oid varchar = 1043;    //VARCHAROID

            static pqxx::oid to_equivalent_OID(tsa::type_t);
        };

        /*
        ** ====================================================
        ** >>>        class postgres::database              <<<
        ** ====================================================
        */

        class dll_export database : public tsa::data_base {
            tsa_declare_testable;
          private:
            struct record_to_write {
                std::string table_name;
                std::vector<std::string> field_names;
                std::vector<tsa::variant> values;
            };
            std::queue<record_to_write> m_records_to_write;
          public:
            bool m_is_running = false;
            bool m_stop_now = false;
            bool m_verbose = false;
            void thread_fn__write_records_in_queue();
            std::thread m_write_rec_thread;
            bool m_thread_fn__once_only = false;
          protected:
            mutable std::mutex m_mutex;
            pqxx::connection* m_db_connection_ptr=nullptr;
            mutable std::mutex m_PG_mutex;
            PGconn* m_PGconn_ptr = nullptr;
            std::string m_conn_string;
            void connect_PGconn(const std::string&);
          public:
            database(void);
            ~database(void);
            void connect(const std::string&);
          protected:
            void verify_connected(void)const;
          private:
            std::string m_connection_string;
          public:
            void connection_string(const std::string& cs) 
            { m_connection_string = cs; }
            bool is_running(void)const;
            bool is_connected(void) const;
            virtual void start_threads(void);
            virtual void stop_threads(void);
          public: //Writing records
            void insert_record__async(const std::string& table_name,
                                      const std::vector<tsa::variant>& values); 
            void insert_record__async(const std::string& table_name,
                                      const std::vector<std::string>& field_names,
                                      const std::vector<tsa::variant>& values);
            // Writes record to database. Any exceptions must be caught at a higher level.
            void insert_record(
                const std::string& table_name,
                const std::vector<tsa::variant>& values);
            void insert_record(
                const std::string& table_name,
                const std::vector<std::string>& field_names,
                const std::vector<tsa::variant>& values);

            // Writes record to database. Any errors are written to cerr
            void insert_record__no_exception(
                const std::string& table_name,
                const std::vector<tsa::variant>& values);
            // Writes a timeseries record (with timestamp)
            // Timestamp must be in first field
            void append_series_record(
                const std::string& table_name,
                const std::vector<tsa::variant>& values);

            void update_record(
                const std::string& table_name, 
                const std::string& where_clause,
                //const std::string& primary_key_value,
                const std::vector<std::string>& fields_to_update,
                const std::vector<tsa::variant>& new_values);
          public:
            //table_info_t table_info(const std::string& table_name);
          public:
            pqxx::result execute__no_lock(const std::string& cmd) const;
            pqxx::result execute(const std::string& cmd) const;
            tsa::variant execute_scalar(const std::string& cmd) const;
          public:
            bool drop_table(const std::string& schema, const std::string& table_name);
            bool drop_table(const std::string& table_name);
            virtual bool table_record_count_is_known(void)const override;
            int64_t table_record_count(const std::string& table_name);
            virtual bool table_exists(const std::string& table_name)const override; //tsa::data_base
            bool table_exists(const std::string& schema_name,
                              const std::string& table_name, bool lock_mutex = true)const;
            bool create_table(const std::string& table_name, const std::string& fields_info);
          private:
            bool table_exists__no_lock(const std::string& table_name);
          public:

            void create_series_table(
                const std::string& table_name, const tsa::column_defs& table_def);  //TS

            virtual void series_append(const std::string& table_name, const tsa::mem_table& data);
            void series_insert(const std::string& table_name,  const std::string& data_buff);

            virtual std::string type(void)const override;
            virtual tsa::column_defs series_table_column_defs(const std::string& table_name)override;

            virtual void series_load(const std::string& table_name, tsa::mem_table& data,
                                     const date_time& from_timestamp, size_t num_records)override;
            //virtual void series_append(const std::string& table_name, tsa::mem_table& data)override;

            virtual date_time series_table_first_timestamp(const std::string& table_name)override;
            virtual date_time series_table_last_timestamp(const std::string& table_name)override;

            virtual bool table_has_data(const std::string& table_name)override;
            virtual date_time series_table_closest_timestamp_LOE(const date_time& ts,
                    const std::string& table_name)override;
            virtual void truncate_series_table(const std::string& table_name, 
                const date_time& new_max_timestamp = date_time::min)override;
            virtual bool drop_series_table(const std::string& table_name)override;
        public:
            void series_load(const std::string& table_name, tsa::mem_table& data);
            void clear_table(const std::string& table_name);
          public:
            virtual out_stream_adaptor* make_output_stream_adaptor(void) override;
            virtual in_stream_adaptor* database::make_input_stream_adaptor(const std::string& stream)override;
        };

        /*
        ** =========================================================
        ** > > >    class postgres_writer_out_stream_adaptor   < < <
        ** =========================================================
        */

        //Internal
        class postgres_writer_out_stream_adaptor
            : public tsa::out_stream_adaptor, public tsa::object {
            tsa_declare_testable;
          private:
            tsa::mem_table m_mt;
            postgres::database* m_db_ptr = nullptr;
            std::string m_table_name;
            std::string m_buff;
            size_t m_flush_target_rec_count = (size_t)5e4;
            size_t m_rec_count = 0;
            size_t m_flush_size = (size_t)1e6;
            tsa::date_time m_most_recent_timestamp;
          public:
            virtual ~postgres_writer_out_stream_adaptor(void) { ; }
            postgres_writer_out_stream_adaptor(postgres::database*);
            virtual void osa__append(const date_time& time_stamp,
                                     std::vector<variant>& record) override;
            virtual void osa__prepare_stream(const std::string& target_name,
                const column_defs& field_list/*, rec_stream_open_flag open_flag*/) override;
            virtual void cmp__finalize(void) override;
            void reset_buffer(size_t num_records);
        };

    }//postgres

    
}//tsa


#endif