#if MySQL

using System;
using System.Data;
using MySql.Data.MySqlClient;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Threading.Tasks;
using System.Text;
using System.Diagnostics;
using System.IO;
using MrWu.Model;
using System.Threading;
using MrWu.Debug;
using System.Data.SqlClient;

namespace MrWu.DB {
    /// <summary>
    /// mysql操作
    /// </summary>
    public class MySQL : SqlEx<MySqlConnection, MySqlCommand, MySqlDbType,MySqlParameter> {
        /// <summary>
        /// 链接池
        /// </summary>
        private class ConnectionPool : IConnectionPool<MySqlConnection> {
            /// <summary>
            /// 链接字符串
            /// </summary>
            private readonly string _connstr;

            string IConnectionPool<MySqlConnection>.connecStr => _connstr;

            private readonly int _capsize;
            /// <summary>
            /// 最大的缓存链接数量
            /// </summary>
            int IPool<MySqlConnection>.capSize => _capsize;

            private readonly int _surviveCountSize;

            /// <summary>
            /// 同时存在最大的链接数量
            /// </summary>
            /// <returns></returns>
            int IPool<MySqlConnection>.surviveCountSize => _surviveCountSize;

            /// <summary>
            /// 当前池中的数量
            /// </summary>
            int IPool<MySqlConnection>.capCount => _conns.Count;

            long _surviveCount = 0;

            /// <summary>
            /// 当前存活数量
            /// </summary>
            int IPool<MySqlConnection>.surviveCount {
                get {
                    return (int)Interlocked.Read(ref _surviveCount);
                }
            }

            /// <summary>
            /// 所有的链接
            /// </summary>
            private readonly ConcurrentQueue<MySqlConnection> _conns = new ConcurrentQueue<MySqlConnection>();

            ConcurrentQueue<MySqlConnection> IConnectionPool<MySqlConnection>.conns => _conns;

            /// <summary>
            /// 用于异步等待连接可用的信号量
            /// </summary>
            private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1);

            public ConnectionPool(SqlConfig config) {
                this._capsize = config.cacheCount;
                this._surviveCountSize = config.surviveCountSize;
                this._connstr = string.Format("server={0};port={1};user={2};password={3};AllowUserVariables={4};pooling=true;database=",
                                              config.host, config.port, config.username, config.pwd, config.AllowUserVariables);
            }

            /// <summary>
            /// 获取一个连接
            /// </summary>
            /// <param name="dbbase"></param>
            /// <returns></returns>
            public MySqlConnection TakeOut(params object[] dbbase) {
                if (dbbase.Length <= 0) return null;
                string dbname = dbbase[0] as string;
                if (dbname == null) return null;
                MySqlConnection conn;
                if (_conns.TryDequeue(out conn)) {
                    if (conn != null) {
                        return conn;
                    }
                }
                if (this._surviveCountSize > 0 && Interlocked.Read(ref _surviveCount) >= this._surviveCountSize)
                    return null;
                Interlocked.Increment(ref _surviveCount);
                return new MySqlConnection(this._connstr + dbname);
            }

            /// <summary>
            /// 异步获取一个连接（连接池满时不会阻塞线程）
            /// </summary>
            public async Task<MySqlConnection> TakeOutAsync(string dbname) {
                if (dbname == null) return null;
                while (true) {
                    MySqlConnection conn;
                    if (_conns.TryDequeue(out conn)) {
                        if (conn != null) return conn;
                    }
                    if (this._surviveCountSize == 0 || Interlocked.Read(ref _surviveCount) < this._surviveCountSize) {
                        Interlocked.Increment(ref _surviveCount);
                        return new MySqlConnection(this._connstr + dbname);
                    }
                    // 连接池满，异步等待
                    await _semaphore.WaitAsync();
                    _semaphore.Release();
                }
            }

