tk/FreeSql
1
0
Fork 0
mirror of https://github.com/nsnail/FreeSql.git synced 2025-06-18 20:08:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

## v0.9.18

Browse Source 
- 增加 PostgreSQL 的 Odbc 访问提供，相比 FreeSql.Provider.PostgreSQL 支持的类型更少；
- 增加 通用的 Odbc 访问提供，不能迁移实体到数据库，不能 Skip 这样来分页，理论上能 crud 所有 odbc 数据库；
This commit is contained in:
28810
2019-09-21 00:48:34 +08:00
parent 90458cca38
commit dcf1da36d4
97 changed files with 18833 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
Providers/FreeSql.Provider.Odbc/PostgreSQL/OdbcPostgreSQLAdo/OdbcPostgreSQLAdo.cs Normal file
View File

@ -0,0 +1,71 @@
using FreeSql.Internal;
using SafeObjectPool;
using System;
using System.Collections;
using System.Collections.Generic;
using System.Data.Common;
using System.Data.Odbc;
using System.Text;
using System.Threading;
namespace FreeSql.Odbc.PostgreSQL
{
class OdbcPostgreSQLAdo : FreeSql.Internal.CommonProvider.AdoProvider
{
public OdbcPostgreSQLAdo() : base(DataType.PostgreSQL) { }
public OdbcPostgreSQLAdo(CommonUtils util, string masterConnectionString, string[] slaveConnectionStrings) : base(DataType.PostgreSQL)
{
base._util = util;
if (!string.IsNullOrEmpty(masterConnectionString))
MasterPool = new OdbcPostgreSQLConnectionPool("主库", masterConnectionString, null, null);
if (slaveConnectionStrings != null)
{
foreach (var slaveConnectionString in slaveConnectionStrings)
{
var slavePool = new OdbcPostgreSQLConnectionPool($"从库{SlavePools.Count + 1}", slaveConnectionString, () => Interlocked.Decrement(ref slaveUnavailables), () => Interlocked.Increment(ref slaveUnavailables));
SlavePools.Add(slavePool);
}
}
}
static DateTime dt1970 = new DateTime(1970, 1, 1);
public override object AddslashesProcessParam(object param, Type mapType)
{
if (param == null) return "NULL";
if (mapType != null && mapType != param.GetType())
param = Utils.GetDataReaderValue(mapType, param);
if (param is bool || param is bool?)
return (bool)param ? "'t'" : "'f'";
else if (param is string || param is char)
return string.Concat("'", param.ToString().Replace("'", "''"), "'");
else if (param is Enum)
return ((Enum)param).ToInt64();
else if (decimal.TryParse(string.Concat(param), out var trydec))
return param;
else if (param is DateTime || param is DateTime?)
return string.Concat("'", ((DateTime)param).ToString("yyyy-MM-dd HH:mm:ss.ffffff"), "'");
else if (param is TimeSpan || param is TimeSpan?)
return ((TimeSpan)param).Ticks / 10;
else if (param is IEnumerable)
{
var sb = new StringBuilder();
var ie = param as IEnumerable;
foreach (var z in ie) sb.Append(",").Append(AddslashesProcessParam(z, mapType));
return sb.Length == 0 ? "(NULL)" : sb.Remove(0, 1).Insert(0, "(").Append(")").ToString();
}
return string.Concat("'", param.ToString().Replace("'", "''"), "'");
}
protected override DbCommand CreateCommand()
{
return new OdbcCommand();
}
protected override void ReturnConnection(ObjectPool<DbConnection> pool, Object<DbConnection> conn, Exception ex)
{
(pool as OdbcPostgreSQLConnectionPool).Return(conn, ex);
}
protected override DbParameter[] GetDbParamtersByObject(string sql, object obj) => _util.GetDbParamtersByObject(sql, obj);
}
}

241
Providers/FreeSql.Provider.Odbc/PostgreSQL/OdbcPostgreSQLAdo/OdbcPostgreSQLConnectionPool.cs Normal file
View File

