feat: Implement BRS-026 walking skeleton

docs/ProcessManager.Client/ReleaseNotes/ReleaseNotes.md

@@ -1,16 +1,20 @@
 # ProcessManager.Client Release Notes
 
+## Version 0.14.4
+
+- Add actor id to start orchestration messages.
+
 ## Version 0.14.3
 
-- No functional changes
+- No functional changes.
 
 ## Version 0.14.2
 
-- No functional changes
+- No functional changes.
 
 ## Version 0.14.1
 
-- Rename service bus client options
+- Rename service bus client options.
 
 ## Version 0.14.0
 

docs/ProcessManager.Orchestrations.Abstractions/ReleaseNotes/ReleaseNotes.md

@@ -1,5 +1,9 @@
 # ProcessManager.Orchestrations.Abstractions Release Notes
 
+## Version 0.2.5
+
+- Add actor id to start orchestration messages.
+
 ## Version 0.2.4
 
 - Update (implement) correct input for starting BRS-026 and BRS-028 orchestrations.

source/ProcessManager.Abstractions/Contracts/StartOrchestrationDto.proto

@@ -20,5 +20,6 @@ option csharp_namespace = "Energinet.DataHub.ProcessManager.Abstractions.Contrac
 message StartOrchestrationDto {
   string orchestration_name = 1; // The name of the orchestration to start.
   int32 orchestration_version = 2; // The version of the orchestration to start.
+  string started_by_actor_id = 4; // The actor id who wants to start the orchestration
   string json_input = 3; // The input to the orchestration serialized as a JSON
 }

source/ProcessManager.Abstractions/ProcessManager.Abstractions.csproj

@@ -7,7 +7,7 @@
 
   <PropertyGroup>
     <PackageId>Energinet.DataHub.ProcessManager.Abstractions</PackageId>
-    <PackageVersion>0.14.3$(VersionSuffix)</PackageVersion>
+    <PackageVersion>0.14.4$(VersionSuffix)</PackageVersion>
     <Title>DH3 Process Manager Abstractions library</Title>
     <Company>Energinet-DataHub</Company>
     <Authors>Energinet-DataHub</Authors>

source/ProcessManager.Client.Tests/Fixtures/ProcessManagerClientFixture.cs

@@ -97,7 +97,9 @@ public async Task InitializeAsync()
 
         ProcessManagerTopic = await ServiceBusResourceProvider.BuildTopic("pm-topic")
             .AddSubscription(brs026SubscriptionName)
+                .AddSubjectFilter("Brs_026")
             .AddSubscription(brs021ForwardMeteredDataSubscriptionName)
+                .AddSubjectFilter("Brs_021_ForwardMeteredData")
             .CreateAsync();
         var brs026Subscription = ProcessManagerTopic.Subscriptions.Single(x => x.SubscriptionName.Equals(brs026SubscriptionName));
         var brs021ForwardMeteredDataSubscription = ProcessManagerTopic.Subscriptions.Single(x => x.SubscriptionName.Equals(brs021ForwardMeteredDataSubscriptionName));

source/ProcessManager.Client.Tests/Integration/BRS_026_028/V1/RequestCalculatedEnergyTimeSeriesTests.cs