            /// <summary>
            /// 放入一个连接
            /// </summary>
            /// <param name="conn"></param>
            public void PutInto(MySqlConnection conn) {
                if (_capsize == 0 || _conns.Count < _capsize) {
                    _conns.Enqueue(conn);
                } else {
                    conn.Dispose();
                    Interlocked.Decrement(ref _surviveCount);
                }
            }
        }

        /*
		 * 1.执行sql语句 获得执行sql语句后的结果
		 * 2.执行存储过程 获得存储过程的结果
		 * 
		 * 每个库都有一个链接
		 * 
		 * */

        private ConnectionPool _ConnPool = null;

        /// <summary>
        /// 链接池
        /// </summary>
        public override IConnectionPool<MySqlConnection> connPool => _ConnPool;

        /// <summary>
        /// 初始化
        /// </summary>
        /// <param name="config"></param>
        public MySQL(SqlConfig config) {
            this._ConnPool = new ConnectionPool(config);
            errorTryAgainCount = config.errorTryAgainCount;
            errorAgainTimeInc = config.errorAgainTimeInc;
            errorAgainTime = config.errorAgainTime;

            if (errorTryAgainCount < 0)
                throw new ArgumentException("errorTryAgainCount 不能小于 0");
            if (errorAgainTimeInc < 0)
                throw new ArgumentException("errorAgainTimeInc 不能小于 0");
            if (errorAgainTime < 0)
                throw new ArgumentException("errorAgainTime 不能小于 0");
        }

        /// <summary>
        /// 获取一个连接
        /// </summary>
        /// <param name="dbbase"></param>
        /// <returns></returns>
        public override MySqlConnection GetConnection(string dbbase) {
            MySqlConnection tmp;
            while (true) {
                tmp = _ConnPool.TakeOut(dbbase);
                if (tmp != null)
                    return tmp;
                Debug.Debug.Warning("链接数超过上限,等待重试");
                Thread.Sleep(2);
            }
        }

        /// <summary>
        /// 异步获取一个连接（不会阻塞线程）
        /// </summary>
        public async Task<MySqlConnection> GetConnectionAsync(string dbbase) {
            return await _ConnPool.TakeOutAsync(dbbase);
        }

        /// <summary>
        /// 开始存储过程
        /// </summary>
        /// <param name="dbbase">存储过程所在的数据库</param>
        /// <param name="procedure">存储过程名</param>
        /// <returns></returns>
        public override IProdureParameter<MySqlCommand> BeginProcedure(string dbbase, string procedure) {
            if (_ConnPool == null)
                throw new Exception("you dont hava init");

            MySqlCommand cmd = new MySqlCommand(procedure, GetConnection(dbbase));
            MySqlProcedureParameter result = new MySqlProcedureParameter(cmd, dbbase);
            //存储过程模式
            cmd.CommandType = CommandType.StoredProcedure;
            return result;
        }

        /// <summary>
        /// 添加参数
        /// </summary>
        /// <param name="spp">存储过程查询键</param>
        /// <param name="paramName">参数名</param>
        /// <param name="value">参数值</param>
        /// <param name="type">参数类型</param>
        /// <param name="pd">参数种类</param>
        public override void AddParam(IProdureParameter<MySqlCommand> spp, string paramName, object value, MySqlDbType type, ParameterDirection pd = ParameterDirection.Input) {
            if (spp == null || spp.command == null)
                throw new Exception("param spp null");
            MySqlParameter param = spp.command.Parameters.Add(paramName, type);
            param.Direction = pd;
            if (value == null)
                param.Value = DBNull.Value;
            else
                param.Value = value;
        }