@ -0,0 +1,241 @@
using SafeObjectPool;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Data;
using System.Data.Common;
using System.Data.Odbc;
using System.Text;
using System.Text.RegularExpressions;
using System.Threading.Tasks;
namespace FreeSql.Odbc.PostgreSQL
{
class OdbcPostgreSQLConnectionPool : ObjectPool<DbConnection>
{
internal Action availableHandler;
internal Action unavailableHandler;
public OdbcPostgreSQLConnectionPool(string name, string connectionString, Action availableHandler, Action unavailableHandler) : base(null)
{
var policy = new OdbcPostgreSQLConnectionPoolPolicy
{
_pool = this,
Name = name
};
this.Policy = policy;
policy.ConnectionString = connectionString;
this.availableHandler = availableHandler;
this.unavailableHandler = unavailableHandler;
}
public void Return(Object<DbConnection> obj, Exception exception, bool isRecreate = false)
{
if (exception != null && exception is OdbcException)
{
if (exception is System.IO.IOException)
{
base.SetUnavailable(exception);
}
else if (obj.Value.Ping() == false)
{
base.SetUnavailable(exception);
}
}
base.Return(obj, isRecreate);
}
}
class OdbcPostgreSQLConnectionPoolPolicy : IPolicy<DbConnection>
{
internal OdbcPostgreSQLConnectionPool _pool;
public string Name { get; set; } = "PostgreSQL NpgsqlConnection 对象池";
public int PoolSize { get; set; } = 50;
public TimeSpan SyncGetTimeout { get; set; } = TimeSpan.FromSeconds(10);
public TimeSpan IdleTimeout { get; set; } = TimeSpan.Zero;
public int AsyncGetCapacity { get; set; } = 10000;
public bool IsThrowGetTimeoutException { get; set; } = true;
public int CheckAvailableInterval { get; set; } = 5;
static ConcurrentDictionary<string, int> dicConnStrIncr = new ConcurrentDictionary<string, int>(StringComparer.CurrentCultureIgnoreCase);
private string _connectionString;
public string ConnectionString
{
get => _connectionString;
set
{
_connectionString = value ?? "";
var pattern = @"Max(imum)?\s*pool\s*size\s*=\s*(\d+)";
Match m = Regex.Match(_connectionString, pattern, RegexOptions.IgnoreCase);
if (m.Success == false || int.TryParse(m.Groups[2].Value, out var poolsize) == false || poolsize <= 0) poolsize = 50;
var connStrIncr = dicConnStrIncr.AddOrUpdate(_connectionString, 1, (oldkey, oldval) => oldval + 1);
PoolSize = poolsize + connStrIncr;
_connectionString = m.Success ?
Regex.Replace(_connectionString, pattern, $"Maximum pool size={PoolSize}", RegexOptions.IgnoreCase) :
$"{_connectionString};Maximum pool size={PoolSize}";
pattern = @"Connection\s*LifeTime\s*=\s*(\d+)";
m = Regex.Match(_connectionString, pattern, RegexOptions.IgnoreCase);
if (m.Success)
{
IdleTimeout = TimeSpan.FromSeconds(int.Parse(m.Groups[1].Value));
_connectionString = Regex.Replace(_connectionString, pattern, "", RegexOptions.IgnoreCase);
}
var minPoolSize = 0;
pattern = @"Min(imum)?\s*pool\s*size\s*=\s*(\d+)";
m = Regex.Match(_connectionString, pattern, RegexOptions.IgnoreCase);
if (m.Success)
{
minPoolSize = int.Parse(m.Groups[2].Value);
_connectionString = Regex.Replace(_connectionString, pattern, "", RegexOptions.IgnoreCase);
}
FreeSql.Internal.CommonUtils.PrevReheatConnectionPool(_pool, minPoolSize);
}
}
public bool OnCheckAvailable(Object<DbConnection> obj)
{
if (obj.Value.State == ConnectionState.Closed) obj.Value.Open();
return obj.Value.Ping(true);
}
public DbConnection OnCreate()
{
var conn = new OdbcConnection(_connectionString);
return conn;
}
public void OnDestroy(DbConnection obj)
{
if (obj.State != ConnectionState.Closed) obj.Close();
obj.Dispose();
}
public void OnGet(Object<DbConnection> obj)
{
if (_pool.IsAvailable)
{
if (obj.Value == null)
{
if (_pool.SetUnavailable(new Exception("连接字符串错误")) == true)
throw new Exception($"【{this.Name}】连接字符串错误，请检查。");
return;
}
if (obj.Value.State != ConnectionState.Open || DateTime.Now.Subtract(obj.LastReturnTime).TotalSeconds > 60 && obj.Value.Ping() == false)
{
try
{
obj.Value.Open();
}
catch (Exception ex)
{
if (_pool.SetUnavailable(ex) == true)
throw new Exception($"【{this.Name}】状态不可用，等待后台检查程序恢复方可使用。{ex.Message}");
}
}
}
}
async public Task OnGetAsync(Object<DbConnection> obj)
{
if (_pool.IsAvailable)
{
if (obj.Value == null)
{
if (_pool.SetUnavailable(new Exception("连接字符串错误")) == true)
throw new Exception($"【{this.Name}】连接字符串错误，请检查。");
return;
}
if (obj.Value.State != ConnectionState.Open || DateTime.Now.Subtract(obj.LastReturnTime).TotalSeconds > 60 && (await obj.Value.PingAsync()) == false)
{
try
{
await obj.Value.OpenAsync();
}
catch (Exception ex)
{
if (_pool.SetUnavailable(ex) == true)
throw new Exception($"【{this.Name}】状态不可用，等待后台检查程序恢复方可使用。{ex.Message}");
}
}
}
}
public void OnGetTimeout()
{
}
public void OnReturn(Object<DbConnection> obj)
{
}
public void OnAvailable()
{
_pool.availableHandler?.Invoke();
}
public void OnUnavailable()
{
_pool.unavailableHandler?.Invoke();
}
}
static class DbConnectionExtensions
{
static DbCommand PingCommand(DbConnection conn)
{
var cmd = conn.CreateCommand();
cmd.CommandTimeout = 5;
cmd.CommandText = "select 1";
return cmd;
}
public static bool Ping(this DbConnection that, bool isThrow = false)
{
try
{
PingCommand(that).ExecuteNonQuery();
return true;
}
catch
{
if (that.State != ConnectionState.Closed) try { that.Close(); } catch { }
if (isThrow) throw;
return false;
}
}
async public static Task<bool> PingAsync(this DbConnection that, bool isThrow = false)
{
try
{
await PingCommand(that).ExecuteNonQueryAsync();
return true;
}
catch
{
if (that.State != ConnectionState.Closed) try { that.Close(); } catch { }
if (isThrow) throw;
return false;
}
}
}
}