Durable functions - 'Function chaining' scenario support

Author: daxian-dbw

src/Durable/Actions.cs

@@ -0,0 +1,139 @@
+//
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+//
+
+using System;
+using System.Collections.Generic;
+
+namespace Microsoft.Azure.Functions.PowerShellWorker.Action
+{
+    /// <summary>
+    /// Action types
+    /// </summary>
+    public enum ActionType
+    {
+        /// <summary>
+        /// Call an activity function.
+        /// </summary>
+        CallActivity = 0,
+
+        /// <summary>
+        /// Call an activity function with retry.
+        /// </summary>
+        CallActivityWithRetry = 1,
+
+        /// <summary>
+        /// Call a sub-orchestration function.
+        /// </summary>
+        CallSubOrchestrator = 2,
+
+        /// <summary>
+        /// Call a sub-orchestration function with retry.
+        /// </summary>
+        CallSubOrchestratorWithRetry = 3,
+
+        /// <summary>
+        /// Run the orchestration function as a loop.
+        /// </summary>
+        ContinueAsNew = 4,
+
+        /// <summary>
+        /// Create a timer.
+        /// </summary>
+        CreateTimer = 5,
+
+        /// <summary>
+        /// Wait for an external event.
+        /// </summary>
+        WaitForExternalEvent = 6,
+    }
+
+    /// <summary>
+    /// Base class that represents an orchestration action.
+    /// </summary>
+    public abstract class AzAction
+    {
+        /// <summary>
+        /// Base constructor for creating an action.
+        /// </summary>
+        protected AzAction(ActionType actionType)
+        {
+            ActionType = actionType;
+        }
+
+        /// <summary>
+        /// Action type.
+        /// </summary>
+        public readonly ActionType ActionType;
+    }
+
+    /// <summary>
+    /// An orchestration action that represents calling an activity function.
+    /// </summary>
+    public class CallActivityAction : AzAction
+    {
+        /// <summary>
+        /// The activity function name.
+        /// </summary>
+        public readonly string FunctionName;
+        
+        /// <summary>
+        /// The input to the activity function.
+        /// </summary>
+        public readonly object Input;
+
+        /// <summary>
+        /// Constructor
+        /// </summary>
+        internal CallActivityAction(string functionName, object input) : base(ActionType.CallActivity)
+        {
+            FunctionName = functionName;
+            Input = input;
+        }
+    }
+
+    /// <summary>
+    /// An orchestration action that represents creating a timer.
+    /// </summary>
+    public class CreateTimerAction : AzAction
+    {
+        /// <summary>
+        /// Time to fire the timer.
+        /// </summary>
+        public readonly DateTime FireAt;
+
+        /// <summary>
+        /// Indicate if the timer is cancelled.
+        /// </summary>
+        public readonly bool IsCanceled;
+
+        /// <summary>
+        /// Constructor.
+        /// </summary>
+        internal CreateTimerAction(DateTime fireAt, bool isCanceled) : base(ActionType.CreateTimer)
+        {
+            FireAt = fireAt;
+            IsCanceled = isCanceled;
+        }
+    }
+
+    /// <summary>
+    /// An orchestration action that represents waiting for an external event.
+    /// </summary>
+    public class WaitForExternalEventAction : AzAction
+    {
+        /// <summary>
+        /// Name of the external event.
+        /// </summary>
+        public readonly string ExternalEventName;
+
+        /// <summary>
+        /// Constructor.
+        /// </summary>
+        internal WaitForExternalEventAction(string externalEventName) : base(ActionType.WaitForExternalEvent)
+        {
+            ExternalEventName = externalEventName;
+        }
+    }
+}

src/Durable/HistoryEvent.cs