        /// <summary>
        /// 提交存储过程
        /// </summary>
        /// <param name="spp">存储过程查询键</param>
        /// <param name="ds">如果 存储过程中有查询表操作,则需要传递DataSet实例</param>
        /// <param name="againCount">重试次数 小于0表示使用配置值 大于等于0表示自定义</param>
        /// <returns>参数值,输出参数从字典中获取</returns>
        public override Dictionary<string, object> SubProcedure(IProdureParameter<MySqlCommand> spp, DataSet ds = null, int againCount = -1) {
            if (spp == null || spp.command == null)
                throw new Exception("param spp null");

            int cnt = Execute(spp.command, spp.dbName, ds, againCount);

            Dictionary<string, object> dic = new Dictionary<string, object>();
            var pms = spp.command.Parameters;
            int len = pms.Count;
            for (int i = 0; i < len; i++) {
                dic.Add(pms[i].ParameterName, pms[i].Value);
            }
            
            if (!dic.ContainsKey(DbConst.RowsAffected))
            {
                dic[DbConst.RowsAffected] = cnt;
            }
            
            spp.Dispose();
            return dic;
        }

        /// <summary>
        /// 执行sql语句
        /// </summary>
        /// <param name="dbbase">数据库</param>
        /// <param name="sql">sql语句</param>
        /// <param name="mysqlparams"></param>
        /// <param name="againCount">重试次数 小于0表示使用配置值 大于等于0表示自定义</param>
        /// <param name="ds">如果是查询 使用传入ds实例</param>
        public override int ExecuteSql(string dbbase, string sql, IEnumerable<MySqlParameter> mysqlparams = null, DataSet ds = null, int againCount = -1) {
            MySqlCommand cmd = new MySqlCommand(sql, GetConnection(dbbase));
            cmd.CommandType = CommandType.Text;

            if (mysqlparams != null) {
                //int len = mysqlparams.Count;
                //for (int i = 0; i < len; i++) {
                //    ISqlAloneParameter<MySqlDbType> msap = mysqlparams[i];
                //    cmd.Parameters.Add(msap.paramname, msap.type, msap.valueLength);
                //    cmd.Parameters[msap.paramname].Value = msap.paramvalue;
                //}
                foreach (var param in mysqlparams) {
                    cmd.Parameters.Add(param);
                }
            }

            return Execute(cmd, dbbase, ds, againCount);
        }

        /// <summary>
        /// 执行sql语句
        /// </summary>
        /// <param name="dbbase">数据库</param>
        /// <param name="sql">sql语句</param>
        /// <param name="mysqlparams">附加参数</param>
        /// <returns></returns>
        public MySqlCommand GetCommandBySql(string dbbase, string sql, IEnumerable<MySqlParameter> mysqlparams = null) {
            MySqlCommand cmd = new MySqlCommand(sql, GetConnection(dbbase));
            cmd.CommandType = CommandType.Text;

            if (mysqlparams != null) {
                //int len = mysqlparams.Count;
                //for (int i = 0; i < len; i++) {
                //    ISqlAloneParameter<MySqlDbType> msap = mysqlparams[i];
                //    cmd.Parameters.Add(msap.paramname, msap.type, msap.valueLength);
                //    cmd.Parameters[msap.paramname].Value = msap.paramvalue;
                //}
                foreach (var param in mysqlparams) {
                    cmd.Parameters.Add(param);
                }
            }
            return cmd;
        }

        /// <summary>
        /// 普通查询一个表
        /// </summary>
        /// <param name="dbbase">数据库名称</param>
        /// <param name="table">表名</param>
        /// <param name="column">列名</param>
        /// <param name="where">条件</param>
        /// <param name="count">查询的数量</param>
        /// <param name="againCount">重试次数 小于0表示使用配置值 大于等于0表示自定义</param>
        /// <param name="sqlparams">sqlparams</param>
        /// <returns>查询到的表</returns>
        public override DataTable Select(string dbbase, string table, string[] column = null, string where = null, int count = 0, int againCount = -1, IEnumerable<MySqlParameter> sqlparams = null) {
            StringBuilder sb = new StringBuilder();
            sb.Append("select ");
            if (column != null && column.Length > 0) {
                int len = column.Length;
                for (int i = len - 1; i >= 0; i--) {
                    sb.Append(column[i]);
                    if (i > 0)
                        sb.Append(",");
                }
            } else {
                sb.Append("*");
            }

            sb.Append(" from ");
            sb.Append(table);
            if (!string.IsNullOrEmpty(where)) {
                sb.Append(" where ");
                sb.Append(where);
            }

            if (count > 0) {
                sb.AppendFormat(" limit {0}", count);
            }

            DataSet ds = new DataSet();

            ExecuteSql(dbbase, sb.ToString(), sqlparams, ds, againCount);
            if (ds.Tables == null || ds.Tables.Count == 0)
                return null;
            return ds.Tables[0];
        }

