.Net: Processes - Cloud Events supporting components + Demos of usabe

dotnet/samples/GettingStartedWithProcesses/Step03/Processes/FriedFishProcess.cs

@@ -6,6 +6,43 @@
 using Step03.Steps;
 namespace Step03.Processes;
 
+public enum FishProcessEvents
+{
+    PrepareFriedFish,
+    MiddleStep,
+    FriedFishFailed,
+    FriedFishReady,
+}
+
+public class FriedFishEventSubscribers : KernelProcessEventsSubscriber<FishProcessEvents>
+{
+    // TODO-estenori: figure out how to disallow and not need constructor on when using KernelProcessEventsSubscriber as base class
+    public FriedFishEventSubscribers(IServiceProvider? serviceProvider = null) : base(serviceProvider) { }
+
+    [ProcessEventSubscriber(FishProcessEvents.MiddleStep)]
+    public void OnMiddleStep(List<string> data)
+    {
+        // do something with data
+        Console.WriteLine($"=============> ON MIDDLE STEP: {data.FirstOrDefault() ?? ""}");
+    }
+
+    [ProcessEventSubscriber(FishProcessEvents.FriedFishReady)]
+    public void OnPrepareFish(object data)
+    {
+        // do something with data
+        // TODO: if event is linked to last event it doesnt get hit
+        // even when it may be linked to StopProcess() -> need additional special step?
+        Console.WriteLine("=============> ON FISH READY");
+    }
+
+    [ProcessEventSubscriber(FishProcessEvents.FriedFishFailed)]
+    public void OnFriedFisFailed(object data)
+    {
+        // do something with data
+        Console.WriteLine("=============> ON FISH FAILED");
+    }
+}
+
 /// <summary>
 /// Sample process that showcases how to create a process with sequential steps and reuse of existing steps.<br/>
 /// </summary>
@@ -26,20 +63,21 @@ public static class ProcessEvents
     /// </summary>
     /// <param name="processName">name of the process</param>
     /// <returns><see cref="ProcessBuilder"/></returns>
