Prepare Worker for External DF SDK support (and a few DF bug fixes)

src/DurableSDK/Commands/GetDurableTaskResult.cs

@@ -0,0 +1,36 @@
+//
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+//
+
+#pragma warning disable 1591 // Missing XML comment for publicly visible type or member 'member'
+
+namespace Microsoft.Azure.Functions.PowerShellWorker.Durable.Commands
+{
+    using System.Collections;
+    using System.Management.Automation;
+    using Microsoft.Azure.Functions.PowerShellWorker.Durable.Tasks;
+
+    [Cmdlet("Get", "DurableTaskResult")]
+    public class GetDurableTaskResultCommand : PSCmdlet
+    {
+        [Parameter(Mandatory = true)]
+        [ValidateNotNull]
+        public DurableTask[] Task { get; set; }
+
+        private readonly DurableTaskHandler _durableTaskHandler = new DurableTaskHandler();
+
+        protected override void EndProcessing()
+        {
+            var privateData = (Hashtable)MyInvocation.MyCommand.Module.PrivateData;
+            var context = (OrchestrationContext)privateData[SetFunctionInvocationContextCommand.ContextKey];
+
+            _durableTaskHandler.GetTaskResult(Task, context, WriteObject);
+        }
+
+        protected override void StopProcessing()
+        {
+            _durableTaskHandler.Stop();
+        }
+    }
+}

src/Durable/Commands/InvokeDurableActivityCommand.cs → src/DurableSDK/Commands/InvokeDurableActivityCommand.cs