        /// <summary>
        /// 执行sql命令
        /// </summary>
        /// <param name="cmd">指令</param>
        /// <param name="dbname">数据库名称</param>
        /// <param name="ds">null表示不需要查询结果集 有值将会把结果集放到 ds中</param>
        /// <param name="againCount">重试次数 小于0表示使用配置值 大于等于0表示自定义</param>
        public override int Execute(MySqlCommand cmd, string dbname, DataSet ds = null, int againCount = -1) {

           // Debug.Debug.Log("执行sql指令!");

            int result = 0;
            int exceCnt = 0;    //执行次数
                                //重试次数
            if (againCount < 0)
                againCount = errorTryAgainCount;
            try {
                while (exceCnt <= againCount) {
                    try {
                        if (cmd.Connection.State == ConnectionState.Closed)
                            cmd.Connection.Open();
                        cmd.Connection.ChangeDatabase(dbname);

                        if (ds == null)
                            result = cmd.ExecuteNonQuery();
                        else {
                            using(MySqlDataAdapter sda = new MySqlDataAdapter()) {
                                sda.SelectCommand = cmd;
                                sda.Fill(ds);
                            }
                        }
                        break;
                    } catch (Exception e) {
                        cmd.Connection.Close();
                        if (exceCnt < againCount)
                            Debug.Debug.Error("数据库执行错误重试次数:" + exceCnt + ",错误类型:" + e.Message);
                        else
                            throw e;
                    }

                    int time = errorAgainTime + errorAgainTimeInc * exceCnt;
                    exceCnt++; //执行次数+1
                    Thread.Sleep(time);
                }
            } finally {
                _ConnPool.PutInto(cmd.Connection);   //放入链接
            }
            return result;
        }

        /// <summary>
        /// 开始一个事务
        /// </summary>
        /// <param name="dbbase">数据库</param>
        /// <returns></returns>
        public override IProdureParameter<MySqlCommand> BeginTransaction(string dbbase) {
            MySqlCommand cmd = new MySqlCommand();
            MySqlProcedureParameter spp = new MySqlProcedureParameter(cmd, dbbase);
            try {
                cmd.Connection = GetConnection(dbbase);
                if (cmd.Connection.State == ConnectionState.Closed)
                    cmd.Connection.Open();
                cmd.Connection.ChangeDatabase(dbbase);
                cmd.Transaction = cmd.Connection.BeginTransaction(); //开始事务，还可以设置隔离等级
            } catch (Exception e) {
                spp.exception = e;
            }
            return spp;
        }

        /// <summary>
        /// 事务添加sql语句
        /// </summary>
        /// <param name="spp"></param>
        /// <param name="sql"></param>
        /// <param name="sqlparams"></param>
        public override int TransactionAddSql(IProdureParameter<MySqlCommand> spp, string sql, List<ISqlAloneParameter<MySqlDbType>> sqlparams = null) {
            if (spp.isException)
                return 0;
            try {
                MySqlCommand cmd = spp.command;
                cmd.CommandText = sql;
                cmd.Parameters.Clear();

                if (sqlparams != null) {
                    int len = sqlparams.Count;
                    for (int i = 0; i < len; i++) {
                        ISqlAloneParameter<MySqlDbType> sap = sqlparams[i];
                        cmd.Parameters.Add(sap.paramname, sap.type, sap.valueLength);
                        cmd.Parameters[sap.paramname].Value = sap.paramvalue;
                    }
                }

                return cmd.ExecuteNonQuery();
            } catch (Exception e) {
                spp.exception = e;
                throw e;
            }
        }

