Adding RaiseEvent to client for scalability, Allow creating queue with specific size

src/DurableTask.Framework/IOrchestrationServiceClient.cs

@@ -33,9 +33,9 @@ public interface IOrchestrationServiceClient
         /// <summary>
         /// Send a new message for an orchestration
         /// </summary>
-        /// <param name="message">Message to send</param>
+        /// <param name="messages">List of messages to send</param>
         /// <returns></returns>
-        Task SendTaskOrchestrationMessageAsync(TaskMessage message);
+        Task SendTaskOrchestrationMessageAsync(params TaskMessage[] messages);
 
         /// <summary>
         /// Wait for an orchestration to reach any terminal state within the given timeout

src/DurableTask.Framework/ServiceBusOrchestrationService.cs

@@ -222,13 +222,13 @@ public async Task CreateAsync(bool recreateInstanceStore)
             NamespaceManager namespaceManager = NamespaceManager.CreateFromConnectionString(connectionString);
 
             await Task.WhenAll(
-                this.SafeDeleteAndCreateQueueAsync(namespaceManager, orchestratorEntityName, true, true, Settings.MaxTaskOrchestrationDeliveryCount),
-                this.SafeDeleteAndCreateQueueAsync(namespaceManager, workerEntityName, false, false, Settings.MaxTaskActivityDeliveryCount)
+                this.SafeDeleteAndCreateQueueAsync(namespaceManager, orchestratorEntityName, true, true, Settings.MaxTaskOrchestrationDeliveryCount, Settings.MaxQueueSizeInMegabytes),
+                this.SafeDeleteAndCreateQueueAsync(namespaceManager, workerEntityName, false, false, Settings.MaxTaskActivityDeliveryCount, Settings.MaxQueueSizeInMegabytes)
                 );
 
             if (InstanceStore != null)
             {
-                await this.SafeDeleteAndCreateQueueAsync(namespaceManager, trackingEntityName, true, false, Settings.MaxTrackingDeliveryCount);
+                await this.SafeDeleteAndCreateQueueAsync(namespaceManager, trackingEntityName, true, false, Settings.MaxTrackingDeliveryCount, Settings.MaxQueueSizeInMegabytes);
                 await InstanceStore.InitializeStoreAsync(recreateInstanceStore);
             }
         }
@@ -241,12 +241,12 @@ public async Task CreateIfNotExistsAsync()
             NamespaceManager namespaceManager = NamespaceManager.CreateFromConnectionString(connectionString);
 
             await Task.WhenAll(
-                SafeCreateQueueAsync(namespaceManager, orchestratorEntityName, true, true, Settings.MaxTaskOrchestrationDeliveryCount),
-                SafeCreateQueueAsync(namespaceManager, workerEntityName, false, false, Settings.MaxTaskActivityDeliveryCount)
+                SafeCreateQueueAsync(namespaceManager, orchestratorEntityName, true, true, Settings.MaxTaskOrchestrationDeliveryCount, Settings.MaxQueueSizeInMegabytes),
+                SafeCreateQueueAsync(namespaceManager, workerEntityName, false, false, Settings.MaxTaskActivityDeliveryCount, Settings.MaxQueueSizeInMegabytes)
                 );
             if (InstanceStore != null)
             {
-                await SafeCreateQueueAsync(namespaceManager, trackingEntityName, true, false, Settings.MaxTrackingDeliveryCount);
+                await SafeCreateQueueAsync(namespaceManager, trackingEntityName, true, false, Settings.MaxTrackingDeliveryCount, Settings.MaxQueueSizeInMegabytes);
                 await InstanceStore.InitializeStoreAsync(false);
             }
         }
@@ -902,7 +902,10 @@ public Task AbandonTaskActivityWorkItemAsync(TaskActivityWorkItem workItem)
         {
             var message = GetAndDeleteBrokeredMessageForWorkItem(workItem);
             TraceHelper.Trace(TraceEventType.Information, $"Abandoning message {workItem?.Id}");
-            return message?.AbandonAsync();
+
+            return message == null 
+                ? Task.FromResult<object>(null) 
+                : message.AbandonAsync();
         }
 
         /// <summary>
@@ -969,29 +972,40 @@ public async Task UpdateJumpStartStoreAsync(TaskMessage creationMessage)
         }
 
         /// <summary>
-        ///    Send an orchestration message
+        ///    Sends a set of orchestration messages
         /// </summary>
