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,85 @@
+using System;
+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
+    {
+        class OperationStatus
+        {
+            public int Successes;
+            public int Failures;
+        }
+
+        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 operationStatus = new OperationStatus();
+
+                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(CancelScheduledMessageWithLog(sender, sequenceNumber, serviceBusHelper.WriteToLog, operationStatus)
+                        .ContinueWith((t, state) => ((SemaphoreSlim)state)?.Release(), semaphore));
+                }
+
+                await Task.WhenAll(tasks);
+                stopwatch.Stop();
+
+                Func<int, string> singleOrPlural = (c) => c > 1 ? "messages" : "message";
+
+                serviceBusHelper.WriteToLog($"Successfully cancelled {operationStatus.Successes} " +
+                    $"scheduled {singleOrPlural(operationStatus.Successes)} in {stopwatch.Elapsed}.");
+
+                if (operationStatus.Failures > 0)
+                {
+                    serviceBusHelper.WriteToLog($"Failed to cancel {operationStatus.Failures} " +
+                        $"{singleOrPlural(operationStatus.Failures)}.");
+                }
+            }
+            finally
+            {
+                await client.DisposeAsync();
+            }
+        }
+
+        static Task CancelScheduledMessageWithLog(ServiceBusSender sender,
+            long sequenceNumber, WriteToLogDelegate writeToLog, OperationStatus operationStatus)
+        {
+            Task task = sender.CancelScheduledMessageAsync(sequenceNumber)
+                .ContinueWith(_ =>
+                    {
+                        writeToLog($"Cancelled scheduled message with sequence number {sequenceNumber}.");
+                        Interlocked.Increment(ref operationStatus.Successes);
+                    },
+                    TaskContinuationOptions.OnlyOnRanToCompletion | TaskContinuationOptions.ExecuteSynchronously)
+                .ContinueWith(_ =>
+                    {
+                        writeToLog($"Failed to cancel scheduled message with sequence number {sequenceNumber}.");
+                        Interlocked.Increment(ref operationStatus.Failures);
+                    },
+                    TaskContinuationOptions.NotOnRanToCompletion | TaskContinuationOptions.ExecuteSynchronously);
+
+            return task;
+        }
+    }
+}

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,17 @@ public bool ConnectionStringContainsEntityPath()
             return false;
         }
 
+        /// <summary>
+        ///  Dispose of the returned ServiceBusClient object by calling DisposeAsync().
+        /// </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);