Update IDataStorage.MongoDB.cs (#1722)

* Update IDataStorage.MongoDB.cs

* Update IDataStorage.MongoDB.cs

Author: demorgi

src/DotNetCore.CAP.MongoDB/IDataStorage.MongoDB.cs

@@ -11,6 +11,7 @@
 using DotNetCore.CAP.Monitoring;
 using DotNetCore.CAP.Persistence;
 using DotNetCore.CAP.Serialization;
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using MongoDB.Bson;
 using MongoDB.Driver;
@@ -25,43 +26,52 @@ public class MongoDBDataStorage : IDataStorage
     private readonly IOptions<MongoDBOptions> _options;
     private readonly ISerializer _serializer;
     private readonly ISnowflakeId _snowflakeId;
+    private readonly ILogger _logger;
 
     public MongoDBDataStorage(
         IOptions<CapOptions> capOptions,
         IOptions<MongoDBOptions> options,
         IMongoClient client,
         ISerializer serializer,
-        ISnowflakeId snowflakeId)
+        ISnowflakeId snowflakeId,
+        ILogger<MongoDBDataStorage> logger)
     {
         _capOptions = capOptions;
         _options = options;
         _client = client;
         _database = _client.GetDatabase(_options.Value.DatabaseName);
         _serializer = serializer;
         _snowflakeId = snowflakeId;
+        _logger = logger;
     }
 
     public async Task<bool> AcquireLockAsync(string key, TimeSpan ttl, string instance,
         CancellationToken token = default)
     {
+
         var collection = _database.GetCollection<Lock>(_options.Value.LockCollection);
         using var session = await _client.StartSessionAsync(cancellationToken: token).ConfigureAwait(false);
         var transactionOptions =
             new TransactionOptions(ReadConcern.Majority, ReadPreference.Primary, WriteConcern.WMajority);
-        session.StartTransaction(transactionOptions);
+
         try
         {
-            var opResult = await collection.UpdateOneAsync(session,
-                model => model.Key == key && model.LastLockTime < DateTime.Now.Subtract(ttl),
-                Builders<Lock>.Update.Set(model => model.Instance, instance)
-                    .Set(model => model.LastLockTime, DateTime.Now), null, token);
-            var isAcquired = opResult.IsModifiedCountAvailable && opResult.ModifiedCount > 0;
-            await session.CommitTransactionAsync(token).ConfigureAwait(false);
-            return isAcquired;
+            var result = await session.WithTransactionAsync(async (handle, cancellationToken) =>
+            {
+                var opResult = await collection.UpdateOneAsync(handle,
+                    model => model.Key == key && model.LastLockTime < DateTime.Now.Subtract(ttl),
+                    Builders<Lock>.Update.Set(model => model.Instance, instance)
+                        .Set(model => model.LastLockTime, DateTime.Now), null, cancellationToken);
+                var isAcquired = opResult.IsModifiedCountAvailable && opResult.ModifiedCount > 0;
+                return isAcquired;
+            }, transactionOptions, token);
+
+            return result;
         }
-        catch (Exception)
+        catch (Exception ex)
         {
-            await session.AbortTransactionAsync(token).ConfigureAwait(false);
+            _logger.LogWarning(
+                ex, "Failed to acquire lock for key '{Key}' with instance '{Instance}'.", key, instance);
             return false;
         }
     }