Add the ability to cancel scheduled messages

src/Common/Helpers/ServiceBusHelper.cs

@@ -5337,7 +5337,7 @@ public string GetAddressRelativeToNamespace(string address)
 
         public ServiceBusHelper2 GetServiceBusHelper2()
         {
-            var serviceBusHelper2 = new ServiceBusHelper2();
+            var serviceBusHelper2 = new ServiceBusHelper2(writeToLog);
             serviceBusHelper2.ConnectionString = ConnectionString;
             serviceBusHelper2.TransportType = UseAmqpWebSockets
                 ? Azure.Messaging.ServiceBus.ServiceBusTransportType.AmqpWebSockets

src/ServiceBus/Helpers/CancelScheduledMessagesHelper.cs

@@ -0,0 +1,53 @@
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Threading;
+using System.Threading.Tasks;
+
+using Azure.Messaging.ServiceBus;
+
+using ServiceBusExplorer.Utilities.Helpers;
+
+namespace ServiceBusExplorer.ServiceBus.Helpers
+{
+    public static class CancelScheduledMessagesHelper
+    {
+        public static async Task CancelScheduledMessages(ServiceBusHelper2 serviceBusHelper,
+            string queueName, List<long> sequenceNumbersToCancel)
+        {
+            var client = serviceBusHelper.CreateServiceBusClient();
+
+            try
+            {
+                var sender = client.CreateSender(queueName);
+
+                serviceBusHelper.WriteToLog($"Starting cancellation of scheduled messages on queue {queueName}.");
+
+                var stopwatch = Stopwatch.StartNew();
+                var semaphore = new SemaphoreSlim(40); // As recommended by https://learn.microsoft.com/en-us/azure/service-bus-messaging/message-transfers-locks-settlement#settling-send-operations
+                var tasks = new List<Task>(sequenceNumbersToCancel.Count);
+
+                foreach (long sequenceNumber in sequenceNumbersToCancel)
+                {
+                    await semaphore.WaitAsync();
+                    tasks.Add(CancelScheduledMessageWithLogAsync(sender, sequenceNumber, serviceBusHelper.WriteToLog)
+                        .ContinueWith((t) => semaphore.Release()));
+                }
+
+                await Task.WhenAll(tasks);
+                stopwatch.Stop();
+                serviceBusHelper.WriteToLog($"Cancelled {sequenceNumbersToCancel.Count} scheduled message(s) in {stopwatch.ElapsedMilliseconds / 1000} seconds.");
+            }
+            finally
+            {
+               await client.DisposeAsync();
+            }
+        }
+
+        static async Task CancelScheduledMessageWithLogAsync(ServiceBusSender sender, 
+            long sequenceNumber, WriteToLogDelegate writeToLog)
+        {
+            await sender.CancelScheduledMessageAsync(sequenceNumber);
+            writeToLog($"Cancelled scheduled message with sequence number {sequenceNumber}.");
+        }
+    }
+}

src/ServiceBus/Helpers/ServiceBusHelper2.cs

@@ -20,22 +20,40 @@
 #endregion
 
 using System.Threading.Tasks;
+
 using Azure.Messaging.ServiceBus;
 using Azure.Messaging.ServiceBus.Administration;
 
+using ServiceBusExplorer.Utilities.Helpers;
+
 // ReSharper disable CheckNamespace
 namespace ServiceBusExplorer.ServiceBus.Helpers
 // ReSharper restore CheckNamespace
 {
     public class ServiceBusHelper2
     {
+        readonly WriteToLogDelegate writeToLog;
+
         public string ConnectionString { get; set; }
         public ServiceBusTransportType TransportType { get; set; }
 
+        public WriteToLogDelegate WriteToLog
+        {
+            get
+            {
+                return writeToLog;
+            }
+        }
+
+        public ServiceBusHelper2(WriteToLogDelegate writeToLog)
+        {
+            this.writeToLog = writeToLog;
+        }
+
         public bool ConnectionStringContainsEntityPath()
         {
             var connectionStringProperties = ServiceBusConnectionStringProperties.Parse(ConnectionString);
-            
+
             if (connectionStringProperties?.EntityPath != null)
             {
                 return true;
@@ -44,6 +62,18 @@ public bool ConnectionStringContainsEntityPath()
             return false;
         }
 
+        // 
+        /// <summary>
+        ///  It is the callers responsibility to dispose the returned ServiceBusClient object by calling DisposeAsync() on it.
+        /// </summary>
+        /// <returns>An Azure.Messaging.ServiceBus.ServiceBusClient</returns>
+        public ServiceBusClient CreateServiceBusClient()
+        {
+            return new ServiceBusClient(
+                ConnectionString,
+                new ServiceBusClientOptions { TransportType = this.TransportType });
+        }
+
         public async Task<bool> IsPremiumNamespace()
         {
             var administrationClient = new ServiceBusAdministrationClient(ConnectionString);