-        /// <param name="message">The task message to be sent for the orchestration</param>
-        public async Task SendTaskOrchestrationMessageAsync(TaskMessage message)
+        /// <param name="messages">The task message to be sent for the orchestration</param>
+        public async Task SendTaskOrchestrationMessageAsync(params TaskMessage[] messages)
         {
-            BrokeredMessage brokeredMessage = await ServiceBusUtils.GetBrokeredMessageFromObjectAsync(
-                message,
-                Settings.MessageCompressionSettings,
-                Settings.MessageSettings,
-                message.OrchestrationInstance,
-                "SendTaskOrchestrationMessage",
-                this.BlobStore,
-                DateTime.MinValue);
+            if (messages.Length == 0)
+            {
+                return;
+            }
 
-            // Use duplicate detection of ExecutionStartedEvent by addin messageId
-            var executionStartedEvent = message.Event as ExecutionStartedEvent;
-            if (executionStartedEvent != null)
+            var brokeredMessages = new BrokeredMessage[messages.Length];
+            for (int i = 0; i < messages.Length; i++)
             {
-                brokeredMessage.MessageId = string.Format(CultureInfo.InvariantCulture, $"{executionStartedEvent.OrchestrationInstance.InstanceId}_{executionStartedEvent.OrchestrationInstance.ExecutionId}");
+                var message = messages[i];
+
+                brokeredMessages[i] = await ServiceBusUtils.GetBrokeredMessageFromObjectAsync(
+                    message,
+                    Settings.MessageCompressionSettings,
+                    Settings.MessageSettings,
+                    message.OrchestrationInstance,
+                    "SendTaskOrchestrationMessage",
+                    this.BlobStore,
+                    DateTime.MinValue);
+
+                // Use duplicate detection of ExecutionStartedEvent by addin messageId
+                var executionStartedEvent = message.Event as ExecutionStartedEvent;
+                if (executionStartedEvent != null)
+                {
+                    brokeredMessages[i].MessageId = $"{executionStartedEvent.OrchestrationInstance.InstanceId}_{executionStartedEvent.OrchestrationInstance.ExecutionId}";
+                }
             }
 
             MessageSender sender = await messagingFactory.CreateMessageSenderAsync(orchestratorEntityName).ConfigureAwait(false);
-            await sender.SendAsync(brokeredMessage).ConfigureAwait(false);
+            await sender.SendBatchAsync(brokeredMessages).ConfigureAwait(false);
             await sender.CloseAsync().ConfigureAwait(false);
         }
 
