Small fix after remote test

.vscode/settings.json

@@ -59,6 +59,8 @@
         "stop_token": "cpp",
         "streambuf": "cpp",
         "cinttypes": "cpp",
-        "typeinfo": "cpp"
+        "typeinfo": "cpp",
+        "any": "cpp",
+        "variant": "cpp"
     }
 }

README.md

@@ -12,8 +12,8 @@ A small framework for basic MySQL database operations via MySQL/Connector++
 - Native C++ containers: vector, tuple
 - Response object
 - Thread safe
-- Exceptions
-- Can use external periodic maintenance for connection management
+- Exceptions and log error callback
+- Can use external time loop for connection management
 
 ## Installation
 
@@ -85,13 +85,16 @@ using namespace marcelb::asynco;
 /**
 * Init
 */
-MySQL mydb("tcp://192.168.2.10:3306", "user_nm", "passss", "my_db", 5, periodical_engine::external);
+MySQL mydb("tcp://192.168.2.10:3306", "user_nm", "passss", "my_db", 5, time_loop_type::external);
 
 periodic mysql_maintenance ( [&mydb] () {
-    cout << "IZVRŠAVA SE ENGINE" << endl;
     mydb.periodic_maintenance();
 }, MYSQL_PERIODIC_INTERNAL_TIME);
 
+mydb.set_on_error( [](const string& error) {
+    cout << error << endl;  // print or log
+});
+
 /**
 * You can call multiple queries asynchronously                                     
 */                              

lib/mysql.hpp

@@ -1,13 +1,14 @@
 #ifndef _MYSQL_
 #define _MYSQL_
 
-#include <deque>
+#include <queue>
 #include <mutex>
 #include <thread>
 #include <future>
 #include <string>
 #include <vector>
 #include <tuple>
+#include "ctime"
 
 #include <mysql_driver.h>
 #include <mysql_connection.h>
