From 16519fab3d99ca277bce00d2bcbfd46b33a46391 Mon Sep 17 00:00:00 2001
From: znone <glyc@sina.com.cn>
Date: Sun, 16 Jun 2019 13:27:41 +0000
Subject: [PATCH] 通过 MariaDB 的非阻塞函数访问 MySQL。
---
include/qtl_mysql.hpp | 1622 ++++++++++++++++++++++++++++++++++++++++++++-------------
1 files changed, 1,234 insertions(+), 388 deletions(-)
diff --git a/include/qtl_mysql.hpp b/include/qtl_mysql.hpp
index ffa55a4..b532588 100644
--- a/include/qtl_mysql.hpp
+++ b/include/qtl_mysql.hpp
@@ -1,5 +1,5 @@
-#ifndef _MYDTL_MYSQL_H_
-#define _MYDTL_MYSQL_H_
+#ifndef _QTL_MYSQL_H_
+#define _QTL_MYSQL_H_
#include <mysql.h>
#include <errmsg.h>
@@ -12,7 +12,17 @@
#include <array>
#include <functional>
#include <algorithm>
+#include <system_error>
#include "qtl_common.hpp"
+#include "qtl_async.hpp"
+
+#if LIBMYSQL_VERSION_ID >=80000
+typedef bool my_bool;
+#endif //MySQL 8
+
+#ifdef MARIADB_VERSION_ID
+#define IS_LONGDATA(t) ((t) >= MYSQL_TYPE_TINY_BLOB && (t) <= MYSQL_TYPE_STRING)
+#endif //MariaDB
namespace qtl
{
@@ -20,13 +30,9 @@
namespace mysql
{
-#if LIBMYSQL_VERSION_ID >=80000
-typedef bool my_bool;
-#endif //MySQL 8
-
struct init
{
- init(int argc=-1, char **argv=NULL, char **groups=NULL)
+ init(int argc=-1, char **argv=nullptr, char **groups=nullptr)
{
//my_init();
mysql_library_init(argc, argv, groups);
@@ -192,19 +198,20 @@
binder.bind(const_cast<char*>(str), static_cast<unsigned long>(length), MYSQL_TYPE_VAR_STRING);
}
-class statement;
-class database;
+class base_statement;
+class basic_database;
class error : public std::exception
{
public:
error() : m_error(0) { }
error(unsigned int err, const char* errmsg) : m_error(err), m_errmsg(errmsg) { }
- explicit error(statement& stmt);
- explicit error(database& db);
+ explicit error(const base_statement& stmt);
+ explicit error(const basic_database& db);
error(const error& src) = default;
virtual ~error() throw() { }
int code() const throw() { return m_error; }
+ operator bool() const { return m_error != 0; }
virtual const char* what() const throw() override { return m_errmsg.data(); }
private:
unsigned int m_error;
@@ -401,8 +408,6 @@
}
}
-
-
private:
MYSQL_STMT* m_stmt;
binder m_binder;
@@ -443,39 +448,328 @@
typedef std::function<void(std::ostream&)> blob_writer;
-class statement final
+class base_statement
{
-public:
- statement() : m_stmt(NULL), m_result(NULL) {}
- explicit statement(database& db);
- statement(const statement&) = delete;
- statement(statement&& src)
+protected:
+ base_statement() : m_stmt(nullptr), m_result(nullptr) {}
+ base_statement(const base_statement&) = delete;
+ explicit base_statement(basic_database& db);
+ base_statement(base_statement&& src)
: m_stmt(src.m_stmt), m_result(src.m_result),
m_binders(std::move(src.m_binders)), m_binderAddins(std::move(src.m_binderAddins))
{
- src.m_stmt=NULL;
- src.m_result=NULL;
+ src.m_stmt=nullptr;
+ src.m_result=nullptr;
}
- statement& operator=(const statement&) = delete;
- statement& operator=(statement&& src)
+ base_statement& operator=(const base_statement&) = delete;
+ base_statement& operator=(base_statement&& src)
{
if(this!=&src)
{
m_stmt=src.m_stmt;
m_result=src.m_result;
- src.m_stmt=NULL;
- src.m_result=NULL;
+ src.m_stmt=nullptr;
+ src.m_result=nullptr;
m_binders=std::move(src.m_binders);
m_binderAddins=std::move(src.m_binderAddins);
}
+ return *this;
+ }
+
+public:
+ operator MYSQL_STMT*() { return m_stmt; }
+
+ unsigned int get_parameter_count() const { return mysql_stmt_param_count(m_stmt); }
+ unsigned int get_column_count() const { return mysql_stmt_field_count(m_stmt); }
+
+ unsigned long length(unsigned int index) const
+ {
+ return m_binderAddins[index].m_length;
+ }
+
+ bool is_null(unsigned int index) const
+ {
+ return m_binderAddins[index].m_isNull!=0;
+ }
+
+ size_t find_field(const char* name) const
+ {
+ if(m_result)
+ {
+ for(size_t i=0; i!=m_result->field_count; i++)
+ {
+ if(strncmp(m_result->fields[i].name, name, m_result->fields[i].name_length)==0)
+ return i;
+ }
+ }
+ return -1;
+ }
+
+ my_ulonglong affetced_rows()
+ {
+ return mysql_stmt_affected_rows(m_stmt);
+ }
+ my_ulonglong insert_id()
+ {
+ return mysql_stmt_insert_id(m_stmt);
+ }
+
+ binder* get_binder(unsigned long index)
+ {
+ return &m_binders[index];
+ }
+
+ unsigned int error() const
+ {
+ return mysql_stmt_errno(m_stmt);
+ }
+ const char* errmsg() const
+ {
+ return mysql_stmt_error(m_stmt);
+ }
+
+ MYSQL_RES* result() { return m_result; }
+
+ void bind_param(size_t index, const char* param, size_t length)
+ {
+ bind(m_binders[index], param, length);
+ }
+
+ void bind_param(size_t index, const std::nullptr_t&)
+ {
+ m_binders[index].bind();
+ }
+
+ void bind_param(size_t index, std::istream& param)
+ {
+ m_binders[index].bind(nullptr, 0, MYSQL_TYPE_LONG_BLOB);
+ m_binderAddins[index].m_after_fetch = [this, index, ¶m](const binder&) {
+ std::array<char, blob_buffer_size> buffer;
+ unsigned long readed = 0;
+ while (!param.eof() && !param.fail())
+ {
+ param.read(buffer.data(), buffer.size());
+ readed = (unsigned long)param.gcount();
+ if (readed > 0)
+ {
+ if (mysql_stmt_send_long_data(m_stmt, index, buffer.data(), readed) != 0)
+ throw_exception();
+ }
+ }
+ };
+ }
+
+ void bind_param(size_t index, const blob_writer& param)
+ {
+ m_binders[index].bind(nullptr, 0, MYSQL_TYPE_LONG_BLOB);
+ m_binderAddins[index].m_after_fetch = [this, index, ¶m](const binder& b) {
+ blobbuf buf;
+ buf.open(m_stmt, index, b, std::ios::out);
+ std::ostream s(&buf);
+ param(s);
+ };
+ }
+
+ template<class Param>
+ void bind_param(size_t index, const Param& param)
+ {
+ bind(m_binders[index], param);
+ }
+
+ template<class Type>
+ void bind_field(size_t index, Type&& value)
+ {
+ if (m_result)
+ {
+ bind(m_binders[index], std::forward<Type>(value));
+ m_binderAddins[index].m_after_fetch = if_null<typename std::remove_reference<Type>::type>(value);
+ }
+ }
+
+ void bind_field(size_t index, char* value, size_t length)
+ {
+ m_binders[index].bind(value, length - 1, MYSQL_TYPE_VAR_STRING);
+ m_binderAddins[index].m_after_fetch = [](const binder& bind) {
+ if (*bind.is_null)
+ memset(bind.buffer, 0, bind.buffer_length + 1);
+ else
+ {
+ char* text = reinterpret_cast<char*>(bind.buffer);
+ text[*bind.length] = '\0';
+ }
+ };
+ }
+
+ template<size_t N>
+ void bind_field(size_t index, std::array<char, N>&& value)
+ {
+ bind_field(index, value.data(), value.size());
+ }
+
+ template<typename T>
+ void bind_field(size_t index, bind_string_helper<T>&& value)
+ {
+ if (m_result)
+ {
+ MYSQL_FIELD* field = mysql_fetch_field_direct(m_result, (unsigned int)index);
+ if (field == nullptr) throw_exception();
+ value.clear();
+ typename bind_string_helper<T>::char_type* data = value.alloc(field->length);
+ m_binderAddins[index].m_before_fetch = [this, value](binder& b) mutable {
+ if (value.size() < b.buffer_length)
+ {
+ value.alloc(b.buffer_length);
+ if (b.buffer != value.data())
+ {
+ b.buffer = const_cast<char*>(value.data());
+ mysql_stmt_bind_result(m_stmt, &m_binders.front());
+ }
+ }
+ };
+ m_binderAddins[index].m_after_fetch = [value](const binder& b) mutable {
+ if (*b.is_null) value.clear();
+ else value.truncate(*b.length);
+ };
+ m_binders[index].bind((void*)data, field->length, field->type);
+ }
+ }
+
+ void bind_field(size_t index, std::ostream&& value)
+ {
+ if (m_result)
+ {
+ MYSQL_FIELD* field = mysql_fetch_field_direct(m_result, (unsigned int)index);
+ assert(IS_LONGDATA(field->type));
+ m_binders[index].bind(nullptr, 0, field->type);
+ m_binderAddins[index].m_after_fetch = [this, index, &value](const binder& b) {
+ unsigned long readed = 0;
+ std::array<char, blob_buffer_size> buffer;
+ binder& bb = const_cast<binder&>(b);
+ if (*b.is_null) return;
+ bb.buffer = const_cast<char*>(buffer.data());
+ bb.buffer_length = buffer.size();
+ while (readed <= *b.length)
+ {
+ int ret = mysql_stmt_fetch_column(m_stmt, &bb, index, readed);
+ if (ret != 0)
+ throw_exception();
+ value.write(buffer.data(), std::min(b.buffer_length, *b.length - b.offset));
+ readed += bb.buffer_length;
+ }
+ };
+ }
+ }
+
+ void bind_field(size_t index, blobbuf&& value)
+ {
+ if (m_result)
+ {
+ MYSQL_FIELD* field = mysql_fetch_field_direct(m_result, (unsigned int)index);
+ assert(IS_LONGDATA(field->type));
+ m_binders[index].bind(nullptr, 0, field->type);
+ m_binderAddins[index].m_after_fetch = [this, index, &value](const binder& b) {
+ if (*b.is_null) return;
+ value.open(m_stmt, index, b, std::ios::in);
+ };
+ }
+ }
+
+ template<typename Type>
+ void bind_field(size_t index, indicator<Type>&& value)
+ {
+ if (m_result)
+ {
+ qtl::bind_field(*this, index, value.data);
+ binder_addin& addin = m_binderAddins[index];
+ auto fetch_fun = addin.m_after_fetch;
+ addin.m_after_fetch = [&addin, fetch_fun, &value](const binder& b) {
+ value.is_null = *b.is_null != 0;
+ value.length = *b.length;
+ value.is_truncated = addin.is_truncated;
+ if (fetch_fun) fetch_fun(b);
+ };
+ }
+ }
+
+ void close()
+ {
+ if (m_result)
+ {
+ mysql_free_result(m_result);
+ m_result = nullptr;
+ }
+ if (m_stmt)
+ {
+ mysql_stmt_close(m_stmt);
+ m_stmt = nullptr;
+ }
+ }
+
+protected:
+ MYSQL_STMT* m_stmt;
+ MYSQL_RES* m_result;
+ std::vector<binder> m_binders;
+
+ struct binder_addin
+ {
+ unsigned long m_length;
+ my_bool m_isNull;
+ my_bool m_error;
+ bool is_truncated;
+ std::function<void(binder&)> m_before_fetch;
+ std::function<void(const binder&)> m_after_fetch;
+ };
+ std::vector<binder_addin> m_binderAddins;
+
+ void resize_binders(size_t n)
+ {
+ m_binders.resize(n);
+ m_binderAddins.resize(n);
+ }
+ void set_binders()
+ {
+ for(size_t i=0; i!=m_binders.size(); i++)
+ {
+ m_binderAddins[i].m_length=0;
+ m_binders[i].length=&m_binderAddins[i].m_length;
+ m_binderAddins[i].m_isNull=0;
+ m_binders[i].is_null=&m_binderAddins[i].m_isNull;
+ m_binderAddins[i].m_error=0;
+ m_binders[i].error=&m_binderAddins[i].m_error;
+ }
+ }
+
+ void throw_exception() const { throw mysql::error(*this); }
+
+ template<typename Value>
+ struct if_null
+ {
+ if_null(Value& value, Value&& def=Value()) : m_value(value), m_def(std::move(def)) { }
+ void operator()(const binder& b)
+ {
+ if(*b.is_null) m_value=m_def;
+ }
+ Value& m_value;
+ Value m_def;
+ };
+
+};
+
+class statement : public base_statement
+{
+public:
+ statement() = default;
+ explicit statement(basic_database& db) : base_statement(db) { }
+ statement(statement&& src) : base_statement(std::move(src)) { }
+ statement& operator=(statement&& src)
+ {
+ base_statement::operator =(std::move(src));
return *this;
}
~statement()
{
close();
}
-
- operator MYSQL_STMT*() { return m_stmt; }
void open(const char *query_text, unsigned long text_length=0)
{
@@ -515,13 +809,13 @@
template<typename Types>
bool fetch(Types&& values)
{
- if(m_result==NULL)
+ if(m_result==nullptr)
{
unsigned long count=mysql_stmt_field_count(m_stmt);
if(count>0)
{
m_result=mysql_stmt_result_metadata(m_stmt);
- if(m_result==NULL) throw_exception();
+ if(m_result==nullptr) throw_exception();
resize_binders(count);
qtl::bind_record(*this, std::forward<Types>(values));
set_binders();
@@ -530,207 +824,6 @@
}
}
return fetch();
- }
-
- void bind_param(size_t index, const char* param, size_t length)
- {
- bind(m_binders[index], param, length);
- }
-
- void bind_param(size_t index, const std::nullptr_t&)
- {
- m_binders[index].bind();
- }
-
- void bind_param(size_t index, std::istream& param)
- {
- m_binders[index].bind(NULL, 0, MYSQL_TYPE_LONG_BLOB);
- m_binderAddins[index].m_after_fetch=[this, index, ¶m](const binder&) {
- std::array<char, blob_buffer_size> buffer;
- unsigned long readed=0;
- while(!param.eof() && !param.fail())
- {
- param.read(buffer.data(), buffer.size());
- readed=(unsigned long)param.gcount();
- if(readed>0)
- {
- if(mysql_stmt_send_long_data(m_stmt, index, buffer.data(), readed)!=0)
- throw_exception();
- }
- }
- };
- }
-
- void bind_param(size_t index, const blob_writer& param)
- {
- m_binders[index].bind(NULL, 0, MYSQL_TYPE_LONG_BLOB);
- m_binderAddins[index].m_after_fetch = [this, index, ¶m](const binder& b) {
- blobbuf buf;
- buf.open(m_stmt, index, b, std::ios::out);
- std::ostream s(&buf);
- param(s);
- };
- }
-
- template<class Param>
- void bind_param(size_t index, const Param& param)
- {
- bind(m_binders[index], param);
- }
-
- template<class Type>
- void bind_field(size_t index, Type&& value)
- {
- if(m_result)
- {
- bind(m_binders[index], std::forward<Type>(value));
- m_binderAddins[index].m_after_fetch=if_null<typename std::remove_reference<Type>::type>(value);
- }
- }
-
- void bind_field(size_t index, char* value, size_t length)
- {
- m_binders[index].bind(value, length-1, MYSQL_TYPE_VAR_STRING);
- m_binderAddins[index].m_after_fetch=[](const binder& bind) {
- if(*bind.is_null)
- memset(bind.buffer, 0, bind.buffer_length+1);
- else
- {
- char* text=reinterpret_cast<char*>(bind.buffer);
- text[*bind.length]='\0';
- }
- };
- }
-
- template<size_t N>
- void bind_field(size_t index, std::array<char, N>&& value)
- {
- bind_field(index, value.data(), value.size());
- }
-
- template<typename T>
- void bind_field(size_t index, bind_string_helper<T>&& value)
- {
- if(m_result)
- {
- MYSQL_FIELD* field=mysql_fetch_field_direct(m_result, (unsigned int)index);
- if(field==NULL) throw_exception();
- value.clear();
- typename bind_string_helper<T>::char_type* data=value.alloc(field->length);
- m_binderAddins[index].m_before_fetch = [this, value](binder& b) mutable {
- if (value.size() < b.buffer_length)
- {
- value.alloc(b.buffer_length);
- if (b.buffer != value.data())
- {
- b.buffer = const_cast<char*>(value.data());
- mysql_stmt_bind_result(m_stmt, &m_binders.front());
- }
- }
- };
- m_binderAddins[index].m_after_fetch= [value](const binder& b) mutable {
- if(*b.is_null) value.clear();
- else value.truncate(*b.length);
- };
- m_binders[index].bind((void*)data, field->length, field->type);
- }
- }
-
- void bind_field(size_t index, std::ostream&& value)
- {
- if(m_result)
- {
- MYSQL_FIELD* field = mysql_fetch_field_direct(m_result, (unsigned int)index);
- assert(IS_LONGDATA(field->type));
- m_binders[index].bind(NULL, 0, field->type);
- m_binderAddins[index].m_after_fetch=[this, index, &value](const binder& b) {
- unsigned long readed=0;
- std::array<char, blob_buffer_size> buffer;
- binder& bb=const_cast<binder&>(b);
- if(*b.is_null) return;
- bb.buffer=const_cast<char*>(buffer.data());
- bb.buffer_length=buffer.size();
- while(readed<=*b.length)
- {
- int ret=mysql_stmt_fetch_column(m_stmt, &bb, index, readed);
- if(ret!=0)
- throw_exception();
- value.write(buffer.data(), std::min(b.buffer_length, *b.length-b.offset));
- readed+=bb.buffer_length;
- }
- };
- }
- }
-
- void bind_field(size_t index, blobbuf&& value)
- {
- if (m_result)
- {
- MYSQL_FIELD* field = mysql_fetch_field_direct(m_result, (unsigned int)index);
- assert(IS_LONGDATA(field->type));
- m_binders[index].bind(NULL, 0, field->type);
- m_binderAddins[index].m_after_fetch = [this, index, &value](const binder& b) {
- if (*b.is_null) return;
- value.open(m_stmt, index, b, std::ios::in);
- };
- }
- }
-
- template<typename Type>
- void bind_field(size_t index, indicator<Type>&& value)
- {
- if(m_result)
- {
- qtl::bind_field(*this, index, value.data);
- binder_addin& addin=m_binderAddins[index];
- auto fetch_fun=addin.m_after_fetch;
- addin.m_after_fetch=[&addin, fetch_fun, &value](const binder& b) {
- value.is_null= *b.is_null!=0;
- value.length=*b.length;
- value.is_truncated=addin.is_truncated;
- if(fetch_fun) fetch_fun(b);
- };
- }
- }
-
- unsigned int get_parameter_count() const { return mysql_stmt_param_count(m_stmt); }
- unsigned int get_column_count() const { return mysql_stmt_field_count(m_stmt); }
-
- unsigned long length(unsigned int index) const
- {
- return m_binderAddins[index].m_length;
- }
-
- bool is_null(unsigned int index) const
- {
- return m_binderAddins[index].m_isNull!=0;
- }
-
- size_t find_field(const char* name) const
- {
- if(m_result)
- {
- for(size_t i=0; i!=m_result->field_count; i++)
- {
- if(strncmp(m_result->fields[i].name, name, m_result->fields[i].name_length)==0)
- return i;
- }
- }
- return -1;
- }
-
- void close()
- {
- if(m_result)
- {
- mysql_free_result(m_result);
- m_result=NULL;
- }
- if(m_stmt)
- {
- mysql_stmt_close(m_stmt);
- m_stmt=NULL;
- }
}
bool fetch()
@@ -761,7 +854,7 @@
if(m_result)
{
mysql_free_result(m_result);
- m_result=NULL;
+ m_result=nullptr;
mysql_stmt_free_result(m_stmt);
}
int ret=0;
@@ -773,109 +866,114 @@
return ret==0;
}
- my_ulonglong affetced_rows()
- {
- return mysql_stmt_affected_rows(m_stmt);
- }
- my_ulonglong insert_id()
- {
- return mysql_stmt_insert_id(m_stmt);
- }
-
- binder* get_binder(unsigned long index)
- {
- return &m_binders[index];
- }
-
- unsigned int error() const
- {
- return mysql_stmt_errno(m_stmt);
- }
- const char* errmsg() const
- {
- return mysql_stmt_error(m_stmt);
- }
-
- MYSQL_RES* result() { return m_result; }
bool reset() { return mysql_stmt_reset(m_stmt)!=0; }
-
-private:
- MYSQL_STMT* m_stmt;
- MYSQL_RES* m_result;
- std::vector<binder> m_binders;
-
- struct binder_addin
- {
- unsigned long m_length;
- my_bool m_isNull;
- my_bool m_error;
- bool is_truncated;
- std::function<void(binder&)> m_before_fetch;
- std::function<void(const binder&)> m_after_fetch;
- };
- std::vector<binder_addin> m_binderAddins;
-
- void resize_binders(size_t n)
- {
- m_binders.resize(n);
- m_binderAddins.resize(n);
- }
- void set_binders()
- {
- for(size_t i=0; i!=m_binders.size(); i++)
- {
- m_binderAddins[i].m_length=0;
- m_binders[i].length=&m_binderAddins[i].m_length;
- m_binderAddins[i].m_isNull=0;
- m_binders[i].is_null=&m_binderAddins[i].m_isNull;
- m_binderAddins[i].m_error=0;
- m_binders[i].error=&m_binderAddins[i].m_error;
- }
- }
-
- void throw_exception() { throw mysql::error(*this); }
-
-private:
- template<typename Value>
- struct if_null
- {
- if_null(Value& value, Value&& def=Value()) : m_value(value), m_def(std::move(def)) { }
- void operator()(const binder& b)
- {
- if(*b.is_null) m_value=m_def;
- }
- Value& m_value;
- Value m_def;
- };
};
-class database final : public qtl::base_database<database, statement>
+/*
+struct LocalInfile
{
+ int read(char *buf, unsigned int buf_len);
+ void close();
+ int error(char *error_msg, unsigned int error_msg_len);
+};
+*/
+template<typename LocalInfile>
+struct local_infile_factory
+{
+ static int local_infile_init(void **ptr, const char *filename, void *userdata)
+ {
+ LocalInfile* object = nullptr;
+ try
+ {
+ object = new LocalInfile(filename);
+ }
+ catch (...)
+ {
+ return -1;
+ }
+ return ptr ? 0 : -1;
+ }
+
+ static int local_infile_read(void *ptr, char *buf, unsigned int buf_len)
+ {
+ LocalInfile* object = reinterpret_cast<LocalInfile>(ptr);
+ return object->read(buf, buf_len);
+ }
+
+ static void local_infile_end(void *ptr)
+ {
+ LocalInfile* object = reinterpret_cast<LocalInfile>(ptr);
+ object->close();
+ delete object;
+ }
+
+ static int local_infile_error(void *ptr, char *error_msg, unsigned int error_msg_len)
+ {
+ LocalInfile* object = reinterpret_cast<LocalInfile>(ptr);
+ return object->error(error_msg, error_msg_len);
+ }
+};
+
+class local_file
+{
+public:
+ local_file(const char* filename)
+ {
+ m_fp = fopen(filename, "rb");
+ if (m_fp == nullptr)
+ throw std::system_error(std::make_error_code(std::errc(errno)));
+ }
+ int read(char *buf, unsigned int buf_len)
+ {
+ fread(buf, 1, buf_len, m_fp);
+ }
+ void close()
+ {
+ fclose(m_fp);
+ }
+ int error(char *error_msg, unsigned int error_msg_len)
+ {
+ int errcode = errno;
+ memset(error_msg, 0, error_msg_len);
+ strncpy(error_msg, strerror(errcode), error_msg_len - 1);
+ return errcode;
+ }
+
+private:
+ FILE* m_fp;
+};
+
+class basic_database
+{
+protected:
+ basic_database()
+ {
+ m_mysql = mysql_init(nullptr);
+ }
+
public:
typedef mysql::error exception_type;
- database()
+ ~basic_database()
{
- m_mysql=mysql_init(NULL);
+ if(m_mysql)
+ mysql_close(m_mysql);
}
- ~database()
- {
- mysql_close(m_mysql);
- }
- database(const database&) = delete;
- database(database&& src)
+ basic_database(const basic_database&) = delete;
+ basic_database(basic_database&& src)
{
m_mysql=src.m_mysql;
- src.m_mysql=NULL;
+ src.m_mysql=nullptr;
}
- database& operator==(const database&) = delete;
- database& operator==(database&& src)
+ basic_database& operator==(const basic_database&) = delete;
+ basic_database& operator==(basic_database&& src)
{
if(this!=&src)
{
- mysql_close(m_mysql);
+ if(m_mysql)
+ mysql_close(m_mysql);
m_mysql=src.m_mysql;
- src.m_mysql=NULL;
+ src.m_mysql=nullptr;
}
return *this;
}
@@ -901,29 +999,6 @@
return options(MYSQL_OPT_RECONNECT, &re);
}
- bool open(const char *host, const char *user, const char *passwd, const char *db,
- unsigned long clientflag=0, unsigned int port=0, const char *unix_socket=NULL)
- {
- if(m_mysql==NULL) m_mysql=mysql_init(NULL);
- return mysql_real_connect(m_mysql, host, user, passwd, db, port, unix_socket, clientflag)!=NULL;
- }
- void close()
- {
- mysql_close(m_mysql);
- m_mysql=NULL;
- }
- void refresh(unsigned int options)
- {
- if(mysql_refresh(m_mysql, options)<0)
- throw_exception();
- }
-
- void select(const char* db)
- {
- if(mysql_select_db(m_mysql, db)!=0)
- throw_exception();
- }
-
const char* current() const
{
return m_mysql->db;
@@ -936,35 +1011,6 @@
const char* errmsg() const
{
return mysql_error(m_mysql);
- }
-
- statement open_command(const char* query_text, size_t text_length)
- {
- statement stmt(*this);
- stmt.open(query_text, text_length);
- return stmt;
- }
- statement open_command(const char* query_text)
- {
- return open_command(query_text, strlen(query_text));
- }
- statement open_command(const std::string& query_text)
- {
- return open_command(query_text.data(), query_text.length());
- }
-
- void simple_execute(const char* query_text, uint64_t* paffected=NULL)
- {
- if(mysql_query(m_mysql, query_text)!=0)
- throw_exception();
- if(paffected) *paffected=affected_rows();
- }
- void simple_execute(const char* query_text, unsigned long text_length, uint64_t* paffected=NULL)
- {
- if(text_length==0) text_length=(unsigned long)strlen(query_text);
- if(mysql_real_query(m_mysql, query_text, text_length)!=0)
- throw_exception();
- if(paffected) *paffected=affected_rows();
}
uint64_t affected_rows()
@@ -982,9 +1028,78 @@
return mysql_insert_id(m_mysql);
}
+protected:
+ MYSQL* m_mysql;
+ void throw_exception() { throw mysql::error(*this); }
+};
+
+#if MARIADB_VERSION_ID >= 050500
+
+class async_connection;
+
+#endif //MariaDB
+
+class database : public basic_database, public qtl::base_database<database, statement>
+{
+public:
+ database() = default;
+
+ bool open(const char *host, const char *user, const char *passwd, const char *db,
+ unsigned long clientflag = 0, unsigned int port = 0, const char *unix_socket = nullptr)
+ {
+ if (m_mysql == nullptr) m_mysql = mysql_init(nullptr);
+ return mysql_real_connect(m_mysql, host, user, passwd, db, port, unix_socket, clientflag) != nullptr;
+ }
+ void close()
+ {
+ mysql_close(m_mysql);
+ m_mysql = nullptr;
+ }
+
+ statement open_command(const char* query_text, size_t text_length)
+ {
+ statement stmt(*this);
+ stmt.open(query_text, text_length);
+ return stmt;
+ }
+ statement open_command(const char* query_text)
+ {
+ return open_command(query_text, strlen(query_text));
+ }
+ statement open_command(const std::string& query_text)
+ {
+ return open_command(query_text.data(), query_text.length());
+ }
+
+ void refresh(unsigned int options)
+ {
+ if (mysql_refresh(m_mysql, options) < 0)
+ throw_exception();
+ }
+
+ void select(const char* db)
+ {
+ if (mysql_select_db(m_mysql, db) != 0)
+ throw_exception();
+ }
+
+ void simple_execute(const char* query_text, uint64_t* paffected = nullptr)
+ {
+ if (mysql_query(m_mysql, query_text) != 0)
+ throw_exception();
+ if (paffected) *paffected = affected_rows();
+ }
+ void simple_execute(const char* query_text, unsigned long text_length, uint64_t* paffected = nullptr)
+ {
+ if (text_length == 0) text_length = (unsigned long)strlen(query_text);
+ if (mysql_real_query(m_mysql, query_text, text_length) != 0)
+ throw_exception();
+ if (paffected) *paffected = affected_rows();
+ }
+
void auto_commit(bool on)
{
- if(mysql_autocommit(m_mysql, on?1:0)!=0)
+ if (mysql_autocommit(m_mysql, on ? 1 : 0) != 0)
throw_exception();
}
void begin_transaction()
@@ -993,20 +1108,20 @@
}
void rollback()
{
- if(mysql_rollback(m_mysql)!=0)
+ if (mysql_rollback(m_mysql) != 0)
throw_exception();
auto_commit(true);
}
void commit()
{
- if(mysql_commit(m_mysql)!=0)
+ if (mysql_commit(m_mysql) != 0)
throw_exception();
auto_commit(true);
}
bool is_alive()
{
- return mysql_ping(m_mysql)==0;
+ return mysql_ping(m_mysql) == 0;
}
template<typename Pred>
@@ -1014,15 +1129,16 @@
{
simple_execute(query, length);
- unsigned int fieldCount=mysql_field_count(m_mysql);
- MYSQL_RES* result=mysql_store_result(m_mysql);
- if(fieldCount>0 && result)
+ unsigned int fieldCount = mysql_field_count(m_mysql);
+ MYSQL_RES* result = mysql_store_result(m_mysql);
+ if (fieldCount > 0 && result)
{
- MYSQL_RES* result=mysql_store_result(m_mysql);
+ MYSQL_RES* result = mysql_store_result(m_mysql);
MYSQL_ROW row;
- while(row=mysql_fetch_row(result))
+ while (row = mysql_fetch_row(result))
{
- pred(*this, row, fieldCount);
+ if (!pred(*this, row, fieldCount))
+ break;
}
mysql_free_result(result);
return true;
@@ -1030,9 +1146,33 @@
return false;
}
-private:
- MYSQL* m_mysql;
- void throw_exception() { throw mysql::error(*this); }
+ template<typename LocalInfile>
+ void set_local_infile_factory(local_infile_factory<LocalInfile>* factory)
+ {
+ typedef local_infile_factory<LocalInfile> factory_type;
+ if (factory == nullptr)
+ {
+ reset_local_infile();
+ }
+ else
+ {
+ mysql_set_local_infile_handler(m_mysql, &factory_type::local_infile_init,
+ &factory_type::local_infile_read,
+ &factory_type::local_infile_end,
+ &factory_type::local_infile_error, factory);
+ }
+ }
+ void reset_local_infile()
+ {
+ mysql_set_local_infile_default(m_mysql);
+ }
+
+#if MARIADB_VERSION_ID >= 050500
+
+ async_connection async_mode();
+
+#endif //MariaDB
+
};
struct time : public MYSQL_TIME
@@ -1100,6 +1240,712 @@
}
};
+#if MARIADB_VERSION_ID >= 100000
+
+int event_flags(int status) NOEXCEPT
+{
+ int flags = 0;
+ if (status&MYSQL_WAIT_READ)
+ flags |= event::ef_read;
+ if (status&MYSQL_WAIT_WRITE)
+ flags |= event::ef_write;
+ if (status&MYSQL_WAIT_EXCEPT)
+ flags |= event::ef_exception;
+ return flags;
+}
+
+int mysql_status(int flags) NOEXCEPT
+{
+ int status = 0;
+ if (flags&event::ef_read)
+ status |= MYSQL_WAIT_READ;
+ if (flags&event::ef_write)
+ status |= MYSQL_WAIT_WRITE;
+ if (flags&event::ef_exception)
+ status |= MYSQL_WAIT_EXCEPT;
+ if (flags&event::ef_timeout)
+ status |= MYSQL_WAIT_TIMEOUT;
+ return status;
+}
+
+class async_connection;
+
+class async_statement : public base_statement
+{
+public:
+ async_statement() = default;
+ async_statement(async_connection& db);
+ async_statement(async_statement&& src)
+ : base_statement(std::move(src))
+ {
+ m_event=src.m_event;
+ src.m_event=nullptr;
+ }
+ async_statement& operator=(async_statement&& src)
+ {
+ base_statement::operator =(std::move(src));
+ m_event=src.m_event;
+ src.m_event=nullptr;
+ return *this;
+ }
+ ~async_statement()
+ {
+ close();
+ }
+
+ /*
+ Handler defiens as:
+ void handler(const qtl::mysql::error& e);
+ */
+ template<typename Handler>
+ void open(Handler&& handler, const char *query_text, size_t text_length=0)
+ {
+ if(text_length==0) text_length=strlen(query_text);
+ if(m_stmt)
+ {
+ std::string text(query_text, text_length);
+ reset([this, text, handler](const mysql::error& e) mutable {
+ if(e)
+ {
+ handler(e);
+ return;
+ }
+
+ prepare(text.data(), text.size(), std::forward<Handler>(handler));
+ });
+ }
+ else
+ {
+ prepare(query_text, text_length, std::forward<Handler>(handler));
+ }
+ }
+
+ /*
+ ExecuteHandler defiens as:
+ void handler(const qtl::mysql::error& e, uint64_t affected);
+ */
+ template<typename ExecuteHandler>
+ void execute(ExecuteHandler&& handler)
+ {
+ resize_binders(0);
+ int ret=0;
+ int status = mysql_stmt_execute_start(&ret, m_stmt);
+ if(status)
+ wait_execute(status, std::forward<ExecuteHandler>(handler));
+ else if(ret)
+ handler(mysql::error(*this), 0);
+ else
+ handler(mysql::error(), affetced_rows());
+ }
+
+ template<typename Types, typename Handler>
+ void execute(const Types& params, Handler&& handler)
+ {
+ unsigned long count=mysql_stmt_param_count(m_stmt);
+ if(count>0)
+ {
+ resize_binders(count);
+ qtl::bind_params(*this, params);
+ if(mysql_stmt_bind_param(m_stmt, &m_binders.front()))
+ throw_exception();
+ for(size_t i=0; i!=count; i++)
+ {
+ if(m_binderAddins[i].m_after_fetch)
+ m_binderAddins[i].m_after_fetch(m_binders[i]);
+ }
+ }
+ execute(std::forward<Handler>(handler));
+ }
+
+ template<typename Types, typename RowHandler, typename FinishHandler>
+ void fetch(Types&& values, RowHandler&& row_handler, FinishHandler&& finish_handler)
+ {
+ if(m_result==nullptr)
+ {
+ unsigned long count=mysql_stmt_field_count(m_stmt);
+ if(count>0)
+ {
+ m_result=mysql_stmt_result_metadata(m_stmt);
+ if(m_result==nullptr) throw_exception();
+ resize_binders(count);
+ qtl::bind_record(*this, std::forward<Types>(values));
+ set_binders();
+ if(mysql_stmt_bind_result(m_stmt, m_binders.data())!=0)
+ {
+ finish_handler(mysql::error(*this));
+ return;
+ }
+ }
+ }
+ fetch(std::forward<RowHandler>(row_handler), std::forward<FinishHandler>(finish_handler));
+ }
+
+ template<typename RowHandler, typename FinishHandler>
+ void fetch(RowHandler&& row_handler, FinishHandler&& finish_handler)
+ {
+ int ret = 0;
+ int status = start_fetch(&ret);
+ if (status == 0)
+ status = after_fetch(status, ret, std::forward<RowHandler>(row_handler), std::forward<FinishHandler>(finish_handler));
+ if (status)
+ wait_fetch(status, std::forward<RowHandler>(row_handler), std::forward<FinishHandler>(finish_handler));
+ }
+
+ template<typename Handler>
+ void reset(Handler&& handler)
+ {
+ my_bool ret=false;
+ int status = mysql_stmt_reset_start(&ret, m_stmt);
+ if(status)
+ wait_operation<my_bool>(status, &mysql_stmt_reset_cont, handler);
+ else
+ handler( (ret) ? mysql::error(*this) : mysql::error());
+ }
+
+ void close()
+ {
+ base_statement::close();
+ }
+
+ template<typename Handler>
+ void close(Handler&& handler)
+ {
+ if(m_result)
+ {
+ my_bool ret=0;
+ int status = mysql_stmt_free_result_start(&ret, m_stmt);
+ if(status)
+ {
+ this->wait_free_result(status, [this, handler](const mysql::error& e) mutable {
+ m_result=nullptr;
+ if (e)
+ handler(e);
+ else
+ close(handler);
+ });
+ }
+ else if(ret)
+ handler(mysql::error(*this));
+ else
+ m_result=nullptr;
+ }
+ if(m_stmt)
+ {
+ my_bool ret=0;
+ int status = mysql_stmt_close_start(&ret, m_stmt);
+ if(status)
+ {
+ wait_operation<my_bool>(status, &mysql_stmt_close_cont, [this, handler](const mysql::error& e) mutable {
+ m_stmt=nullptr;
+ handler(e);
+ });;
+ }
+ else
+ {
+ m_stmt = nullptr;
+ handler((ret) ? mysql::error(*this) : mysql::error());
+ }
+ }
+ }
+
+ template<typename Handler>
+ void next_result(Handler&& handler)
+ {
+ if(m_result)
+ {
+ mysql_free_result(m_result);
+ m_result = nullptr;
+ }
+ int ret=0;
+ do
+ {
+ int status = mysql_stmt_next_result_start(&ret, m_stmt);
+ if (status)
+ {
+ wait_next_result(status, std::forward<Handler>(handler));
+ return;
+ }
+ }while(ret==0 && mysql_stmt_field_count(m_stmt)<=0);
+ handler((ret) ? mysql::error(*this) : mysql::error());
+ }
+
+
+private:
+ event* m_event;
+
+ template<typename Handler>
+ void prepare(const char *query_text, size_t text_length, Handler&& handler)
+ {
+ int ret;
+ int status=mysql_stmt_prepare_start(&ret, m_stmt, query_text, (unsigned long)text_length);
+ if(status)
+ wait_operation<int>(status, &mysql_stmt_prepare_cont, handler);
+ else
+ handler((ret) ? mysql::error(*this) : mysql::error());
+ }
+
+ template<typename Ret, typename Func, typename Handler>
+ void wait_operation(int status, Func func, Handler&& handler)
+ {
+ m_event->set_io_handler(event_flags(status), mysql_get_timeout_value(m_stmt->mysql),
+ [this, func, handler](int flags) mutable {
+ Ret ret = 0;
+ int status = func(&ret, m_stmt, mysql_status(flags));
+ if (status)
+ wait_operation<Ret>(status, func, handler);
+ else
+ handler((ret) ? mysql::error(*this) : mysql::error());
+ });
+ }
+
+ template<typename ExecuteHandler>
+ void wait_execute(int status, ExecuteHandler&& handler)
+ {
+ m_event->set_io_handler(event_flags(status), mysql_get_timeout_value(m_stmt->mysql),
+ [this, handler](int flags) mutable {
+ int ret = 0;
+ int status = mysql_stmt_execute_cont(&ret, m_stmt, mysql_status(flags));
+ if (status)
+ wait_execute(status, handler);
+ else if(ret)
+ handler(mysql::error(*this), 0);
+ else
+ handler(mysql::error(), affetced_rows());
+ });
+ }
+
+ template<typename RowHandler, typename FinishHandler>
+ void wait_fetch(int status, RowHandler&& row_handler, FinishHandler&& finish_handler)
+ {
+ m_event->set_io_handler(event_flags(status), mysql_get_timeout_value(m_stmt->mysql),
+ [this, row_handler, finish_handler](int flags) mutable {
+ int ret = 0;
+ int status = mysql_stmt_fetch_cont(&ret, m_stmt, mysql_status(flags));
+ if (status == 0)
+ status = after_fetch(status, ret, std::forward<RowHandler>(row_handler), std::forward<FinishHandler>(finish_handler));
+ if (status)
+ wait_fetch(status, std::forward<RowHandler>(row_handler), std::forward<FinishHandler>(finish_handler));
+ });
+ }
+
+ int start_fetch(int* ret)
+ {
+ for (size_t i = 0; i != m_binders.size(); i++)
+ {
+ if (m_binderAddins[i].m_before_fetch)
+ m_binderAddins[i].m_before_fetch(m_binders[i]);
+ }
+ return mysql_stmt_fetch_start(ret, m_stmt);
+ }
+
+ template<typename RowHandler, typename FinishHandler>
+ int after_fetch(int status, int ret, RowHandler&& row_handler, FinishHandler&& finish_handler)
+ {
+ while (status == 0)
+ {
+ if (ret == 0 || ret == MYSQL_DATA_TRUNCATED)
+ {
+ for (size_t i = 0; i != m_binders.size(); i++)
+ {
+ m_binderAddins[i].is_truncated = (ret == MYSQL_DATA_TRUNCATED);
+ if (m_binderAddins[i].m_after_fetch)
+ m_binderAddins[i].m_after_fetch(m_binders[i]);
+ }
+ if (row_handler())
+ status = start_fetch(&ret);
+ else
+ {
+ finish_handler(mysql::error());
+ break;
+ }
+ }
+ else if (ret == 1)
+ {
+ finish_handler(mysql::error(*this));
+ break;
+ }
+ else
+ {
+ finish_handler(mysql::error());
+ break;
+ }
+ };
+ return status;
+ }
+
+ template<typename ResultHandler>
+ void wait_free_result(int status, ResultHandler&& handler) NOEXCEPT
+ {
+ m_event->set_io_handler(event_flags(status), mysql_get_timeout_value(m_stmt->mysql),
+ [this, handler](int flags) mutable {
+ my_bool ret = false;
+ int status = mysql_stmt_free_result_cont(&ret, m_stmt, mysql_status(flags));
+ if (status)
+ wait_free_result(status, handler);
+ else
+ handler(mysql::error());
+ });
+ }
+
+ template<typename Handler>
+ void wait_next_result(int status, Handler&& handler)
+ {
+ m_event->set_io_handler(event_flags(status), mysql_get_timeout_value(m_stmt->mysql),
+ [this, handler](int flags) mutable {
+ int ret = 0;
+ int status = mysql_stmt_next_result_cont(&ret, m_stmt, mysql_status(flags));
+ if (status)
+ wait_next_result(status, handler);
+ else if (ret)
+ handler(mysql::error(*this));
+ else if (mysql_stmt_field_count(m_stmt) > 0)
+ handler(mysql::error());
+ else
+ next_result(std::forward<Handler>(handler));
+ });
+ }
+
+};
+
+class async_connection : public basic_database, public qtl::async_connection<async_connection, async_statement>
+{
+public:
+ async_connection() = default;
+
+ /*
+ OpenHandler defines as:
+ void handler(const qtl::mysql::error& e) NOEXCEPT;
+ */
+ template<typename EventLoop, typename OpenHandler>
+ void open(EventLoop& ev, OpenHandler&& handler, const char *host, const char *user, const char *passwd, const char *db,
+ unsigned long clientflag = 0, unsigned int port = 0, const char *unix_socket = nullptr)
+ {
+ if (m_mysql == nullptr)
+ m_mysql = mysql_init(nullptr);
+
+ mysql_options(m_mysql, MYSQL_OPT_NONBLOCK, 0);
+
+ MYSQL* ret = nullptr;
+ int status = mysql_real_connect_start(&ret, m_mysql, host, user, passwd, db, port, unix_socket, clientflag);
+ bind(ev);
+ if (status)
+ wait_connect(status, std::forward<OpenHandler>(handler));
+ else
+ handler((ret == nullptr) ? mysql::error(*this) : mysql::error());
+ }
+
+ /*
+ CloseHandler defines as:
+ void handler() NOEXCEPT;
+ */
+ template<typename CloseHandler >
+ void close(CloseHandler&& handler) NOEXCEPT
+ {
+ int status = mysql_close_start(m_mysql);
+ if (status)
+ {
+ wait_close(status, [this, handler]() {
+ handler();
+ m_mysql = nullptr;
+ });
+ }
+ else
+ {
+ handler();
+ m_mysql = nullptr;
+ }
+ }
+
+ /*
+ Handler defines as:
+ void handler(const qtl::mysql::error& e) NOEXCEPT;
+ */
+ template<typename Handler>
+ void refresh(Handler&& handler, unsigned int options) NOEXCEPT
+ {
+ int ret = 0;
+ int status = mysql_refresh_start(&ret, m_mysql, options);
+ if (status)
+ wait_operation(status, &mysql_refresh_cont, std::forward<Handler>(handler));
+ else
+ handler((ret) ? mysql::error(*this) : mysql::error());
+ }
+
+ template<typename Handler>
+ void select(Handler&& handler, const char* db) NOEXCEPT
+ {
+ int ret = 0;
+ int status = mysql_select_db_start(&ret, m_mysql, db);
+ if (status)
+ wait_operation(status, &mysql_select_db_cont, std::forward<Handler>(handler));
+ else
+ handler((ret) ? mysql::error(*this) : mysql::error());
+ }
+
+ template<typename Handler>
+ void is_alive(Handler&& handler) NOEXCEPT
+ {
+ int ret = 0;
+ int status = mysql_ping_start(&ret, m_mysql);
+ if (status)
+ wait_operation(status, &mysql_ping_cont, std::forward<Handler>(handler));
+ else
+ handler((ret) ? mysql::error(*this) : mysql::error());
+ }
+
+ /*
+ Handler defines as:
+ void handler(const qtl::mysql::error& e, uint64_t affected) NOEXCEPT;
+ */
+ template<typename ExecuteHandler>
+ void simple_execute(ExecuteHandler&& handler, const char* query_text) NOEXCEPT
+ {
+ int ret = 0;
+ int status = mysql_query_start(&ret, m_mysql, query_text);
+ if (status)
+ {
+ wait_operation(status, &mysql_query_cont, [this, handler](const mysql::error& e) {
+ uint64_t affected = 0;
+ if (!e) affected = affected_rows();
+ handler(e, affected);
+ });
+ }
+ else
+ {
+ uint64_t affected = 0;
+ if (ret >= 0) affected = affected_rows();
+ handler((ret) ? mysql::error(*this) : mysql::error(), affected);
+ }
+ }
+
+ template<typename ExecuteHandler>
+ void simple_execute(ExecuteHandler&& handler, const char* query_text, unsigned long text_length) NOEXCEPT
+ {
+ int ret = 0;
+ int status = mysql_real_query_start(&ret, m_mysql, query_text, text_length);
+ if (status)
+ {
+ wait_operation(status, &mysql_real_query_cont, [this, handler](const mysql::error& e) mutable {
+ uint64_t affected = 0;
+ if (!e) affected = affected_rows();
+ handler(e, affected);
+ });
+ }
+ else
+ {
+ uint64_t affected = 0;
+ if (ret >= 0) affected = affected_rows();
+ handler((ret) ? mysql::error(*this) : mysql::error(), affected);
+ }
+ }
+
+ template<typename Handler>
+ void auto_commit(Handler&& handler, bool on) NOEXCEPT
+ {
+ my_bool ret;
+ int status = mysql_autocommit_start(&ret, m_mysql, on ? 1 : 0);
+ if(status)
+ wait_operation(status, &mysql_autocommit_cont, std::forward<Handler>(handler));
+ else
+ handler((ret) ? mysql::error(*this) : mysql::error());
+ }
+
+ template<typename Handler>
+ void begin_transaction(Handler&& handler) NOEXCEPT
+ {
+ auto_commit(handler, false);
+ }
+
+ template<typename Handler>
+ void rollback(Handler&& handler) NOEXCEPT
+ {
+ my_bool ret;
+ int status = mysql_rollback_start(&ret, m_mysql);
+ if (status)
+ wait_operation(status, &mysql_rollback_cont, std::forward<Handler>(handler));
+ else
+ handler((ret) ? mysql::error(*this) : mysql::error());
+ }
+
+ template<typename Handler>
+ void commit(Handler&& handler) NOEXCEPT
+ {
+ my_bool ret;
+ int status = mysql_commit_start(&ret, m_mysql);
+ if (status)
+ wait_operation(status, &mysql_commit_cont, std::forward<Handler>(handler));
+ else
+ handler((ret) ? mysql::error(*this) : mysql::error());
+ }
+
+ /*
+ RowHandler defines as:
+ bool row_handler(MYSQL_ROW row, int field_count) NOEXCEPT;
+ ResultHandler defines as:
+ void result_handler(const qtl::mysql::error& e, size_t row_count) NOEXCEPT;
+ */
+ template<typename RowHandler, typename ResultHandler>
+ void simple_query(const char* query, unsigned long length, RowHandler&& row_handler, ResultHandler&& result_handler) NOEXCEPT
+ {
+ simple_execute([this, row_handler, result_handler](const mysql::error& e, uint64_t affected) mutable {
+ if (e)
+ {
+ result_handler(e, 0);
+ return;
+ }
+
+ unsigned int field_count = mysql_field_count(m_mysql);
+ if (field_count > 0)
+ {
+ MYSQL_RES* result = nullptr;
+ int status = mysql_store_result_start(&result, m_mysql);
+ if (status)
+ wait_query(status, field_count, row_handler, result_handler);
+ else if (result)
+ fetch_rows(result, field_count, 0, row_handler, result_handler);
+ else
+ result_handler(mysql::error(*this), 0);
+ }
+ else
+ {
+ result_handler(mysql::error(), 0);
+ }
+ }, query, length);
+ }
+
+ template<typename Handler>
+ void open_command(const char* query_text, size_t text_length, Handler&& handler)
+ {
+ std::shared_ptr<async_statement> stmt=std::make_shared<async_statement>(*this);
+ stmt->open([stmt, handler](const mysql::error& e) mutable {
+ handler(e, stmt);
+ }, query_text, (unsigned long)text_length);
+ }
+
+ socket_type socket() const NOEXCEPT { return mysql_get_socket(m_mysql); }
+
+private:
+ template<typename OpenHandler>
+ void wait_connect(int status, OpenHandler&& handler) NOEXCEPT
+ {
+ m_event_handler->set_io_handler(event_flags(status), mysql_get_timeout_value(m_mysql),
+ [this, handler](int flags) mutable {
+ MYSQL* ret = nullptr;
+ int status = mysql_real_connect_cont(&ret, m_mysql, mysql_status(flags));
+ if (status)
+ wait_connect(status, handler);
+ else
+ handler((ret == nullptr) ? mysql::error(*this) : mysql::error());
+ });
+ }
+
+ template<typename CloseHandler>
+ void wait_close(int status, CloseHandler&& handler) NOEXCEPT
+ {
+ m_event_handler->set_io_handler(event_flags(status), mysql_get_timeout_value(m_mysql),
+ [this, handler](int flags) {
+ MYSQL* ret = nullptr;
+ int status = mysql_close_cont(m_mysql, mysql_status(flags));
+ if (status)
+ wait_close(status, handler);
+ else
+ handler();
+ });
+ }
+
+ template<typename Func, typename Handler>
+ void wait_operation(int status, Func func, Handler&& handler) NOEXCEPT
+ {
+ m_event_handler->set_io_handler(event_flags(status), mysql_get_timeout_value(m_mysql),
+ [this, func, handler](int flags) mutable {
+ int ret = 0;
+ int status = func(&ret, m_mysql, mysql_status(flags));
+ if (status)
+ wait_operation(status, func, handler);
+ else
+ handler((ret) ? mysql::error(*this) : mysql::error());
+ });
+ }
+
+ template<typename RowHandler, typename ResultHandler>
+ void wait_query(int status, int field_count, RowHandler&& row_handler, ResultHandler&& result_handler) NOEXCEPT
+ {
+ m_event_handler->set_io_handler(event_flags(status), mysql_get_timeout_value(m_mysql),
+ [this, field_count, row_handler, result_handler](int flags) {
+ MYSQL_RES* result = 0;
+ int status = mysql_store_result_cont(&result, m_mysql, mysql_status(flags));
+ if (status)
+ wait_query(status, field_count, row_handler, result_handler);
+ else if (result)
+ fetch_rows(result, field_count, 0, row_handler, result_handler);
+ else
+ result_handler(mysql::error(*this), 0);
+ });
+ }
+
+ template<typename RowHandler, typename ResultHandler>
+ void fetch_rows(MYSQL_RES* result, int field_count, size_t row_count, RowHandler&& row_handler, ResultHandler&& result_handler) NOEXCEPT
+ {
+ MYSQL_ROW row;
+ int status = mysql_fetch_row_start(&row, result);
+ if (status)
+ wait_fetch(status, result, field_count, row_count, row_handler, result_handler);
+ else if(row && row_handler(row, field_count))
+ fetch_rows(result, field_count, row_count + 1, row_handler, result_handler);
+ else
+ free_result(result, row_count, result_handler);
+ }
+
+ template<typename ResultHandler>
+ void free_result(MYSQL_RES* result, size_t row_count, ResultHandler&& result_handler) NOEXCEPT
+ {
+ int status = mysql_free_result_start(result);
+ if (status)
+ wait_free_result(status, result, row_count, result_handler);
+ else
+ result_handler(mysql::error(), row_count);
+ }
+
+ template<typename RowHandler, typename ResultHandler>
+ void wait_fetch(int status, MYSQL_RES* result, int field_count, size_t row_count, RowHandler&& row_handler, ResultHandler&& result_handler)
+ {
+ m_event_handler->set_io_handler(event_flags(status), mysql_get_timeout_value(m_mysql),
+ [this, result, field_count, row_count, row_handler, result_handler](int flags) mutable {
+ MYSQL_ROW row;
+ int status = mysql_fetch_row_cont(&row, result, mysql_status(flags));
+ if (status)
+ wait_fetch(status, result, field_count, row_count, row_handler, result_handler);
+ else if (result && row_handler(row, field_count))
+ fetch_rows(result, field_count, row_count+1, row_handler, result_handler);
+ else
+ free_result(result, row_count, result_handler);
+ });
+ }
+
+ template<typename ResultHandler>
+ void wait_free_result(int status, MYSQL_RES* result, size_t row_count, ResultHandler&& handler) NOEXCEPT
+ {
+ m_event_handler->set_io_handler(event_flags(status), mysql_get_timeout_value(m_mysql),
+ [this, result, row_count, handler](int flags) mutable {
+ MYSQL* ret = nullptr;
+ int status = mysql_free_result_cont(result, mysql_status(flags));
+ if (status)
+ wait_free_result(status, result, row_count, handler);
+ else
+ handler(mysql::error(), row_count);
+ });
+ }
+
+};
+
+inline async_statement::async_statement(async_connection& db)
+: base_statement(static_cast<basic_database&>(db))
+{
+ m_event=db.event();
+}
+
+#endif //MariaDB 10.0
+
typedef qtl::transaction<database> transaction;
template<typename Record>
@@ -1116,28 +1962,28 @@
return stmt;
}
-inline error::error(statement& stmt)
+inline error::error(const base_statement& stmt)
{
const char* errmsg=stmt.errmsg();
m_error=stmt.error();
if(errmsg) m_errmsg=errmsg;
}
-inline error::error(database& db)
+inline error::error(const basic_database& db)
{
const char* errmsg=db.errmsg();
m_error=db.error();
if(errmsg) m_errmsg=errmsg;
}
-inline statement::statement(database& db)
+inline base_statement::base_statement(basic_database& db)
{
m_stmt=mysql_stmt_init(db.handle());
- m_result=NULL;
+ m_result=nullptr;
}
}
}
-#endif //_MYDTL_MYSQL_H_
+#endif //_QTL_MYSQL_H_
--
Gitblit v1.9.3