Compare commits

..

No commits in common. "dev" and "v0.8" have entirely different histories.
dev ... v0.8

5 changed files with 193 additions and 233 deletions

View File

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

View File

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

View File

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

View File

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

View File

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