@@ -38,7 +39,7 @@ namespace mysql {
  * external - expects periodic_maintenance() to be run periodically outside the library
  * 
 */
-enum class periodical_engine {
+enum class time_loop_type {
     internal,
     external
 };
@@ -66,7 +67,7 @@ inline int getValue<int>(ResultSet* res, int column) {
     return res->getInt(column);
 }
 template<> 
-inline uint getValue<uint>(ResultSet* res, int column) {
+inline uint32_t getValue<uint32_t>(ResultSet* res, int column) {
     return res->getUInt(column);
 }
 template<>
@@ -97,34 +98,36 @@ inline bool getValue<bool>(ResultSet* res, int column) {
 
 class MySQL {
     mutex io;
+    condition_variable condition;
     MySQL_Driver *drv;
-    deque<Connection*> con;
-    string path, username, password, db;
-    uint available;
-    uint reconTrys = 3;
-    bool run_engin = true;
-    future<void> periodic_engin;
-    periodical_engine engine_type;
-    
-    /**
-     * Open one database
-    */
-    bool open_one(Connection* con_ptr);
+    queue<Connection*> connection_pool;
+    string path, username, password, database;
+    uint32_t pool_size;
+    bool run_tloop = true;
+    future<void> tloop_future;
+    time_loop_type tloop_type;
+    time_t last_loop_time;
     
     /**
      * Open one database server connection
     */
-    Connection* create_con();
+    Connection* create_connection();
 
     /**
      * Close one database connection
     */
-    bool disconnect_one(Connection* con_ptr);
+    bool disconnect_connection(Connection* connection);
 
     /**
-     * Take an available database connection
+     * Take an pool_size database connection
     */
-    Connection* shift_con();
+    Connection* occupy_connection();
+
+    /**
+     * Free an database connection
+    */
+
+    void release_connection(Connection* connection);
 
     /**
      * Function parses a parameterized row
@@ -135,7 +138,27 @@ class MySQL {
         return make_tuple(getValue<Types>(res, Is + 1)...);
     }
 
-    public:
+    /**
+     * Connect all connections to server
+     */
+    
+    void connect_pool();
+
+    /**
+     * Disconnect all connections to server
+    */
+    void disconnect_pool();
+
+    /**
+     * Internal tloop periodic
+     */
+
+    void _tloop();
+
+public:
+    function<void(const string&)> on_error;
+    uint32_t connect_trys = 3;
+
 
     /**
      * MySQL constructor,
@@ -143,30 +166,19 @@ class MySQL {
      * username, password, database name,
      * and number of active connections (optional)
     */
-    MySQL(const string _path, const string _username, const string _password, const string _db, const uint _available = 1, const periodical_engine _engine_type = periodical_engine::internal);
-
-    /**
-     * Disconnect all connections to server
-    */
-    bool disconnect();
-
-    /**
-     * Define the maximum number of attempts to
-     * reconnect to the server 
-    */
-    void reconnectTrys(const uint _trys);
+    MySQL(const string _path, const string _username, const string _password, const string _db, const uint32_t _available = 1, const time_loop_type _engine_type = time_loop_type::internal);
 
     /**
      * Execute the SQL statement
     */
     template<typename... Types> 
     MySQL_Res<Types...> exec(const string& sql_q) {
-        Connection* con_ptr = shift_con(); 
+        Connection* connection = occupy_connection(); 
         MySQL_Res<Types...> result;
         
         try {
             Statement *stmt;
-            stmt = con_ptr->createStatement();
+            stmt = connection->createStatement();
             result.have_result = stmt->execute(sql_q);
 
             if (result.have_result) {
@@ -192,10 +204,12 @@ class MySQL {
 
             stmt->close();
             delete stmt;
-            disconnect_one(con_ptr);
+            release_connection(connection);
 
         } catch (sql::SQLException& e) {
             throw runtime_error(e.what());
+            // std::cerr << "SQLState: " << e.getSQLState() << std::endl;
+            // std::cerr << "Error code: " << e.getErrorCode() << std::endl;
         }
 
         return result;
@@ -206,7 +220,7 @@ class MySQL {
      * please call this function in it for proper operation at a certain time interval. 
      * You can use the default MYSQL_PERIODIC_INTERNAL_TIME
     */
-    void periodic_maintenance();
+    void tloop();
 
     /**
      * Destruktor

src/mysql.cpp

@@ -1,76 +1,88 @@
 #include "../lib/mysql.hpp"
 
 
-marcelb::mysql::MySQL::MySQL(const string _path, const string _username, const string _password, const string _db, const uint _available, const periodical_engine _engine_type) {
+marcelb::mysql::MySQL::MySQL(const string _path, const string _username, const string _password, const string _db, const uint32_t _available, const time_loop_type _engine_type) {
    path = _path;
    username = _username;
    password = _password;
-   db = _db;
-   available = _available > 0 ? _available : 1;
-   engine_type = _engine_type;
+   database = _db;
+   pool_size = _available > 0 ? _available : 1;
+   tloop_type = _engine_type;
 
    drv = get_mysql_driver_instance();
+   connect_pool();
 
-   if (engine_type == periodical_engine::internal) {
-      periodic_engin = async(launch::async, [&](){
-         while (run_engin) {
+   if (tloop_type == time_loop_type::internal) {
+      tloop_future = async(launch::async, [&](){
+         while (run_tloop) {
             usleep(MYSQL_PERIODIC_INTERNAL_TIME*1000);
-            periodic_maintenance();
+            _tloop();
          }
          return;
       });  
    }
+   // set on initialization to avoid the error
+   last_loop_time = time(nullptr);
 }
 
 
-Connection* marcelb::mysql::MySQL::create_con() {
-   uint trys = 0;
+Connection* marcelb::mysql::MySQL::create_connection() {
+   uint32_t trys = 0;
    bool status = true;
    Connection* new_con = NULL;
 
-   while (reconTrys == unlimited ? status : (trys <= reconTrys && status)) {
+   while (connect_trys == unlimited ? status : (trys <= connect_trys && status)) {
       try {
          Connection* con_can = drv->connect(path, username, password);
+         con_can->setSchema(database);
          status = !con_can->isValid();
          if (!status) {
             new_con = con_can;
          }
          else if (!con_can->isClosed()) {
-            disconnect_one(con_can);
+            disconnect_connection(con_can);
          }
       }
       catch (const SQLException &error) {
-         cout << error.what() << endl;
+         if (on_error) {
+            on_error(error.what() + string(", SQL state: ") + error.getSQLState() + string(", Error code: ") + to_string(error.getErrorCode()));
+         }
          usleep(reconnectSleep);
-         reconTrys == unlimited ? trys : trys++;
+         connect_trys == unlimited ? trys : trys++;
       }         
    }
 
    return new_con;
 }
 
-bool marcelb::mysql::MySQL::disconnect() {
-   io.lock();
-   bool status = true;
-
-   for (uint i=0; i<con.size(); i++) {
-      status = disconnect_one(con[i]) ;
+void marcelb::mysql::MySQL::connect_pool() {
+   lock_guard<mutex> lock(io);
+   for (uint32_t i=0; i<pool_size; i++) {
+      Connection* connection = create_connection();
+      connection_pool.push(connection);
    }
-
-   io.unlock();
-   return status;
 }
 
-bool marcelb::mysql::MySQL::disconnect_one(Connection* con_ptr) {
-   bool status = !con_ptr->isClosed();
+void marcelb::mysql::MySQL::disconnect_pool() {
+   lock_guard<mutex> lock(io);
+   for (uint32_t i=0; i<connection_pool.size(); i++) {
+      Connection* connection = connection_pool.front();
+      connection_pool.pop();
+      disconnect_connection(connection) ;
+   }
+}
 
+bool marcelb::mysql::MySQL::disconnect_connection(Connection* connection) {
+   bool status = !connection->isClosed();
    if (status) {
       try {
-         con_ptr->close();
-         status = !con_ptr->isClosed();
+         connection->close();
+         status = !connection->isClosed();
       } 
       catch (const SQLException &error) {
-         cout << error.what() << endl;
+         if (on_error) {
+            on_error(error.what() + string(", SQL state: ") + error.getSQLState() + string(", Error code: ") + to_string(error.getErrorCode()));
+         }
          status = true;
       }
    }
@@ -79,97 +91,77 @@ bool marcelb::mysql::MySQL::disconnect_one(Connection* con_ptr) {
       status = false; // već je zatvorena
    }
 
-   delete con_ptr;
+   delete connection;
    return status;
 }
 
-bool marcelb::mysql::MySQL::open_one(Connection* con_ptr) {
-   bool status = true; // ako true greška je
-   uint trys = 0;
-
-   while (reconTrys == unlimited ? status : (trys <= reconTrys && status)) {
+void marcelb::mysql::MySQL::_tloop() {
+   if (!run_tloop) {
+      return;
+   }
+   lock_guard<mutex> lock(io);
+   for (size_t i=0; i<connection_pool.size(); i++) {
       try {
-         if (con_ptr->isValid()) {
-            con_ptr->setSchema(db);
-            status = false;
+         Connection *conn = connection_pool.front();
+         connection_pool.pop();
+         if (conn->isValid()) {
+            connection_pool.push(conn);
+         } else {
+            if (!conn->isClosed()){
+                  conn->close();
+            }
+            Connection *n_conn = create_connection();
+            release_connection(n_conn);
          }
-         else {
-            break;
+      } catch (const SQLException &error) {
+         if (on_error) {
+            on_error(error.what() + string(", SQL state: ") + error.getSQLState() + string(", Error code: ") + to_string(error.getErrorCode()));
          }
       }
-      catch (const SQLException &error) {
-         cout << error.what() << endl;
-         usleep(reconnectSleep);
-         reconTrys == unlimited ? trys : trys++;
-      }
    }
 
-   return status;
+   last_loop_time = time(nullptr);
 }
 
-/**
- * Broj pokušaja usljed povezivanja s bazom od 1 do unlimited;
-*/
-
-void marcelb::mysql::MySQL::reconnectTrys(const uint _trys) {
-   io.lock();
-   reconTrys = _trys;
-   io.unlock();
-}
-
-Connection* marcelb::mysql::MySQL::shift_con() {
-   while (true) {
-      while(con.size()) {
-         io.lock();
-         Connection* con_ptr = con[0];
-         con.pop_front();
-         if (con_ptr->isValid()) {
-            io.unlock();
-            return con_ptr;
-         } 
-         io.unlock();
+Connection* marcelb::mysql::MySQL::occupy_connection() {
+   if (last_loop_time + (MYSQL_PERIODIC_INTERNAL_TIME*3/1000) < time(nullptr)) {
+      if (on_error) {
+         on_error("The time loop is not executing properly");
       }
-      usleep(1000);
    }
+   unique_lock<mutex> lock(io);
+   while (connection_pool.empty()) {
+      condition.wait(lock);
+   }
+   Connection *connection = connection_pool.front();
+   connection_pool.pop();
+   return connection;
+}
+
+void marcelb::mysql::MySQL::release_connection(Connection* connection) {
+   lock_guard<std::mutex> lock(io);
+   connection_pool.push(connection);
+   condition.notify_one();
 }
 
 marcelb::mysql::MySQL::~MySQL() {
-   if (engine_type == periodical_engine::internal) {
-      run_engin = false;
-      periodic_engin.get();
+   if (tloop_type == time_loop_type::internal) {
+      run_tloop = false;
+      tloop_future.get();
    } else {
-
+      run_tloop = false;
    }
    
-   disconnect();
+   disconnect_pool();
 }
 
 
-void marcelb::mysql::MySQL::periodic_maintenance() {
-   while (available>con.size() && run_engin) {
-      try {
-         Connection* new_con_ptr = create_con();
-         if (!db.empty()) {
-            if (open_one(new_con_ptr)) {
-               throw string("[ERROR] Unable to open database " + db);
-            }
-         }
-         io.lock();
-         con.push_back(new_con_ptr);       
-         io.unlock();
-      } catch (const SQLException except) {
-         cout << except.what() << endl;
-      } catch (const string except) {
-         cout << except << endl;
-      }     
-   }
-
-   for (int i=0; i<con.size() && run_engin; i++) {
-      if (!con[i]->isValid()) {
-         io.lock();
-         con.erase(con.begin()+i);
-         io.unlock();
-         i--;
+void marcelb::mysql::MySQL::tloop() {
+   if (tloop_type == time_loop_type::internal) {
+      if (on_error) {
+         on_error("Can't start external call tloop, internal is active!");
       }
+      return;
    }
+   _tloop();
 }

test/test.cpp

@@ -13,18 +13,25 @@ using namespace marcelb::asynco;
 
 int main() {
     try {
-        MySQL mydb("tcp://192.168.2.10:3306", "dinio", "H€r5elfInd1aH@nds", "dinio", 10, periodical_engine::external);
+        // MySQL mydb("tcp://192.168.2.10:3306", "dinio", "H€r5elfInd1aH@nds", "dinio", 5, time_loop_type::internal);
+        MySQL mydb("tcp://bitelex.ddns.net:3306", "dinio", "H€r5elfInd1aH@nds", "dinio", 5, time_loop_type::external);
         // MySQL mydb("tcp://bitelex.ddns.net:3306", "dinio", "H€r5elfInd1aH@nds", "dinio", 5);
 
-        periodic mysql_maintenance ( [&mydb] () {
-            cout << "IZVRŠAVA SE ENGINE" << endl;
-            mydb.periodic_maintenance();
+        mydb.on_error = [](const string& error) {
+            cout << error << endl;
+        };
+
+        periodic mysql_tloop ( [&mydb] () {
+            cout << "loop---------------------------" << endl;
+            mydb.tloop();
         }, MYSQL_PERIODIC_INTERNAL_TIME);
 
+while (true) {
         sleep(5);
+
         auto start = high_resolution_clock::now();
 
-        auto a1 = atask ( [&mydb] () {
+        auto a1 = nonsync ( [&mydb] () {
             try {
                 auto response =  mydb.exec<int,string>("SELECT id,domain FROM records WHERE enabled = 1;");
                 cout << response.affected << " " << response.have_result << endl;
@@ -38,12 +45,12 @@ int main() {
                     cout << column_name << endl;
                 }
 
-            } catch (const string err) {
-                cout << err << endl;
+            } catch (const SQLException error) {
+                cout << error.what() << endl;
             }
         });
 
-        auto a2 = atask ( [&mydb] () {
+        auto a2 = nonsync ( [&mydb] () {
             try {
                 auto response =  mydb.exec<string,string>("SELECT zonename,auth_key FROM zones;");
                 cout << response.affected << " " << response.have_result << endl;
@@ -57,12 +64,12 @@ int main() {
                     cout << column_name << endl;
                 }
 
-            } catch (const string err) {
-                cout << err << endl;
+            } catch (const SQLException error) {
+                cout << error.what() << endl;
             }
         });
 
-        auto a3 = atask ( [&mydb] () {
+        auto a3 = nonsync ( [&mydb] () {
             try {
                 auto response =  mydb.exec<string,string>("SELECT username,email FROM users WHERE enabled = 1;");
                 cout << response.affected << " " << response.have_result << endl;
@@ -76,61 +83,83 @@ int main() {
                     cout << column_name << endl;
                 }
 
-            } catch (const string err) {
-                cout << err << endl;
+            } catch (const SQLException error) {
+                cout << error.what() << endl;
+            }
+        });
+
+        auto a4 = nonsync ( [&mydb] () {
+            try {
+                auto response =  mydb.exec<int,string>("SELECT id,domain FROM records WHERE enabled = 1;");
+                cout << response.affected << " " << response.have_result << endl;
+                cout << response.rows << " " << response.columns << endl;
+
+                for (auto row : response) {
+                    cout << get<0>(row) << " " << get<1>(row) << endl;
+                }
+
+                for (auto column_name : response.columns_name) {
+                    cout << column_name << endl;
+                }
+
+            } catch (const SQLException error) {
+                cout << error.what() << endl;
+            }
+        });
+
+        auto a5 = nonsync ( [&mydb] () {
+            try {
+                auto response =  mydb.exec<string,string>("SELECT zonename,auth_key FROM zones;");
+                cout << response.affected << " " << response.have_result << endl;
+                cout << response.rows << " " << response.columns << endl;
+
+                for (auto row : response) {
+                    cout << get<0>(row) << " " << get<1>(row) << endl;
+                }
+
+                for (auto column_name : response.columns_name) {
+                    cout << column_name << endl;
+                }
+
+            } catch (const SQLException error) {
+                cout << error.what() << endl;
+            }
+        });
+
+        auto a6 = nonsync ( [&mydb] () {
+            try {
+                auto response =  mydb.exec<string,string>("SELECT username,email FROM users WHERE enabled = 1;");
+                cout << response.affected << " " << response.have_result << endl;
+                cout << response.rows << " " << response.columns << endl;
+
+                for (auto row : response) {
+                    cout << get<0>(row) << " " << get<1>(row) << endl;
+                }
+
+                for (auto column_name : response.columns_name) {
+                    cout << column_name << endl;
+                }
+
+            } catch (const SQLException error) {
+                cout << error.what() << endl;
             }
         });
 
         wait(a1);
         wait(a2);
         wait(a3);
+        wait(a4);
+        wait(a5);
+        wait(a6);
 
-//        one by one
-        // try {
-        //     auto response =  mydb.exec<int,string>("SELECT id,domain FROM records WHERE enabled = 1;");
-        //     // auto response =  mydb.exec<int,string>("UPDATE records SET enabled = 1;");
-        //     cout << response.affected << " " << response.have_result << endl;
-        //     cout << response.rows << " " << response.columns << endl;
-
-        //     for (auto row : response) {
-        //         cout << get<0>(row) << " " << get<1>(row) << endl;
-        //     }
-
-        //     for (auto column_name : response.columns_name) {
-        //         cout << column_name << endl;
-        //     }
-
-        // } catch (const string err) {
-        //     cout << err << endl;
-        // }
-
-        // auto a1 = atask ( [&mydb] () {
-        //     try {
-        //         auto response =  mydb.exec<string,string>("SELECT username,email FROM records WHERE enabled = 1;");
-        //         cout << response.affected << " " << response.have_result << endl;
-        //         cout << response.rows << " " << response.columns << endl;
-
-        //         for (auto row : response) {
-        //             cout << get<0>(row) << " " << get<1>(row) << endl;
-        //         }
-
-        //         for (auto column_name : response.columns_name) {
-        //             cout << column_name << endl;
-        //         }
-
-        //     } catch (const string err) {
-        //         cout << err << endl;
-        //     }
-        // });
-
-        // wait(a1);
 
         auto end = high_resolution_clock::now();
         auto duration = duration_cast<microseconds>(end - start);
         cout << "-------------Izvršilo se za: " << (double)(duration.count() / 1000.0) << " ms"<< endl;
         
-        sleep(100);
         
+}       
+        sleep(100);
 
     } catch (const SQLException error) {
         cout << error.what() << endl;