        /// <summary>
        /// 提交事务
        /// </summary>
        /// <param name="spp"></param>
        /// <returns>true 表示提交成功 false 表示提交失败</returns>
        public override bool SubTransaction(IProdureParameter<MySqlCommand> spp) {
            try {
                spp.command.Transaction.Commit();
            } catch (Exception e) {
                spp.exception = e;
                spp.command.Transaction.Rollback();
            } finally {
                _ConnPool.PutInto(spp.command.Connection);
                spp.Dispose();
            }
            return spp.exception == null;
        }

        /// <summary>
        /// 回滚事务
        /// </summary>
        /// <param name="spp"></param>
        public override void RollbackTransaction(IProdureParameter<MySqlCommand> spp) {
            try {
                spp.command.Transaction.Rollback();
            } catch (Exception ex) {
                Debug.Debug.Error("RollbackTransaction ex: " + ex);
            }
        }

        #region 异步方法

        /// <summary>
        /// 异步提交存储过程
        /// </summary>
        public override async Task<Dictionary<string, object>> SubProcedureAsync(IProdureParameter<MySqlCommand> spp, DataSet ds = null, int againCount = -1) {
            if (spp == null || spp.command == null)
                throw new Exception("param spp null");

            int cnt = await ExecuteAsync(spp.command, spp.dbName, ds, againCount);

            Dictionary<string, object> dic = new Dictionary<string, object>();
            var pms = spp.command.Parameters;
            int len = pms.Count;
            for (int i = 0; i < len; i++) {
                dic.Add(pms[i].ParameterName, pms[i].Value);
            }

            if (!dic.ContainsKey(DbConst.RowsAffected)) {
                dic[DbConst.RowsAffected] = cnt;
            }

            spp.Dispose();
            return dic;
        }

        /// <summary>
        /// 异步执行sql语句
        /// </summary>
        public override async Task<int> ExecuteSqlAsync(string dbbase, string sql, IEnumerable<MySqlParameter> mysqlparams = null, DataSet ds = null, int againCount = -1) {
            MySqlCommand cmd = new MySqlCommand(sql, await GetConnectionAsync(dbbase));
            cmd.CommandType = CommandType.Text;

            if (mysqlparams != null) {
                foreach (var param in mysqlparams) {
                    cmd.Parameters.Add(param);
                }
            }

            return await ExecuteAsync(cmd, dbbase, ds, againCount);
        }

        /// <summary>
        /// 异步查询
        /// </summary>
        public override async Task<DataTable> SelectAsync(string dbbase, string table, string[] column = null, string where = null, int count = 0, int againCount = -1, IEnumerable<MySqlParameter> sqlparams = null) {
            StringBuilder sb = new StringBuilder();
            sb.Append("select ");
            if (column != null && column.Length > 0) {
                int len = column.Length;
                for (int i = len - 1; i >= 0; i--) {
                    sb.Append(column[i]);
                    if (i > 0)
                        sb.Append(",");
                }
            } else {
                sb.Append("*");
            }

            sb.Append(" from ");
            sb.Append(table);
            if (!string.IsNullOrEmpty(where)) {
                sb.Append(" where ");
                sb.Append(where);
            }

            if (count > 0) {
                sb.AppendFormat(" limit {0}", count);
            }

            DataSet ds = new DataSet();

            await ExecuteSqlAsync(dbbase, sb.ToString(), sqlparams, ds, againCount);
            if (ds.Tables == null || ds.Tables.Count == 0)
                return null;
            return ds.Tables[0];
        }