@@ -0,0 +1,145 @@
+//
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+//
+
+using System;
+using System.Runtime.Serialization;
+
+namespace Microsoft.Azure.Functions.PowerShellWorker.History
+{
+    [DataContract]
+    internal class HistoryEvent
+    {
+        #region Common_Fields
+
+        [DataMember]
+        internal int EventId { get; set; }
+        [DataMember]
+        internal bool IsPlayed { get; set; }
+        [DataMember]
+        internal DateTime Timestamp { get; set; }
+        [DataMember]
+        internal EventType EventType { get; set; }
+
+        #endregion
+
+        #region Timer_Event_Fields
+
+        [DataMember]
+        internal DateTime FireAt { get; set; }
+        [DataMember]
+        internal int TimerId { get; set; }
+
+        #endregion
+
+        #region Overloaded_Fields
+
+        [DataMember]
+        internal int TaskScheduledId { get; set; }
+        [DataMember]
+        internal string Input { get; set; }
+        [DataMember]
+        internal string Name { get; set; }
+        [DataMember]
+        internal string Result { get; set; }
+
+        #endregion
+
+        // Internal used only
+        internal bool IsProcessed { get; set; }
+    }
+
+    internal enum EventType
+    {
+        /// <summary>
+        /// Orchestration execution has started event
+        /// </summary>
+        ExecutionStarted,
+
+        /// <summary>
+        /// Orchestration execution has completed event
+        /// </summary>
+        ExecutionCompleted,
+
+        /// <summary>
+        /// Orchestration execution has failed event
+        /// </summary>
+        ExecutionFailed,
+
+        /// <summary>
+        /// Orchestration was terminated event
+        /// </summary>
+        ExecutionTerminated,
+
+        /// <summary>
+        /// Task Activity scheduled event 
+        /// </summary>
+        TaskScheduled,
+
+        /// <summary>
+        /// Task Activity completed event
+        /// </summary>
+        TaskCompleted,
+
+        /// <summary>
+        /// Task Activity failed event
+        /// </summary>
+        TaskFailed,
+
+        /// <summary>
+        /// Sub Orchestration instance created event
+        /// </summary>
+        SubOrchestrationInstanceCreated,
+
+        /// <summary>
+        /// Sub Orchestration instance completed event
+        /// </summary>
+        SubOrchestrationInstanceCompleted,
+
+        /// <summary>
+        /// Sub Orchestration instance failed event
+        /// </summary>
+        SubOrchestrationInstanceFailed,
+
+        /// <summary>
+        /// Timer created event
+        /// </summary>
+        TimerCreated,
+
+        /// <summary>
+        /// Timer fired event
+        /// </summary>
+        TimerFired,
+
+        /// <summary>
+        /// Orchestration has started event
+        /// </summary>
+        OrchestratorStarted,
+
+        /// <summary>
+        /// Orchestration has completed event
+        /// </summary>
+        OrchestratorCompleted,
+
+        /// <summary>
+        /// External Event raised to orchestration event
+        /// </summary>
+        EventRaised,
+
+        /// <summary>
+        /// Orchestration Continued as new event
+        /// </summary>
+        ContinueAsNew,
+
+        /// <summary>
+        /// Generic event for tracking event existence
+        /// </summary>
+        GenericEvent,
+
+        /// <summary>
+        /// Orchestration state history event
+        /// </summary>
+        HistoryState,
+    }
+}

src/Durable/InvokeActivityFunctionCommand.cs

@@ -0,0 +1,84 @@
+//
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+//
+
+using System.Collections;
+using System.Collections.Generic;
+using System.Linq;
+using System.Management.Automation;
+using System.Threading;
+
+using Newtonsoft.Json;
+using Microsoft.Azure.Functions.PowerShellWorker.Action;
+using Microsoft.Azure.Functions.PowerShellWorker.History;
+
+namespace Microsoft.Azure.Functions.PowerShellWorker.Commands
+{
+    /// <summary>
+    /// Invoke a function asynchronously.
+    /// </summary>
+    [Cmdlet("Invoke", "ActivityFunctionAsync")]
+    public class InvokeActivityFunctionCommand : PSCmdlet
+    {
+        /// <summary>
+        /// Gets and sets the activity function name.
+        /// </summary>
+        [Parameter(Mandatory = true)]
+        public string FunctionName { get; set; }
+
+        /// <summary>
+        /// Gets and sets the input for an activity function.
+        /// </summary>
+        /// <remarks>
+        /// Copy the default value from durable-js, in case that it's a magic value for specifying no input value.
+        /// </remarks>
+        [Parameter]
+        [ValidateNotNull]
+        public object Input { get; set; } = "__activity__default";
+
+        // Used for waiting on the pipeline to be stopped.
+        private ManualResetEvent waitForStop = new ManualResetEvent(initialState: false);
+        private OrchestrationContext context;
+
+        /// <summary>
+        /// Implement the EndProcessing method.
+        /// </summary>
+        protected override void EndProcessing()
+        {
+            var privateData = (Hashtable)this.MyInvocation.MyCommand.Module.PrivateData;
+            context = (OrchestrationContext)privateData[SetFunctionInvocationContextCommand.ContextKey];
+
+            context.Actions.Add(new List<AzAction>() { new CallActivityAction(FunctionName, Input) });
+
+            HistoryEvent taskScheduled = context.History
+                .FirstOrDefault(e => e.EventType == EventType.TaskScheduled &&
+                                e.Name == FunctionName &&
+                                !e.IsProcessed);
+
+            HistoryEvent taskCompleted = taskScheduled == null ? null : context.History
+                .FirstOrDefault(e => e.EventType == EventType.TaskCompleted &&
+                                e.TaskScheduledId == taskScheduled.EventId);
+
+            if (taskCompleted != null)
+            {
+                taskScheduled.IsProcessed = true;
+                taskCompleted.IsProcessed = true;
+                WriteObject(taskCompleted.Result);
+            }
+            else
+            {
+                context.ActionEvent.Set();
+                waitForStop.WaitOne();
+            }
+        }
+
+        /// <summary>
+        /// Implement the StopProcessing method.
+        /// </summary>
+        protected override void StopProcessing()
+        {
+            waitForStop.Set();
+        }
+    }
+}