@@ -106,7 +106,7 @@ public async Task RequestCalculatedEnergyTimeSeries_WhenStarted_OrchestrationCom
         // Assert
         var orchestration = await _fixture.DurableClient.WaitForOrchestationStartedAsync(
             createdTimeFrom: orchestrationCreatedAfter,
-            name: "RequestCalculatedEnergyTimeSeriesOrchestrationV1");
+            name: "Orchestration_RequestCalculatedEnergyTimeSeries_V1");
         orchestration.Input.ToString().Should().Contain(businessReason);
 
         var completedOrchestration = await _fixture.DurableClient.WaitForOrchestrationCompletedAsync(

source/ProcessManager.Client/ProcessManager.Client.csproj

@@ -7,7 +7,7 @@
 
   <PropertyGroup>
     <PackageId>Energinet.DataHub.ProcessManager.Client</PackageId>
-    <PackageVersion>0.14.3$(VersionSuffix)</PackageVersion>
+    <PackageVersion>0.14.4$(VersionSuffix)</PackageVersion>
     <Title>DH3 Process Manager Client library</Title>
     <Company>Energinet-DataHub</Company>
     <Authors>Energinet-DataHub</Authors>

source/ProcessManager.Client/ProcessManagerMessageClient.cs

@@ -34,32 +34,28 @@ public Task StartNewOrchestrationInstanceAsync<TInputParameterDto>(
         CancellationToken cancellationToken)
             where TInputParameterDto : IInputParameterDto
     {
-        var serviceBusMessage = CreateServiceBusMessage(
-            command.OrchestrationDescriptionUniqueName.Name,
-            command.OrchestrationDescriptionUniqueName.Version,
-            command);
+        var serviceBusMessage = CreateServiceBusMessage(command);
 
         return SendServiceBusMessage(
                 serviceBusMessage,
                 cancellationToken);
     }
 
     private ServiceBusMessage CreateServiceBusMessage<TInputParameterDto>(
-        string orchestrationName,
-        int orchestrationVersion,
         MessageCommand<TInputParameterDto> command)
     where TInputParameterDto : IInputParameterDto
     {
         var message = new StartOrchestrationDto
         {
-            OrchestrationName = orchestrationName,
-            OrchestrationVersion = orchestrationVersion,
+            OrchestrationName = command.OrchestrationDescriptionUniqueName.Name,
+            OrchestrationVersion = command.OrchestrationDescriptionUniqueName.Version,
+            StartedByActorId = command.OperatingIdentity.ActorId.ToString(),
             JsonInput = JsonSerializer.Serialize(command.InputParameter),
         };
 
         ServiceBusMessage serviceBusMessage = new(JsonFormatter.Default.Format(message))
         {
-            Subject = orchestrationName,
+            Subject = command.OrchestrationDescriptionUniqueName.Name,
             MessageId = command.MessageId,
             ContentType = "application/json",
         };

source/ProcessManager.Core/Domain/OrchestrationInstance/OrchestrationInstance.cs

@@ -82,6 +82,39 @@ private OrchestrationInstance()
     /// </summary>
     internal OrchestrationDescriptionId OrchestrationDescriptionId { get; }
 
+    /// <summary>
+    /// Transition a step's lifecycle to running
+    /// </summary>
+    /// <param name="sequence">The sequence number of the step to transition</param>
+    /// <param name="clock"></param>
+    /// <exception cref="ArgumentOutOfRangeException">Thrown if a step with the given <paramref name="sequence"/> isn't found.</exception>
+    public void TransitionStepToRunning(int sequence, IClock clock)
+    {
+        var step = Steps.SingleOrDefault(s => s.Sequence == sequence);
+
+        if (step == null)
+            throw new ArgumentOutOfRangeException(nameof(sequence), sequence, "A step with the given sequence does not exist");
+
+        step.Lifecycle.TransitionToRunning(clock);
+    }
+
+    /// <summary>
+    /// Transition a step's lifecycle to terminated, with the given <paramref name="terminationState"/>
+    /// </summary>
+    /// <param name="sequence">The sequence number of the step to transition</param>
+    /// <param name="terminationState">The state of the termination step (Succeeded, failed etc.)</param>
+    /// <param name="clock"></param>
+    /// <exception cref="ArgumentOutOfRangeException">Thrown if a step with the given <paramref name="sequence"/> isn't found.</exception>
+    public void TransitionStepToTerminated(int sequence, OrchestrationStepTerminationStates terminationState, IClock clock)
+    {
+        var step = Steps.SingleOrDefault(s => s.Sequence == sequence);
+
+        if (step == null)
+            throw new ArgumentOutOfRangeException(nameof(sequence), sequence, "A step with the given sequence does not exist");
+
+        step.Lifecycle.TransitionToTerminated(clock, terminationState);
+    }
+
     /// <summary>
     /// Factory method that ensures domain rules are obeyed when creating a new
     /// orchestration instance.

source/ProcessManager.Orchestrations.Abstractions/ProcessManager.Orchestrations.Abstractions.csproj

@@ -7,7 +7,7 @@
 
   <PropertyGroup>
     <PackageId>Energinet.DataHub.ProcessManager.Orchestrations.Abstractions</PackageId>
-    <PackageVersion>0.2.4$(VersionSuffix)</PackageVersion>
+    <PackageVersion>0.2.5$(VersionSuffix)</PackageVersion>
     <Title>DH3 Process Manager Orchestrations Abstractions library</Title>
     <Company>Energinet-DataHub</Company>
     <Authors>Energinet-DataHub</Authors>

source/ProcessManager.Orchestrations.Tests/Fixtures/OrchestrationsAppManager.cs

@@ -126,12 +126,14 @@ public async Task StartAsync(
 
             if (brs026Subscription is null)
             {
-               topicResourceBuilder.AddSubscription(brs026SubscriptionName);
+               topicResourceBuilder.AddSubscription(brs026SubscriptionName)
+                   .AddSubjectFilter("Brs_026");
             }
 
             if (brs021ForwardMeteredDataSubscription is null)
             {
-                topicResourceBuilder.AddSubscription(brs021ForwardMeteredDataSubscriptionName);
+                topicResourceBuilder.AddSubscription(brs021ForwardMeteredDataSubscriptionName)
+                    .AddSubjectFilter("Brs_021_ForwardMeteredData");
             }
 
             var topicResource = await topicResourceBuilder.CreateAsync();

source/ProcessManager.Orchestrations/ProcessManager.Orchestrations.csproj

@@ -32,7 +32,6 @@
   <ItemGroup>
     <Folder Include="Api\Mappers\" />
     <Folder Include="Processes\BRS_021\ForwardMeteredData\" />
-    <Folder Include="Processes\BRS_026\V1\Activities\" />
   </ItemGroup>
   <ItemGroup>
     <Compile Include="..\Shared\ProcessManager\Api\Mappers\OrchestrationInstanceMapperExtensions.cs" Link="Api\Mappers\OrchestrationInstanceMapperExtensions.cs" />

source/ProcessManager.Orchestrations/Processes/BRS_023_027/V1/Activities/CalculationStepStartActivity_Brs_023_027_V1.cs

@@ -28,10 +28,10 @@ internal class CalculationStepStartActivity_Brs_023_027_V1(
 {
     [Function(nameof(CalculationStepStartActivity_Brs_023_027_V1))]
     public async Task Run(
-        [ActivityTrigger] Guid orchestrationInstanceId)
+        [ActivityTrigger] ActivityInput input)
     {
         var orchestrationInstance = await ProgressRepository
-            .GetAsync(new OrchestrationInstanceId(orchestrationInstanceId))
+            .GetAsync(input.InstanceId)
             .ConfigureAwait(false);
 
         var step = orchestrationInstance.Steps.Single(x => x.Sequence == Orchestration_Brs_023_027_V1.CalculationStepSequence);
@@ -41,4 +41,7 @@ public async Task Run(
         // TODO: For demo purposes; remove when done
         await Task.Delay(TimeSpan.FromSeconds(3)).ConfigureAwait(false);
     }
+
+    public record ActivityInput(
+        OrchestrationInstanceId InstanceId);
 }

source/ProcessManager.Orchestrations/Processes/BRS_023_027/V1/Activities/CalculationStepTerminateActivity_Brs_023_027_V1.cs

@@ -28,10 +28,10 @@ internal class CalculationStepTerminateActivity_Brs_023_027_V1(
 {
     [Function(nameof(CalculationStepTerminateActivity_Brs_023_027_V1))]
     public async Task Run(
-        [ActivityTrigger] Guid orchestrationInstanceId)
+        [ActivityTrigger] ActivityInput input)
     {
         var orchestrationInstance = await ProgressRepository
-            .GetAsync(new OrchestrationInstanceId(orchestrationInstanceId))
+            .GetAsync(input.InstanceId)
             .ConfigureAwait(false);
 
         var step = orchestrationInstance.Steps.Single(x => x.Sequence == Orchestration_Brs_023_027_V1.CalculationStepSequence);
@@ -41,4 +41,7 @@ public async Task Run(
         // TODO: For demo purposes; remove when done
         await Task.Delay(TimeSpan.FromSeconds(1)).ConfigureAwait(false);
     }
+
+    public record ActivityInput(
+        OrchestrationInstanceId InstanceId);
 }

source/ProcessManager.Orchestrations/Processes/BRS_023_027/V1/Activities/EnqueueMessagesStepStartActivity_Brs_023_027_V1.cs

@@ -28,10 +28,10 @@ internal class EnqueueMessagesStepStartActivity_Brs_023_027_V1(
 {
     [Function(nameof(EnqueueMessagesStepStartActivity_Brs_023_027_V1))]
     public async Task Run(
-        [ActivityTrigger] Guid orchestrationInstanceId)
+        [ActivityTrigger] ActivityInput input)
     {
         var orchestrationInstance = await ProgressRepository
-            .GetAsync(new OrchestrationInstanceId(orchestrationInstanceId))
+            .GetAsync(input.InstanceId)
             .ConfigureAwait(false);
 
         var step = orchestrationInstance.Steps.Single(x => x.Sequence == Orchestration_Brs_023_027_V1.EnqueueMessagesStepSequence);
@@ -44,4 +44,7 @@ public async Task Run(
             await Task.Delay(TimeSpan.FromSeconds(3)).ConfigureAwait(false);
         }
     }
+
+    public record ActivityInput(
+        OrchestrationInstanceId InstanceId);
 }

source/ProcessManager.Orchestrations/Processes/BRS_023_027/V1/Activities/EnqueueMessagesStepTerminateActivity_Brs_023_027_V1.cs

@@ -28,10 +28,10 @@ internal class EnqueueMessagesStepTerminateActivity_Brs_023_027_V1(
 {
     [Function(nameof(EnqueueMessagesStepTerminateActivity_Brs_023_027_V1))]
     public async Task Run(
-        [ActivityTrigger] Guid orchestrationInstanceId)
+        [ActivityTrigger] ActivityInput input)
     {
         var orchestrationInstance = await ProgressRepository
-            .GetAsync(new OrchestrationInstanceId(orchestrationInstanceId))
+            .GetAsync(input.InstanceId)
             .ConfigureAwait(false);
 
         var step = orchestrationInstance.Steps.Single(x => x.Sequence == Orchestration_Brs_023_027_V1.EnqueueMessagesStepSequence);
@@ -44,4 +44,7 @@ public async Task Run(
             await Task.Delay(TimeSpan.FromSeconds(1)).ConfigureAwait(false);
         }
     }
+
+    public record ActivityInput(
+        OrchestrationInstanceId InstanceId);
 }

source/ProcessManager.Orchestrations/Processes/BRS_023_027/V1/Activities/OrchestrationInitializeActivity_Brs_023_027_V1.cs

@@ -32,10 +32,10 @@ internal class OrchestrationInitializeActivity_Brs_023_027_V1(
 {
     [Function(nameof(OrchestrationInitializeActivity_Brs_023_027_V1))]
     public async Task Run(
-        [ActivityTrigger] Guid orchestrationInstanceId)
+        [ActivityTrigger] ActivityInput input)
     {
         var orchestrationInstance = await ProgressRepository
-            .GetAsync(new OrchestrationInstanceId(orchestrationInstanceId))
+            .GetAsync(input.InstanceId)
             .ConfigureAwait(false);
 
         orchestrationInstance.Lifecycle.TransitionToRunning(Clock);
@@ -44,4 +44,7 @@ public async Task Run(
         // TODO: For demo purposes; remove when done
         await Task.Delay(TimeSpan.FromSeconds(1)).ConfigureAwait(false);
     }
+
+    public record ActivityInput(
+        OrchestrationInstanceId InstanceId);
 }

source/ProcessManager.Orchestrations/Processes/BRS_023_027/V1/Activities/OrchestrationTerminateActivity_Brs_023_027_V1.cs

@@ -32,10 +32,10 @@ internal class OrchestrationTerminateActivity_Brs_023_027_V1(
 {
     [Function(nameof(OrchestrationTerminateActivity_Brs_023_027_V1))]
     public async Task Run(
-        [ActivityTrigger] Guid orchestrationInstanceId)
+        [ActivityTrigger] ActivityInput input)
     {
         var orchestrationInstance = await ProgressRepository
-            .GetAsync(new OrchestrationInstanceId(orchestrationInstanceId))
+            .GetAsync(input.InstanceId)
             .ConfigureAwait(false);
 
         orchestrationInstance.Lifecycle.TransitionToSucceeded(Clock);
@@ -44,4 +44,7 @@ public async Task Run(
         // TODO: For demo purposes; remove when done
         await Task.Delay(TimeSpan.FromSeconds(1)).ConfigureAwait(false);
     }
+
+    public record ActivityInput(
+        OrchestrationInstanceId InstanceId);
 }