        /// <summary>
        /// 异步执行sql命令
        /// </summary>
        public override async Task<int> ExecuteAsync(MySqlCommand cmd, string dbname, DataSet ds = null, int againCount = -1) {
            int result = 0;
            int exceCnt = 0;
            if (againCount < 0)
                againCount = errorTryAgainCount;
            try {
                while (exceCnt <= againCount) {
                    try {
                        if (cmd.Connection.State == ConnectionState.Closed)
                            await cmd.Connection.OpenAsync();
                        cmd.Connection.ChangeDatabase(dbname);

                        if (ds == null)
                            result = await cmd.ExecuteNonQueryAsync();
                        else {
                            using(MySqlDataAdapter sda = new MySqlDataAdapter()) {
                                sda.SelectCommand = cmd;
                                sda.Fill(ds);
                            }
                        }
                        break;
                    } catch (Exception e) {
                        cmd.Connection.Close();
                        if (exceCnt < againCount)
                            Debug.Debug.Error("数据库异步执行错误重试次数:" + exceCnt + ",错误类型:" + e.Message);
                        else
                            throw;
                    }

                    int time = errorAgainTime + errorAgainTimeInc * exceCnt;
                    exceCnt++;
                    await Task.Delay(time);
                }
            } finally {
                _ConnPool.PutInto(cmd.Connection);
            }
            return result;
        }

        #endregion

        /// <summary>
        /// 保存每个存储过程为单个文件
        /// </summary>
        /// <param name="dbbase">那个数据库的</param>
        /// <param name="path">存储的路径</param>
        /// <param name="method">如果导出其中一个存储过程，则填存储过程名称</param>
        public void SaveMethod(string dbbase, string path, string method = null) {

            DataTable dt;

            if (method == null)
                dt = Select("information_schema", "routines", new string[] { "routine_name", "routine_type" }, string.Format("routine_schema='{0}'", dbbase));
            else
                dt = Select("information_schema", "routines", new string[] { "routine_name", "routine_type" }, string.Format("routine_schema='{0}' and routine_name='{1}'", dbbase, method));

            foreach (DataColumn item in dt.Columns) {
                Console.WriteLine(item.ColumnName);
            }
            int len = dt.Rows.Count;
            for (int i = 0; i < len; i++) {
                SaveMethod(dbbase, path, dt.Rows[i]["routine_name"] as string, dt.Rows[i]["routine_type"] as string);
            }
        }

        /// <summary>
        /// 保存方法
        /// </summary>
        /// <param name="dbbase"></param>
        /// <param name="path"></param>
        /// <param name="methodName"></param>
        /// <param name="methodType"></param>
        private void SaveMethod(string dbbase, string path, string methodName, string methodType) {
            if (!Directory.Exists(path)) {
                Directory.CreateDirectory(path);
            }

            string sql = string.Format("show create {0} {1};", methodType, methodName);
            DataSet ds = new DataSet();
            ExecuteSql(dbbase, sql, null, ds);

            if (ds.Tables == null || ds.Tables.Count == 0 || ds.Tables[0].Rows == null || ds.Tables[0].Rows.Count == 0)
                return;
            path = string.Format("{0}/{1}.sql", path, methodName);

            string sqlCode = string.Format("Drop {0} if exists {1};{2}", methodType, methodName, Environment.NewLine);
            sqlCode += "Commit;" + Environment.NewLine;
            sqlCode += ds.Tables[0].Rows[0][string.Format("Create {0}", methodType)] as string;

            File.WriteAllText(path, sqlCode, Encoding.UTF8);
        }

        /// <summary>
        /// 执行sql文件
        /// </summary>
        /// <param name="dbbase"></param>
        /// <param name="path"></param>
        public void Execute(string dbbase, string path) {
            if (File.Exists(path)) {
                try {
                    string sql = File.ReadAllText(path);
                    ExecuteSql(dbbase, sql);
                } catch (Exception e) {
                    Console.WriteLine("运行出错:" + path + Environment.NewLine + e.ToString());
                }
            } else if (Directory.Exists(path)) {
                DirectoryInfo dinfo = new DirectoryInfo(path);

                string[] paths = Directory.GetDirectories(path);

                foreach (var item in paths) {

                    Console.WriteLine(item);

                    Execute(dbbase, item);
                }

                FileInfo[] finfos = dinfo.GetFiles();

                foreach (var fl in finfos) {
                    Execute(dbbase, fl.FullName);
                }
            }
        }
    }
}
#endif