@@ -43,10 +43,8 @@ protected override void EndProcessing()
         {
             var privateData = (Hashtable)MyInvocation.MyCommand.Module.PrivateData;
             var context = (OrchestrationContext)privateData[SetFunctionInvocationContextCommand.ContextKey];
-            var loadedFunctions = FunctionLoader.GetLoadedFunctions();
 
             var task = new ActivityInvocationTask(FunctionName, Input, RetryOptions);
-            ActivityInvocationTask.ValidateTask(task, loadedFunctions);
 
             _durableTaskHandler.StopAndInitiateDurableTaskOrReplay(
                 task, context, NoWait.IsPresent,

src/Durable/Commands/SetFunctionInvocationContextCommand.cs → src/DurableSDK/Commands/SetFunctionInvocationContextCommand.cs

@@ -9,6 +9,7 @@ namespace Microsoft.Azure.Functions.PowerShellWorker.Durable.Commands
 {
     using System.Collections;
     using System.Management.Automation;
+    using Microsoft.PowerShell.Commands;
 
     /// <summary>
     /// Set the orchestration context.

src/Durable/DurableTaskHandler.cs → src/DurableSDK/DurableTaskHandler.cs

@@ -6,10 +6,12 @@
 namespace Microsoft.Azure.Functions.PowerShellWorker.Durable
 {
     using System;
+    using System.Collections;
     using System.Collections.Generic;
+    using System.Management.Automation;
     using System.Threading;
     using Microsoft.Azure.Functions.PowerShellWorker.Durable.Tasks;
-    using Utility;
+    using Microsoft.PowerShell.Commands;
 
     internal class DurableTaskHandler
     {
@@ -47,6 +49,7 @@ public void StopAndInitiateDurableTaskOrReplay(
                     }
 
                     completedHistoryEvent.IsProcessed = true;
+                    context.IsReplaying = completedHistoryEvent.IsPlayed;
 
                     switch (completedHistoryEvent.EventType)
                     {
@@ -57,6 +60,13 @@ public void StopAndInitiateDurableTaskOrReplay(
                                 output(eventResult);
                             }
                             break;
+                        case HistoryEventType.EventRaised:
+                            var eventRaisedResult = GetEventResult(completedHistoryEvent);
+                            if (eventRaisedResult != null)
+                            {
+                                output(eventRaisedResult);
+                            }
+                            break;
 
                         case HistoryEventType.TaskFailed:
                             if (retryOptions == null)
@@ -76,7 +86,7 @@ public void StopAndInitiateDurableTaskOrReplay(
                                         retryOptions.MaxNumberOfAttempts,
                                         onSuccess:
                                             result => {
-                                                output(TypeExtensions.ConvertFromJson(result));
+                                                output(ConvertFromJson(result));
                                             },
                                         onFailure);
 
@@ -126,6 +136,7 @@ public void WaitAll(
             var allTasksCompleted = completedEvents.Count == tasksToWaitFor.Count;
             if (allTasksCompleted)
             {
+                context.IsReplaying = completedEvents.Count == 0 ? false : completedEvents[0].IsPlayed;
                 CurrentUtcDateTimeUpdater.UpdateCurrentUtcDateTime(context);
 
                 foreach (var completedHistoryEvent in completedEvents)
@@ -164,6 +175,7 @@ public void WaitAny(
                 if (scheduledHistoryEvent != null)
                 {
                     scheduledHistoryEvent.IsProcessed = true;
+                    scheduledHistoryEvent.IsPlayed = true;
                 }
 
                 if (completedHistoryEvent != null)
@@ -179,12 +191,14 @@ public void WaitAny(
                     }
 
                     completedHistoryEvent.IsProcessed = true;
+                    completedHistoryEvent.IsPlayed = true;
                 }
             }
 
             var anyTaskCompleted = completedTasks.Count > 0;
             if (anyTaskCompleted)
             {
+                context.IsReplaying = context.History[firstCompletedHistoryEventIndex].IsPlayed;
                 CurrentUtcDateTimeUpdater.UpdateCurrentUtcDateTime(context);
                 // Return a reference to the first completed task
                 output(firstCompletedTask);
@@ -195,6 +209,21 @@ public void WaitAny(
             }
         }
 
+        public void GetTaskResult(
+            IReadOnlyCollection<DurableTask> tasksToQueryResultFor,
+            OrchestrationContext context,
+            Action<object> output)
+        {
+            foreach (var task in tasksToQueryResultFor) {
+                var scheduledHistoryEvent = task.GetScheduledHistoryEvent(context, true);
+                var processedHistoryEvent = task.GetCompletedHistoryEvent(context, scheduledHistoryEvent, true);
+                if (processedHistoryEvent != null)
+                {
+                    output(GetEventResult(processedHistoryEvent));
+                }
+            }
+        }
+
         public void Stop()
         {
             _waitForStop.Set();
@@ -206,15 +235,41 @@ private static object GetEventResult(HistoryEvent historyEvent)
 
             if (historyEvent.EventType == HistoryEventType.TaskCompleted)
             {
-                return TypeExtensions.ConvertFromJson(historyEvent.Result);
+                return ConvertFromJson(historyEvent.Result);
             }
             else if (historyEvent.EventType == HistoryEventType.EventRaised)
             {
-                return TypeExtensions.ConvertFromJson(historyEvent.Input);
+                return ConvertFromJson(historyEvent.Input);
             }
             return null;
         }
 
+        public static object ConvertFromJson(string json)
+        {
+            object retObj = JsonObject.ConvertFromJson(json, returnHashtable: true, error: out _);
+
+            if (retObj is PSObject psObj)
+            {
+                retObj = psObj.BaseObject;
+            }
+
+            if (retObj is Hashtable hashtable)
+            {
+                try
+                {
+                    // ConvertFromJson returns case-sensitive Hashtable by design -- JSON may contain keys that only differ in case.
+                    // We try casting the Hashtable to a case-insensitive one, but if that fails, we keep using the original one.
+                    retObj = new Hashtable(hashtable, StringComparer.OrdinalIgnoreCase);
+                }
+                catch
+                {
+                    retObj = hashtable;
+                }
+            }
+
+            return retObj;
+        }
+
         private void InitiateAndWaitForStop(OrchestrationContext context)
         {
             context.OrchestrationActionCollector.Stop();

src/Durable/IOrchestrationInvoker.cs → src/DurableSDK/IOrchestrationInvoker.cs

@@ -5,10 +5,13 @@
 
 namespace Microsoft.Azure.Functions.PowerShellWorker.Durable
 {
+    using System;
     using System.Collections;
+    using System.Management.Automation;
 
     internal interface IOrchestrationInvoker
     {
         Hashtable Invoke(OrchestrationBindingInfo orchestrationBindingInfo, IPowerShellServices pwsh);
+        void SetExternalInvoker(Action<PowerShell> externalInvoker);
     }
 }

src/Durable/IPowerShellServices.cs → src/DurableSDK/IPowerShellServices.cs

@@ -5,17 +5,26 @@
 
 namespace Microsoft.Azure.Functions.PowerShellWorker.Durable
 {
+    using Microsoft.Azure.WebJobs.Script.Grpc.Messages;
     using System;
     using System.Management.Automation;
 
     internal interface IPowerShellServices
     {
+        PowerShell GetPowerShell();
+
+        bool UseExternalDurableSDK();
+
         void SetDurableClient(object durableClient);
 
-        void SetOrchestrationContext(OrchestrationContext orchestrationContext);
+        OrchestrationBindingInfo SetOrchestrationContext(ParameterBinding orchestrationContext, out Action<object> externalInvoker);
 
         void ClearOrchestrationContext();
 
+        public void TracePipelineObject();
+        public void AddParameter(string name, object value);
+
+
         IAsyncResult BeginInvoke(PSDataCollection<object> output);
 
         void EndInvoke(IAsyncResult asyncResult);

src/Durable/OrchestrationActionCollector.cs → src/DurableSDK/OrchestrationActionCollector.cs

@@ -11,6 +11,7 @@ namespace Microsoft.Azure.Functions.PowerShellWorker.Durable
     using System.Threading;
 
     using Microsoft.Azure.Functions.PowerShellWorker.Durable.Actions;
+    using Newtonsoft.Json;
 
     internal class OrchestrationActionCollector
     {

src/Durable/OrchestrationContext.cs → src/DurableSDK/OrchestrationContext.cs

@@ -20,13 +20,13 @@ public class OrchestrationContext
         public object Input { get; internal set; }
 
         [DataMember]
-        internal string InstanceId { get; set; }
+        public string InstanceId { get; set; }
 
         [DataMember]
         internal string ParentInstanceId { get; set; }
 
         [DataMember]
-        internal bool IsReplaying { get; set; }
+        public bool IsReplaying { get; set; }
 
         [DataMember]
         internal HistoryEvent[] History { get; set; }
@@ -35,6 +35,15 @@ public class OrchestrationContext
 
         internal OrchestrationActionCollector OrchestrationActionCollector { get; } = new OrchestrationActionCollector();
 
+        internal object ExternalResult;
+        internal bool ExternalIsError;
+
+        internal void SetExternalResult(object result, bool isError)
+        {
+            this.ExternalResult = result;
+            this.ExternalIsError = isError;
+        }
+
         internal object CustomStatus { get; set; }
     }
 }