Add DbDataSource Multiplexing support, add MySql DbDataSource support

docs/DistributedLock.MySql.md

@@ -21,7 +21,7 @@ await using (await @lock.AcquireAsync())
 
 MySQL-based locks have been tested against and work with both the [MySQL](https://www.mysql.com/) and [MariaDB](https://mariadb.org/).
 
-MySQL-based locks locks can be constructed with a `connectionString`, an `IDbConnection` or an `IDbTransaction` as a means of connecting to the database. In most cases, using a `connectionString` is preferred because it allows for the library to efficiently multiplex connections under the hood and eliminates the risk that the passed-in `IDbConnection` gets used in a way that disrupts the locking process. Using an `IDbTransaction` is generally equivalent to using an `IDbConnection` (the lock is still connection-scoped), but it allows the lock to participate in an ongoing transaction. **NOTE that since `IDbConnection`/`IDbTransaction` objects are not thread-safe, lock objects constructed with them can only be used by one thread at a time.**
+MySQL-based locks locks can be constructed with a `connectionString`, a `DbDataSource` (.NET 7+), an `IDbConnection` or an `IDbTransaction` as a means of connecting to the database. In most cases, using a `connectionString` or a `DbDataSource` is preferred because it allows for the library to efficiently multiplex connections under the hood and eliminates the risk that the passed-in `IDbConnection` gets used in a way that disrupts the locking process. Note that connections are multiplexed per `DbDataSource` instance: two data sources with the same connection string do not share connections. Using an `IDbTransaction` is generally equivalent to using an `IDbConnection` (the lock is still connection-scoped), but it allows the lock to participate in an ongoing transaction. **NOTE that since `IDbConnection`/`IDbTransaction` objects are not thread-safe, lock objects constructed with them can only be used by one thread at a time.**
 
 Natively, MySQL's locking functions are case-insensitive with respect to the lock name. Since the DistributedLock library as a whole uses case-sensitive names, lock names containing uppercase characters will be transformed/hashed under the hood (as will empty names or names that are too long). If your program needs to coordinate with other code that is using `GET_LOCK` directly, be sure to express the name in lower case and to pass `exactName: true` when constructing the lock instance (in `exactName` mode, an invalid name will throw an exception rather than silently being transformed into a valid one).
 

docs/DistributedLock.Postgres.md

@@ -39,7 +39,7 @@ Under the hood, [Postgres advisory locks can be based on either one 64-bit integ
 - Passing an ASCII string with 0-9 characters, which will be mapped to a `long` based on a custom scheme.
 - Passing an arbitrary string with the `allowHashing` option set to `true` which will be hashed to a `long`. Note that hashing will only be used if other methods of interpreting the string fail.
 
-In addition to specifying the `key`, Postgres-based locks allow you to specify either a `connectionString`, an `IDbConnection`, or a `DbDataSource` as a means of connecting to the database. In most cases, using a `connectionString` is preferred because it allows for the library to efficiently multiplex connections under the hood and, in the case of `IDbConnection`, eliminates the risk that the passed-in `IDbConnection` gets used in a way that disrupts the locking process. **NOTE that since `IDbConnection` objects are not thread-safe, lock objects constructed with them can only be used by one thread at a time.**
+In addition to specifying the `key`, Postgres-based locks allow you to specify either a `connectionString`, an `IDbConnection`, or a `DbDataSource` as a means of connecting to the database. In most cases, using a `connectionString` or a `DbDataSource` is preferred because it allows for the library to efficiently multiplex connections under the hood and eliminates the risk that the passed-in `IDbConnection` gets used in a way that disrupts the locking process. Note that connections are multiplexed per `DbDataSource` instance: two data sources with the same connection string do not share connections. **NOTE that since `IDbConnection` objects are not thread-safe, lock objects constructed with them can only be used by one thread at a time.**
 
 ## Options
 

src/DistributedLock.Core/Internal/Data/MultiplexedConnectionLockPool.cs

@@ -8,31 +8,33 @@ namespace Medallion.Threading.Internal.Data;
 #else
 internal 
 #endif
-    sealed class MultiplexedConnectionLockPool
+    sealed class MultiplexedConnectionLockPool<TConnectionSource>
+    where TConnectionSource : notnull
 {
     private readonly AsyncLock _lock = AsyncLock.Create();
 
-    private readonly Dictionary<string, Queue<MultiplexedConnectionLock>> _poolsByConnectionString = [];
+    private readonly Dictionary<TConnectionSource, Queue<MultiplexedConnectionLock>> _poolsByConnectionSource;
 
     /// <summary>
-    /// The number of times we've called <see cref="StoreOrDisposeLockAsync(string, MultiplexedConnectionLock, bool)"/>
+    /// The number of times we've called <see cref="StoreOrDisposeLockAsync(TConnectionSource, MultiplexedConnectionLock, bool)"/>
     /// since we last called <see cref="PrunePoolsNoLockAsync"/>
     /// </summary>
     private uint _storeCountSinceLastPrune;
     /// <summary>
-    /// The number of <see cref="MultiplexedConnectionLock"/>s stored in <see cref="_poolsByConnectionString"/>
+    /// The number of <see cref="MultiplexedConnectionLock"/>s stored in <see cref="_poolsByConnectionSource"/>
     /// </summary>
     private uint _pooledLockCount;
 
-    public MultiplexedConnectionLockPool(Func<string, DatabaseConnection> connectionFactory) 
+    public MultiplexedConnectionLockPool(Func<TConnectionSource, DatabaseConnection> connectionFactory, IEqualityComparer<TConnectionSource>? comparer = null)
     {
         this.ConnectionFactory = connectionFactory;
+        this._poolsByConnectionSource = new(comparer);
     }
 
-    internal Func<string, DatabaseConnection> ConnectionFactory { get; }
+    internal Func<TConnectionSource, DatabaseConnection> ConnectionFactory { get; }
 
     public async ValueTask<IDistributedSynchronizationHandle?> TryAcquireAsync<TLockCookie>(
-        string connectionString,
+        TConnectionSource connectionSource,
         string name,
         TimeoutValue timeout,
         IDbSynchronizationStrategy<TLockCookie> strategy,
@@ -42,7 +44,7 @@ public MultiplexedConnectionLockPool(Func<string, DatabaseConnection> connection
     {
         // opportunistic phase: see if we can use a connection that is already holding a lock
         // to acquire the current lock
-        var existingLock = await this.GetExistingLockOrDefaultAsync(connectionString).ConfigureAwait(false);
+        var existingLock = await this.GetExistingLockOrDefaultAsync(connectionSource).ConfigureAwait(false);
         if (existingLock != null)
         {
             var canSafelyDisposeExistingLock = false;
@@ -70,12 +72,12 @@ public MultiplexedConnectionLockPool(Func<string, DatabaseConnection> connection
             finally
             {
                 // since we took this lock from the pool, always return it to the pool
-                await this.StoreOrDisposeLockAsync(connectionString, existingLock, shouldDispose: canSafelyDisposeExistingLock).ConfigureAwait(false);
+                await this.StoreOrDisposeLockAsync(connectionSource, existingLock, shouldDispose: canSafelyDisposeExistingLock).ConfigureAwait(false);
             }
         }
 
         // normal phase: if we were not able to be opportunistic, ensure that we have a lock
-        var @lock = new MultiplexedConnectionLock(this.ConnectionFactory(connectionString));
+        var @lock = new MultiplexedConnectionLock(this.ConnectionFactory(connectionSource));
         MultiplexedConnectionLock.Result? result = null;
         try
         {
@@ -85,19 +87,19 @@ public MultiplexedConnectionLockPool(Func<string, DatabaseConnection> connection
         finally
         {
             // if we failed to even acquire a result on a brand new lock, then there's definitely no reason to store it
-            await this.StoreOrDisposeLockAsync(connectionString, @lock, shouldDispose: result?.CanSafelyDispose ?? true).ConfigureAwait(false);
+            await this.StoreOrDisposeLockAsync(connectionSource, @lock, shouldDispose: result?.CanSafelyDispose ?? true).ConfigureAwait(false);
         }
         return result.Value.Handle;
 
         ValueTask<MultiplexedConnectionLock.Result> TryAcquireAsync(MultiplexedConnectionLock @lock, bool opportunistic) =>
             @lock.TryAcquireAsync(name, timeout, strategy, keepaliveCadence, cancellationToken, opportunistic);
     }
 
-    private async ValueTask<MultiplexedConnectionLock?> GetExistingLockOrDefaultAsync(string connectionString)
+    private async ValueTask<MultiplexedConnectionLock?> GetExistingLockOrDefaultAsync(TConnectionSource connectionSource)
     {
         using var _ = await this._lock.AcquireAsync(CancellationToken.None).ConfigureAwait(false);
 
-        if (this._poolsByConnectionString.TryGetValue(connectionString, out var pool) && pool.Count != 0)
+        if (this._poolsByConnectionSource.TryGetValue(connectionSource, out var pool) && pool.Count != 0)
         {
             --this._pooledLockCount;
             return pool.Dequeue();
@@ -106,7 +108,7 @@ public MultiplexedConnectionLockPool(Func<string, DatabaseConnection> connection
         return null;
     }
 
-    private async ValueTask StoreOrDisposeLockAsync(string connectionString, MultiplexedConnectionLock @lock, bool shouldDispose)
+    private async ValueTask StoreOrDisposeLockAsync(TConnectionSource connectionSource, MultiplexedConnectionLock @lock, bool shouldDispose)
     {
         if (shouldDispose)
         {
@@ -121,26 +123,26 @@ private async ValueTask StoreOrDisposeLockAsync(string connectionString, Multipl
             if (shouldDispose)
             {
                 // If we're about to dispose the lock, check if it has an empty pool that can be removed from our dictionary.
-                // By itself this doesn't guarantee cleanup: after a successful acquire we'll have an empty lock left over that won't 
-                // go away unless we use THAT connection string again. To help with this, we have pruning
-                if (this._poolsByConnectionString.TryGetValue(connectionString, out var pool) && pool.Count == 0)
+                // By itself this doesn't guarantee cleanup: after a successful acquire we'll have an empty lock left over that won't
+                // go away unless we use THAT connection source again. To help with this, we have pruning
+                if (this._poolsByConnectionSource.TryGetValue(connectionSource, out var pool) && pool.Count == 0)
                 {
-                    this._poolsByConnectionString.Remove(connectionString);
+                    this._poolsByConnectionSource.Remove(connectionSource);
                 }
             }
             else // otherwise, store the lock
             {
                 ++this._pooledLockCount;
 
-                if (this._poolsByConnectionString.TryGetValue(connectionString, out var existing))
+                if (this._poolsByConnectionSource.TryGetValue(connectionSource, out var existing))
                 {
                     existing.Enqueue(@lock);
                 }
                 else
                 {
                     var newPool = new Queue<MultiplexedConnectionLock>();
                     newPool.Enqueue(@lock);
-                    this._poolsByConnectionString.Add(connectionString, newPool);
+                    this._poolsByConnectionSource.Add(connectionSource, newPool);
                 }
             }
 
@@ -160,16 +162,16 @@ private bool IsDueForPruningNoLock()
         // The whole reason to prune is to avoid memory bloat (connection bloat isn't an issue since we only keep connections
         // open when needed). So, we don't even consider pruning below a certain storage threshold
 
-        var pruningCost = this._pooledLockCount + this._poolsByConnectionString.Count;
+        var pruningCost = this._pooledLockCount + this._poolsByConnectionSource.Count;
         return pruningCost > 64 && this._storeCountSinceLastPrune >= pruningCost;
     }
 
     private async ValueTask PrunePoolsNoLockAsync()
     {
         this._storeCountSinceLastPrune = 0; // reset
 
-        List<string>? connectionStringsToRemove = null; 
-        foreach (var kvp in this._poolsByConnectionString)
+        List<TConnectionSource>? connectionSourcesToRemove = null;
+        foreach (var kvp in this._poolsByConnectionSource)
         {
             var pool = kvp.Value;
             MultiplexedConnectionLock? firstRetainedLock = null;
@@ -191,15 +193,15 @@ private async ValueTask PrunePoolsNoLockAsync()
 
             if (pool.Count == 0)
             {
-                (connectionStringsToRemove ??= new List<string>()).Add(kvp.Key);
+                (connectionSourcesToRemove ??= new List<TConnectionSource>()).Add(kvp.Key);
             }
         }
 
-        if (connectionStringsToRemove != null)
+        if (connectionSourcesToRemove != null)
         {
-            foreach (var connectionStringToRemove in connectionStringsToRemove)
+            foreach (var connectionSourceToRemove in connectionSourcesToRemove)
             {
-                this._poolsByConnectionString.Remove(connectionStringToRemove);
+                this._poolsByConnectionSource.Remove(connectionSourceToRemove);
             }
         }
     }

src/DistributedLock.Core/Internal/Data/OptimisticConnectionMultiplexingDbDistributedLock.cs

@@ -8,26 +8,28 @@ namespace Medallion.Threading.Internal.Data;
 #else
 internal 
 #endif
-sealed class OptimisticConnectionMultiplexingDbDistributedLock : IDbDistributedLock
+sealed class OptimisticConnectionMultiplexingDbDistributedLock<TConnectionSource> : IDbDistributedLock
+    where TConnectionSource : notnull
 {
-    private readonly string _name, _connectionString;
-    private readonly MultiplexedConnectionLockPool _multiplexedConnectionLockPool;
+    private readonly string _name;
+    private readonly TConnectionSource _connectionSource;
+    private readonly MultiplexedConnectionLockPool<TConnectionSource> _multiplexedConnectionLockPool;
     private readonly TimeoutValue _keepaliveCadence;
     private readonly IDbDistributedLock _fallbackLock;
 
     public OptimisticConnectionMultiplexingDbDistributedLock(
-        string name, 
-        string connectionString, 
-        MultiplexedConnectionLockPool multiplexedConnectionLockPool,
+        string name,
+        TConnectionSource connectionSource,
+        MultiplexedConnectionLockPool<TConnectionSource> multiplexedConnectionLockPool,
         TimeoutValue keepaliveCadence)
     {
         this._name = name;
-        this._connectionString = connectionString;
+        this._connectionSource = connectionSource;
         this._multiplexedConnectionLockPool = multiplexedConnectionLockPool;
         this._keepaliveCadence = keepaliveCadence;
         this._fallbackLock = new DedicatedConnectionOrTransactionDbDistributedLock(
-            name, 
-            () => this._multiplexedConnectionLockPool.ConnectionFactory(this._connectionString),
+            name,
+            () => this._multiplexedConnectionLockPool.ConnectionFactory(this._connectionSource),
             useTransaction: false,
             keepaliveCadence: keepaliveCadence
         );
@@ -44,7 +46,7 @@ public OptimisticConnectionMultiplexingDbDistributedLock(
         // to an exclusive lock which asks for a long timeout
         if (!strategy.IsUpgradeable && contextHandle == null)
         {
-            return this._multiplexedConnectionLockPool.TryAcquireAsync(this._connectionString, this._name, timeout, strategy, keepaliveCadence: this._keepaliveCadence, cancellationToken);
+            return this._multiplexedConnectionLockPool.TryAcquireAsync(this._connectionSource, this._name, timeout, strategy, keepaliveCadence: this._keepaliveCadence, cancellationToken);
         }
 
         // otherwise, fall back to our fallback lock

src/DistributedLock.MySql/DistributedLock.MySql.csproj

@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
-    <TargetFrameworks>netstandard2.0;netstandard2.1;net462</TargetFrameworks>
+    <TargetFrameworks>netstandard2.0;netstandard2.1;net462;net8.0</TargetFrameworks>
     <RootNamespace>Medallion.Threading.MySql</RootNamespace>
     <GenerateDocumentationFile>True</GenerateDocumentationFile>
     <WarningLevel>4</WarningLevel>
@@ -44,6 +44,11 @@
     <DefineConstants>TRACE;DEBUG</DefineConstants>
   </PropertyGroup>
 
+  <ItemGroup Condition="'$(TargetFramework)' == 'net8.0'">
+    <AdditionalFiles Include="PublicAPI/$(TargetFramework)/PublicAPI.Shipped.txt" />
+    <AdditionalFiles Include="PublicAPI/$(TargetFramework)/PublicAPI.Unshipped.txt" />
+  </ItemGroup>
+
   <ItemGroup>
     <!-- Used as the default over MySql.Data because of better licensing -->
     <PackageReference Include="MySqlConnector" />

src/DistributedLock.MySql/MySqlDatabaseConnection.cs

@@ -1,6 +1,9 @@
 using Medallion.Threading.Internal.Data;
 using MySqlConnector;
 using System.Data;
+#if NET7_0_OR_GREATER
+using System.Data.Common;
+#endif
 
 namespace Medallion.Threading.MySql;
 
@@ -16,6 +19,13 @@ public MySqlDatabaseConnection(IDbTransaction transaction)
     {
     }
 
+#if NET7_0_OR_GREATER
+    public MySqlDatabaseConnection(DbDataSource dbDataSource)
+        : base(dbDataSource.CreateConnection(), isExternallyOwned: false)
+    {
+    }
+#endif
+
     public MySqlDatabaseConnection(string connectionString)
         : base(new MySqlConnection(connectionString), isExternallyOwned: false)
     {