-    public static ProcessBuilder CreateProcess(string processName = "FriedFishProcess")
+    public static ProcessBuilder<FishProcessEvents> CreateProcess(string processName = "FriedFishProcess")
     {
-        var processBuilder = new ProcessBuilder(processName);
+        var processBuilder = new ProcessBuilder<FishProcessEvents>(processName);
 
         var gatherIngredientsStep = processBuilder.AddStepFromType<GatherFriedFishIngredientsStep>();
         var chopStep = processBuilder.AddStepFromType<CutFoodStep>();
         var fryStep = processBuilder.AddStepFromType<FryFoodStep>();
 
         processBuilder
-            .OnInputEvent(ProcessEvents.PrepareFriedFish)
+            .OnInputEvent(FishProcessEvents.PrepareFriedFish)
             .SendEventTo(new ProcessFunctionTargetBuilder(gatherIngredientsStep));
 
         gatherIngredientsStep
             .OnEvent(GatherFriedFishIngredientsStep.OutputEvents.IngredientsGathered)
+            .EmitAsProcessEvent(processBuilder.GetProcessEvent(FishProcessEvents.MiddleStep))
             .SendEventTo(new ProcessFunctionTargetBuilder(chopStep, functionName: CutFoodStep.Functions.ChopFood));
 
         chopStep
@@ -48,8 +86,14 @@ public static ProcessBuilder CreateProcess(string processName = "FriedFishProces
 
         fryStep
             .OnEvent(FryFoodStep.OutputEvents.FoodRuined)
+            .EmitAsProcessEvent(processBuilder.GetProcessEvent(FishProcessEvents.FriedFishFailed))
             .SendEventTo(new ProcessFunctionTargetBuilder(gatherIngredientsStep));
 
+        fryStep
+            .OnEvent(FryFoodStep.OutputEvents.FriedFoodReady)
+            .EmitAsProcessEvent(processBuilder.GetProcessEvent(FishProcessEvents.FriedFishReady))
+            .StopProcess();
+
         return processBuilder;
     }
 
@@ -81,23 +125,23 @@ public static ProcessBuilder CreateProcessWithStatefulStepsV1(string processName
         return processBuilder;
     }
 
-    /// <summary>
-    /// For a visual reference of the FriedFishProcess with stateful steps check this
-    /// <see href="https://github.com/microsoft/semantic-kernel/blob/main/dotnet/samples/GettingStartedWithProcesses/README.md#fried-fish-preparation-with-knife-sharpening-and-ingredient-stock-process" >diagram</see>
-    /// </summary>
-    /// <param name="processName">name of the process</param>
-    /// <returns><see cref="ProcessBuilder"/></returns>
-    public static ProcessBuilder CreateProcessWithStatefulStepsV2(string processName = "FriedFishWithStatefulStepsProcess")
+	/// <summary>
+	/// For a visual reference of the FriedFishProcess with stateful steps check this
+	/// <see href="https://github.com/microsoft/semantic-kernel/blob/main/dotnet/samples/GettingStartedWithProcesses/README.md#fried-fish-preparation-with-knife-sharpening-and-ingredient-stock-process" >diagram</see>
+	/// </summary>
+	/// <param name="processName">name of the process</param>
+	/// <returns><see cref="ProcessBuilder"/></returns>
+	public static ProcessBuilder CreateProcessWithStatefulStepsV2(string processName = "FriedFishWithStatefulStepsProcess")
     {
         // It is recommended to specify process version in case this process is used as a step by another process
-        var processBuilder = new ProcessBuilder(processName) { Version = "FriedFishProcess.v2" };
+        var processBuilder = new ProcessBuilder<FishProcessEvents>(processName) { Version = "FriedFishProcess.v2" };
 
         var gatherIngredientsStep = processBuilder.AddStepFromType<GatherFriedFishIngredientsWithStockStep>(name: "gatherFishIngredientStep", aliases: ["GatherFriedFishIngredientsWithStockStep"]);
         var chopStep = processBuilder.AddStepFromType<CutFoodWithSharpeningStep>(name: "chopFishStep", aliases: ["CutFoodStep"]);
         var fryStep = processBuilder.AddStepFromType<FryFoodStep>(name: "fryFishStep", aliases: ["FryFoodStep"]);
 
         processBuilder
-            .OnInputEvent(ProcessEvents.PrepareFriedFish)
+            .GetProcessEvent(FishProcessEvents.PrepareFriedFish)
             .SendEventTo(new ProcessFunctionTargetBuilder(gatherIngredientsStep));
 
         gatherIngredientsStep
@@ -106,6 +150,7 @@ public static ProcessBuilder CreateProcessWithStatefulStepsV2(string processName
 
         gatherIngredientsStep
             .OnEvent(GatherFriedFishIngredientsWithStockStep.OutputEvents.IngredientsOutOfStock)
+            .EmitAsProcessEvent(processBuilder.GetProcessEvent(FishProcessEvents.FriedFishFailed))
             .StopProcess();
 
         chopStep
@@ -124,6 +169,9 @@ public static ProcessBuilder CreateProcessWithStatefulStepsV2(string processName
             .OnEvent(FryFoodStep.OutputEvents.FoodRuined)
             .SendEventTo(new ProcessFunctionTargetBuilder(gatherIngredientsStep));
 
+        fryStep.OnEvent(FryFoodStep.OutputEvents.FriedFoodReady)
+            .EmitAsProcessEvent(processBuilder.GetProcessEvent(FishProcessEvents.FriedFishReady));
+
         return processBuilder;
     }
 

dotnet/samples/GettingStartedWithProcesses/Step03/Step03a_FoodPreparation.cs

@@ -22,6 +22,8 @@ public class Step03a_FoodPreparation(ITestOutputHelper output) : BaseTest(output
     public async Task UsePrepareFriedFishProcessAsync()
     {
         var process = FriedFishProcess.CreateProcess();
+        process.LinkEventSubscribersFromType<FriedFishEventSubscribers>();
+
         await UsePrepareSpecificProductAsync(process, FriedFishProcess.ProcessEvents.PrepareFriedFish);
     }
 

dotnet/src/Experimental/Process.Abstractions/KernelProcess.cs

@@ -9,13 +9,15 @@
 /// <summary>
 /// A serializable representation of a Process.
 /// </summary>
-public sealed record KernelProcess : KernelProcessStepInfo
+public sealed record KernelProcess : KernelProcessStepInfo // TODO: Should be renamed to KernelProcessInfo to keep consistent names
 {
     /// <summary>
     /// The collection of Steps in the Process.
     /// </summary>
     public IList<KernelProcessStepInfo> Steps { get; }
 
+    public KernelProcessEventsSubscriberInfo? EventsSubscriber { get; set; } = null;
+
     /// <summary>
     /// Captures Kernel Process State into <see cref="KernelProcessStateMetadata"/> after process has run
     /// </summary>
@@ -31,12 +33,14 @@
     /// <param name="state">The process state.</param>
     /// <param name="steps">The steps of the process.</param>
     /// <param name="edges">The edges of the process.</param>
-    public KernelProcess(KernelProcessState state, IList<KernelProcessStepInfo> steps, Dictionary<string, List<KernelProcessEdge>>? edges = null)
+    /// TODO: may need to reorder params
+    public KernelProcess(KernelProcessState state, IList<KernelProcessStepInfo> steps, Dictionary<string, List<KernelProcessEdge>>? edges = null, KernelProcessEventsSubscriberInfo? eventsSubscriber = null)
         : base(typeof(KernelProcess), state, edges ?? [])
     {
         Verify.NotNull(steps);
         Verify.NotNullOrWhiteSpace(state.Name);
 
         this.Steps = [.. steps];
+        this.EventsSubscriber = eventsSubscriber;
     }
 }

dotnet/src/Experimental/Process.Abstractions/KernelProcessEdge.cs

@@ -17,6 +17,12 @@
     [DataMember]
     public string SourceStepId { get; init; }
 
+    [DataMember]
+    public string SourceEventName { get; init; }
+
+    [DataMember]
+    public string SourceEventId { get; init; }
+
     /// <summary>
     /// The collection of <see cref="KernelProcessFunctionTarget"/>s that are the output of the source Step.
     /// </summary>
@@ -26,12 +32,16 @@
     /// <summary>
     /// Creates a new instance of the <see cref="KernelProcessEdge"/> class.
     /// </summary>
-    public KernelProcessEdge(string sourceStepId, KernelProcessFunctionTarget outputTarget)
+    public KernelProcessEdge(string sourceStepId, KernelProcessFunctionTarget outputTarget, string sourceEventName, string sourceEventId)
     {
         Verify.NotNullOrWhiteSpace(sourceStepId);
+        Verify.NotNullOrWhiteSpace(sourceEventId);
+        Verify.NotNullOrWhiteSpace(sourceEventName);
         Verify.NotNull(outputTarget);
 
         this.SourceStepId = sourceStepId;
+        this.SourceEventId = sourceEventId;
+        this.SourceEventName = sourceEventName;
         this.OutputTarget = outputTarget;
     }
 }

dotnet/src/Experimental/Process.Abstractions/KernelProcessEventsSubscriber.cs

@@ -0,0 +1,49 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+
+namespace Microsoft.SemanticKernel.Process;
+/// <summary>
+/// Attribute to set Process related steps to link Process Events to specific functions to execute when the event is emitted outside the Process
+/// </summary>
+/// <typeparam name="TEvents">Enum that contains all process events that could be subscribed to</typeparam>
+public class KernelProcessEventsSubscriber<TEvents> where TEvents : Enum
+{
+    protected readonly IServiceProvider? ServiceProvider;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="KernelProcessEventsSubscriber{TEvents}"/> class.
+    /// </summary>
+    /// <param name="serviceProvider">Optional service provider for resolving dependencies</param>
+    public KernelProcessEventsSubscriber(IServiceProvider? serviceProvider = null)
+    {
+        this.ServiceProvider = serviceProvider;
+    }
+
+    /// <summary>
+    /// Attribute to set Process related steps to link Process Events to specific functions to execute when the event is emitted outside the Process
+    /// </summary>
+    [AttributeUsage(AttributeTargets.Method, AllowMultiple = false)]
+    public class ProcessEventSubscriberAttribute : Attribute
+    {
+        /// <summary>
+        /// Gets the enum of the event that the function is linked to
+        /// </summary>
+        public TEvents EventEnum { get; }
+
+        /// <summary>
+        /// Gets the string of the event name that the function is linked to
+        /// </summary>
+        public string EventName { get; }
+
+        /// <summary>
+        /// Initializes the attribute.
+        /// </summary>
+        /// <param name="eventEnum">Specific Process Event enum</param>
+        public ProcessEventSubscriberAttribute(TEvents eventEnum)
+        {
+            this.EventEnum = eventEnum;
+            this.EventName = Enum.GetName(typeof(TEvents), eventEnum) ?? "";
+        }
+    }
+}

dotnet/src/Experimental/Process.Abstractions/KernelProcessEventsSubscriberInfo.cs

@@ -0,0 +1,92 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Collections.Generic;
+using System.Linq;
+using System.Reflection;
+using System;
+using Microsoft.SemanticKernel.Process;
+
+namespace Microsoft.SemanticKernel;
+
+public class KernelProcessEventsSubscriberInfo
+{
+    private readonly Dictionary<string, List<MethodInfo>> _eventHandlers = [];
+    private readonly Dictionary<string, string> _stepEventProcessEventMap = [];
+    private Type? _processEventSubscriberType = null;
+
+    private IServiceProvider? _subscriberServiceProvider = null;
+
+    protected void Subscribe(string eventName, MethodInfo method)
+    {
+        if (this._eventHandlers.TryGetValue(eventName, out List<MethodInfo>? eventHandlers) && eventHandlers != null)
+        {
+            eventHandlers.Add(method);
+        }
+    }
+
+    public void LinkStepEventToProcessEvent(string stepEventId, string processEventId)
+    {
+        this._stepEventProcessEventMap.Add(stepEventId, processEventId);
+        if (!this._eventHandlers.ContainsKey(processEventId))
+        {
+            this._eventHandlers.Add(processEventId, []);
+        }
+    }
+
+    public void TryInvokeProcessEventFromStepMessage(string stepEventId, object? data)
+    {
+        if (this._stepEventProcessEventMap.TryGetValue(stepEventId, out var processEvent) && processEvent != null)
+        {
+            this.InvokeProcessEvent(processEvent, data);
+        }
+    }
+
+    public void InvokeProcessEvent(string eventName, object? data)
+    {
+        if (this._processEventSubscriberType != null && this._eventHandlers.TryGetValue(eventName, out List<MethodInfo>? linkedMethods) && linkedMethods != null)
+        {
+            foreach (var method in linkedMethods)
+            {
+                // TODO-estenori: Avoid creating a new instance every time a function is invoked - create instance once only?
+                var instance = Activator.CreateInstance(this._processEventSubscriberType, [this._subscriberServiceProvider]);
+                method.Invoke(instance, [data]);
+            }
+        }
+    }
+
+    /// <summary>
+    /// Extracts the event properties and function details of the functions with the annotator
+    /// <see cref="KernelProcessEventsSubscriber{TEvents}.ProcessEventSubscriberAttribute"/>
+    /// </summary>
+    /// <typeparam name="TEventListeners">Type of the class that make uses of the annotators and contains the functionality to be executed</typeparam>
+    /// <typeparam name="TEvents">Enum that contains the process subscribable events</typeparam>
+    /// <exception cref="InvalidOperationException"></exception>
+    public void SubscribeToEventsFromClass<TEventListeners, TEvents>(IServiceProvider? serviceProvider = null) where TEventListeners : KernelProcessEventsSubscriber<TEvents> where TEvents : Enum
+    {
+        if (this._subscriberServiceProvider != null)
+        {
+            throw new KernelException("Already linked process to a specific service provider class");
+        }
+
+        var methods = typeof(TEventListeners).GetMethods(BindingFlags.Instance | BindingFlags.NonPublic | BindingFlags.Public | BindingFlags.DeclaredOnly);
+        foreach (var method in methods)
+        {
+            if (method.GetCustomAttributes(typeof(KernelProcessEventsSubscriber<>.ProcessEventSubscriberAttribute), false).FirstOrDefault() is KernelProcessEventsSubscriber<TEvents>.ProcessEventSubscriberAttribute attribute)
+            {
+                if (attribute.EventEnum.GetType() != typeof(TEvents))
+                {
+                    throw new InvalidOperationException($"The event type {attribute.EventEnum.GetType().Name} does not match the expected type {typeof(TEvents).Name}");
+                }
+
+                this.Subscribe(attribute.EventName, method);
+            }
+        }
+
+        this._subscriberServiceProvider = serviceProvider;
+        this._processEventSubscriberType = typeof(TEventListeners);
+    }
+
+    public KernelProcessEventsSubscriberInfo()
+    {
+    }
+}