[Providers.Azure] Move away from deprecated libs and support TokenCredential

src/providers/WorkflowCore.Providers.Azure/Models/ControlledLock.cs

@@ -1,15 +1,14 @@
-using System;
-using Microsoft.WindowsAzure.Storage.Blob;
+using Azure.Storage.Blobs.Specialized;
 
 namespace WorkflowCore.Providers.Azure.Models
 {
     class ControlledLock
     {
         public string Id { get; set; }
         public string LeaseId { get; set; }
-        public CloudBlockBlob Blob { get; set; }
+        public BlockBlobClient Blob { get; set; }
 
-        public ControlledLock(string id, string leaseId, CloudBlockBlob blob)
+        public ControlledLock(string id, string leaseId, BlockBlobClient blob)
         {
             Id = id;
             LeaseId = leaseId;

src/providers/WorkflowCore.Providers.Azure/ServiceCollectionExtensions.cs

@@ -1,4 +1,6 @@
-using Microsoft.Extensions.Logging;
+using System;
+using Azure.Core;
+using Microsoft.Extensions.Logging;
 using WorkflowCore.Interface;
 using WorkflowCore.Models;
 using WorkflowCore.Providers.Azure.Interface;
@@ -15,6 +17,13 @@ public static WorkflowOptions UseAzureSynchronization(this WorkflowOptions optio
             return options;
         }
 
+        public static WorkflowOptions UseAzureSynchronization(this WorkflowOptions options, Uri blobEndpoint, Uri queueEndpoint, TokenCredential tokenCredential)
+        {
+            options.UseQueueProvider(sp => new AzureStorageQueueProvider(queueEndpoint, tokenCredential, sp.GetService<ILoggerFactory>()));
+            options.UseDistributedLockManager(sp => new AzureLockManager(blobEndpoint, tokenCredential, sp.GetService<ILoggerFactory>()));
+            return options;
+        }
+
         public static WorkflowOptions UseAzureServiceBusEventHub(
             this WorkflowOptions options,
             string connectionString,
@@ -27,6 +36,19 @@ public static WorkflowOptions UseAzureSynchronization(this WorkflowOptions optio
             return options;
         }
 
+        public static WorkflowOptions UseAzureServiceBusEventHub(
+            this WorkflowOptions options,
+            string fullyQualifiedNamespace,
+            TokenCredential tokenCredential,
+            string topicName,
+            string subscriptionName)
+        {
+            options.UseEventHub(sp => new ServiceBusLifeCycleEventHub(
+                fullyQualifiedNamespace, tokenCredential, topicName, subscriptionName, sp.GetService<ILoggerFactory>()));
+
+            return options;
+        }
+
         public static WorkflowOptions UseCosmosDbPersistence(
             this WorkflowOptions options,
             string connectionString,
@@ -44,5 +66,24 @@ public static WorkflowOptions UseAzureSynchronization(this WorkflowOptions optio
             options.UsePersistence(sp => new CosmosDbPersistenceProvider(sp.GetService<ICosmosClientFactory>(), databaseId, sp.GetService<ICosmosDbProvisioner>(), cosmosDbStorageOptions));
             return options;
         }
+
+        public static WorkflowOptions UseCosmosDbPersistence(
+            this WorkflowOptions options,
+            string accountEndpoint,
+            TokenCredential tokenCredential,
+            string databaseId,
+            CosmosDbStorageOptions cosmosDbStorageOptions = null)
+        {
+            if (cosmosDbStorageOptions == null)
+            {
+                cosmosDbStorageOptions = new CosmosDbStorageOptions();
+            }
+
+            options.Services.AddSingleton<ICosmosClientFactory>(sp => new CosmosClientFactory(accountEndpoint, tokenCredential));
+            options.Services.AddTransient<ICosmosDbProvisioner>(sp => new CosmosDbProvisioner(sp.GetService<ICosmosClientFactory>(), cosmosDbStorageOptions));
+            options.Services.AddSingleton<IWorkflowPurger>(sp => new WorkflowPurger(sp.GetService<ICosmosClientFactory>(), databaseId, cosmosDbStorageOptions));
+            options.UsePersistence(sp => new CosmosDbPersistenceProvider(sp.GetService<ICosmosClientFactory>(), databaseId, sp.GetService<ICosmosDbProvisioner>(), cosmosDbStorageOptions));
+            return options;
+        }
     }
 }

src/providers/WorkflowCore.Providers.Azure/Services/AzureLockManager.cs

@@ -1,50 +1,57 @@
 using System;
 using System.Collections.Generic;
+using System.IO;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+using Azure.Core;
+using Azure.Storage.Blobs;
+using Azure.Storage.Blobs.Specialized;
 using Microsoft.Extensions.Logging;
-using Microsoft.WindowsAzure.Storage;
-using Microsoft.WindowsAzure.Storage.Blob;
 using WorkflowCore.Interface;
 using WorkflowCore.Providers.Azure.Models;
 
 namespace WorkflowCore.Providers.Azure.Services
 {
     public class AzureLockManager: IDistributedLockProvider
     {
-        private readonly CloudBlobClient _client;        
+        private readonly BlobServiceClient _client;
         private readonly ILogger _logger;
         private readonly List<ControlledLock> _locks = new List<ControlledLock>();
         private readonly AutoResetEvent _mutex = new AutoResetEvent(true);
-        private CloudBlobContainer _container;
+        private BlobContainerClient _container;
         private Timer _renewTimer;
         private TimeSpan LockTimeout => TimeSpan.FromMinutes(1);
         private TimeSpan RenewInterval => TimeSpan.FromSeconds(45);
 
         public AzureLockManager(string connectionString, ILoggerFactory logFactory)
         {
             _logger = logFactory.CreateLogger<AzureLockManager>();
-            var account = CloudStorageAccount.Parse(connectionString);
-            _client = account.CreateCloudBlobClient();
+            _client = new BlobServiceClient(connectionString);
+        }
+
+        public AzureLockManager(Uri blobEndpoint, TokenCredential tokenCredential, ILoggerFactory logFactory)
+        {
+            _logger = logFactory.CreateLogger<AzureLockManager>();
+            _client = new BlobServiceClient(blobEndpoint, tokenCredential);
         }
 
         public async Task<bool> AcquireLock(string Id, CancellationToken cancellationToken)
         {
-            var blob = _container.GetBlockBlobReference(Id);
+            var blob = _container.GetBlockBlobClient(Id);
 
             if (!await blob.ExistsAsync())
-                await blob.UploadTextAsync(string.Empty);
+                await blob.UploadAsync(new MemoryStream());
 
             if (_mutex.WaitOne())
             {
                 try
                 {
-                    var leaseId = await blob.AcquireLeaseAsync(LockTimeout);
-                    _locks.Add(new ControlledLock(Id, leaseId, blob));                    
+                    var lease = await blob.GetBlobLeaseClient().AcquireAsync(LockTimeout);
+                    _locks.Add(new ControlledLock(Id, lease.Value.LeaseId, blob));
                     return true;
                 }
-                catch (StorageException ex)
+                catch (Exception ex)
                 {
                     _logger.LogDebug($"Failed to acquire lock {Id} - {ex.Message}");
                     return false;
@@ -69,7 +76,7 @@ public async Task ReleaseLock(string Id)
                     {
                         try
                         {
-                            await entry.Blob.ReleaseLeaseAsync(AccessCondition.GenerateLeaseCondition(entry.LeaseId));
+                            await entry.Blob.GetBlobLeaseClient(entry.LeaseId).ReleaseAsync();
                         }
                         catch (Exception ex)
                         {
@@ -87,7 +94,7 @@ public async Task ReleaseLock(string Id)
 
         public async Task Start()
         {
-            _container = _client.GetContainerReference("workflowcore-locks");
+            _container = _client.GetBlobContainerClient("workflowcore-locks");
             await _container.CreateIfNotExistsAsync();
             _renewTimer = new Timer(RenewLeases, null, RenewInterval, RenewInterval);
         }
@@ -128,7 +135,7 @@ private async Task RenewLock(ControlledLock entry)
         {
             try
             {
-                await entry.Blob.RenewLeaseAsync(AccessCondition.GenerateLeaseCondition(entry.LeaseId));
+                await entry.Blob.GetBlobLeaseClient(entry.LeaseId).RenewAsync();
             }
             catch (Exception ex)
             {

src/providers/WorkflowCore.Providers.Azure/Services/AzureStorageQueueProvider.cs

@@ -2,9 +2,9 @@
 using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
+using Azure.Core;
+using Azure.Storage.Queues;
 using Microsoft.Extensions.Logging;
-using Microsoft.WindowsAzure.Storage;
-using Microsoft.WindowsAzure.Storage.Queue;
 using WorkflowCore.Interface;
 
 namespace WorkflowCore.Providers.Azure.Services
@@ -13,41 +13,44 @@ public class AzureStorageQueueProvider : IQueueProvider
     {
         private readonly ILogger _logger;
 
-        private readonly Dictionary<QueueType, CloudQueue> _queues = new Dictionary<QueueType, CloudQueue>();
+        private readonly Dictionary<QueueType, QueueClient> _queues = new Dictionary<QueueType, QueueClient>();
 
         public bool IsDequeueBlocking => false;
 
         public AzureStorageQueueProvider(string connectionString, ILoggerFactory logFactory)
         {
             _logger = logFactory.CreateLogger<AzureStorageQueueProvider>();
-            var account = CloudStorageAccount.Parse(connectionString);
-            var client = account.CreateCloudQueueClient();
+            var client = new QueueServiceClient(connectionString);
 
-            _queues[QueueType.Workflow] = client.GetQueueReference("workflowcore-workflows");
-            _queues[QueueType.Event] = client.GetQueueReference("workflowcore-events");
-            _queues[QueueType.Index] = client.GetQueueReference("workflowcore-index");
+            _queues[QueueType.Workflow] = client.GetQueueClient("workflowcore-workflows");
+            _queues[QueueType.Event] = client.GetQueueClient("workflowcore-events");
+            _queues[QueueType.Index] = client.GetQueueClient("workflowcore-index");
+        }
+
+        public AzureStorageQueueProvider(Uri queueEndpoint, TokenCredential tokenCredential, ILoggerFactory logFactory)
+        {
+            _logger = logFactory.CreateLogger<AzureStorageQueueProvider>();
+            var client = new QueueServiceClient(queueEndpoint, tokenCredential);
+
+            _queues[QueueType.Workflow] = client.GetQueueClient("workflowcore-workflows");
+            _queues[QueueType.Event] = client.GetQueueClient("workflowcore-events");
+            _queues[QueueType.Index] = client.GetQueueClient("workflowcore-index");
         }
 
         public async Task QueueWork(string id, QueueType queue)
         {
-            var msg = new CloudQueueMessage(id);
-            await _queues[queue].AddMessageAsync(msg);
+            await _queues[queue].SendMessageAsync(id);
         }
 
         public async Task<string> DequeueWork(QueueType queue, CancellationToken cancellationToken)
         {
-            CloudQueue cloudQueue = _queues[queue];
-
-            if (cloudQueue == null)
-                return null;
-
-            var msg = await cloudQueue.GetMessageAsync();
+            var msg = await _queues[queue].ReceiveMessageAsync();
 
-            if (msg == null)
+            if (msg == null || msg.Value == null)
                 return null;
 
-            await cloudQueue.DeleteMessageAsync(msg);
-            return msg.AsString;
+            await _queues[queue].DeleteMessageAsync(msg.Value.MessageId, msg.Value.PopReceipt);
+            return msg.Value.Body.ToString();
         }
 
         public async Task Start()

src/providers/WorkflowCore.Providers.Azure/Services/CosmosClientFactory.cs

@@ -1,4 +1,5 @@
 using System;
+using Azure.Core;
 using Microsoft.Azure.Cosmos;
 using WorkflowCore.Providers.Azure.Interface;
 
@@ -15,6 +16,11 @@ public CosmosClientFactory(string connectionString)
             _client = new CosmosClient(connectionString);
         }
 
+        public CosmosClientFactory(string accountEndpoint, TokenCredential tokenCredential)
+        {
+            _client = new CosmosClient(accountEndpoint, tokenCredential);
+        }
+
         public CosmosClient GetCosmosClient()
         {
             return this._client;