@@ -1471,13 +1485,14 @@ async Task SafeCreateQueueAsync(
             string path,
             bool requiresSessions,
             bool requiresDuplicateDetection,
-            int maxDeliveryCount)
+            int maxDeliveryCount,
+            long maxSizeInMegabytes)
         {
             await Utils.ExecuteWithRetries(async () =>
             {
                 try
                 {
-                await CreateQueueAsync(namespaceManager, path, requiresSessions, requiresDuplicateDetection, maxDeliveryCount);
+                await CreateQueueAsync(namespaceManager, path, requiresSessions, requiresDuplicateDetection, maxDeliveryCount, maxSizeInMegabytes);
                 }
                 catch (MessagingEntityAlreadyExistsException)
                 {
@@ -1491,25 +1506,35 @@ async Task SafeDeleteAndCreateQueueAsync(
             string path,
             bool requiresSessions,
             bool requiresDuplicateDetection,
-            int maxDeliveryCount)
+            int maxDeliveryCount,
+            long maxSizeInMegabytes)
         {
             await SafeDeleteQueueAsync(namespaceManager, path);
-            await SafeCreateQueueAsync(namespaceManager, path, requiresSessions, requiresDuplicateDetection, maxDeliveryCount);
+            await SafeCreateQueueAsync(namespaceManager, path, requiresSessions, requiresDuplicateDetection, maxDeliveryCount, maxSizeInMegabytes);
         }
 
+        private static readonly long[] ValidQueueSizes = { 1024L, 2048L, 3072L, 4096L, 5120L };
+
         async Task CreateQueueAsync(
             NamespaceManager namespaceManager,
             string path,
             bool requiresSessions,
             bool requiresDuplicateDetection,
-            int maxDeliveryCount)
+            int maxDeliveryCount,
+            long maxSizeInMegabytes)
         {
+            if (!ValidQueueSizes.Contains(maxSizeInMegabytes))
+            {
+                throw new ArgumentException($"The specified value {maxSizeInMegabytes} is invalid for the maximum queue size in megabytes.\r\nIt must be one of the following values:\r\n{string.Join(";", ValidQueueSizes)}", nameof(maxSizeInMegabytes));
+            }
+
             var description = new QueueDescription(path)
             {
                 RequiresSession = requiresSessions,
                 MaxDeliveryCount = maxDeliveryCount,
                 RequiresDuplicateDetection = requiresDuplicateDetection,
-                DuplicateDetectionHistoryTimeWindow = TimeSpan.FromHours(DuplicateDetectionWindowInHours)
+                DuplicateDetectionHistoryTimeWindow = TimeSpan.FromHours(DuplicateDetectionWindowInHours),
+                MaxSizeInMegabytes = maxSizeInMegabytes
             };
 
             await namespaceManager.CreateQueueAsync(description);

src/DurableTask.Framework/Settings/ServiceBusOrchestrationServiceSettings.cs

@@ -60,6 +60,11 @@ public ServiceBusOrchestrationServiceSettings()
         /// </summary>
         public int MaxTrackingDeliveryCount { get; set; }
 
+        /// <summary>
+        ///     Maximum queue size, in megabytes, for the service bus queues
+        /// </summary>
+        public long MaxQueueSizeInMegabytes { get; set; } = 1024;
+
         /// <summary>
         /// Gets the message prefetch count
         /// </summary>

src/DurableTask.Framework/TaskHubClient.cs

@@ -156,6 +156,69 @@ public async Task<OrchestrationInstance> CreateOrchestrationInstanceAsync(
             return orchestrationInstance;
         }
 
+        /// <summary>
+        ///     Creates an orchestration instance, and raises an event for it, which eventually causes the OnEvent() method in the
+        ///     orchestration to fire.
+        /// </summary>
+        /// <param name="orchestrationName">Name of the orchestration as specified by the ObjectCreator</param>
+        /// <param name="orchestrationVersion">Name of the orchestration as specified by the ObjectCreator</param>
+        /// <param name="instanceId">Instance id for the orchestration to be created, must be unique across the Task Hub</param>
+        /// <param name="eventName">Name of the event</param>
+        /// <param name="eventData">Data for the event</param>
+        public async Task RaiseEventAsync(
+            string orchestrationName, 
+            string orchestrationVersion, 
+            string instanceId,
+            string eventName, 
+            object eventData)
+        {
+            if (string.IsNullOrWhiteSpace(instanceId))
+            {
+                instanceId = Guid.NewGuid().ToString("N");
+            }
+
+            var orchestrationInstance = new OrchestrationInstance
+            {
+                InstanceId = instanceId,
+                ExecutionId = Guid.NewGuid().ToString("N"),
+            };
+
+            var startedEvent = new ExecutionStartedEvent(-1, null)
+            {
+                Tags = null,
+                Name = orchestrationName,
+                Version = orchestrationVersion,
+                OrchestrationInstance = orchestrationInstance
+            };
+
+            string serializedInput = defaultConverter.Serialize(eventData);
+
+            var taskMessages = new[]
+            {
+                new TaskMessage
+                {
+                    OrchestrationInstance = orchestrationInstance,
+                    Event = startedEvent
+                },
+                new TaskMessage
+                {
+                    OrchestrationInstance = new OrchestrationInstance
+                    {
+                        InstanceId = instanceId,
+
+                        // to ensure that the event gets raised on the running
+			            // orchestration instance, null the execution id
+			            // so that it will find out which execution
+			            // it should use for processing
+                        ExecutionId = null  
+                    },
+                    Event = new EventRaisedEvent(-1, serializedInput) {Name = eventName}
+                }
+            };
+
+            await this.serviceClient.SendTaskOrchestrationMessageAsync(taskMessages);
+        }
+
         /// <summary>
         ///     Raises an event in the specified orchestration instance, which eventually causes the OnEvent() method in the
         ///     orchestration to fire.

test/DurableTask.Framework.Tests/Mocks/LocalOrchestrationService.cs

@@ -189,9 +189,13 @@ public Task CreateTaskOrchestrationAsync(TaskMessage creationMessage)
             return Task.FromResult<object>(null);
         }
 
-        public Task SendTaskOrchestrationMessageAsync(TaskMessage message)
+        public Task SendTaskOrchestrationMessageAsync(params TaskMessage[] messages)
         {
-            this.orchestratorQueue.SendMessage(message);
+            foreach (var message in messages)
+            {
+                this.orchestratorQueue.SendMessage(message);
+            }
+
             return Task.FromResult<object>(null);
         }