Skip to content

Commit 06ce593

Browse files
bindsibindsi
committed
finalized workflow-module
1 parent 0470696 commit 06ce593

File tree

9 files changed

+82
-67
lines changed

9 files changed

+82
-67
lines changed

‎iotedge/Distributed.IoT.Edge/Distributed.IoT.Edge.DataGatewayModule/Program.cs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,8 +33,8 @@
3333

3434
// Additional configuration is required to successfully run gRPC on macOS.
3535
// For instructions on how to configure Kestrel and gRPC clients on macOS, visit https://go.microsoft.com/fwlink/?linkid=2099682
36-
// builder.WebHost.ConfigureKestrel(k => k.ListenLocalhost(5001, op => op.Protocols =
37-
// HttpProtocols.Http2));
36+
builder.WebHost.ConfigureKestrel(k => k.ListenLocalhost(5001, op => op.Protocols =
37+
HttpProtocols.Http2));
3838

3939
// Add services to the container.
4040
builder.Services.AddGrpc();

‎iotedge/Distributed.IoT.Edge/Distributed.IoT.Edge.WorkflowModule/Activities/EnrichmentActivity.cs

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,9 +2,7 @@
22
{
33
using System;
44
using System.Threading.Tasks;
5-
using Dapr.Client;
65
using Dapr.Workflow;
7-
using Distributed.IoT.Edge.WorkflowModule;
86
using Microsoft.Extensions.Logging;
97
using Newtonsoft.Json;
108

‎iotedge/Distributed.IoT.Edge/Distributed.IoT.Edge.WorkflowModule/Activities/PublishActivity.cs

Lines changed: 4 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -6,14 +6,13 @@
66
using Dapr.Workflow;
77
using Distributed.IoT.Edge.WorkflowModule;
88
using Microsoft.Extensions.Logging;
9-
using Newtonsoft.Json;
109

1110
public class PublishActivity : WorkflowActivity<string, bool>
1211
{
1312
private readonly ILogger<PublishActivity> _logger;
1413
private readonly DaprClient _daprClient;
15-
private readonly string _senderPubsubName;
16-
private readonly string _senderPubsubTopicName;
14+
private readonly string? _senderPubsubName;
15+
private readonly string? _senderPubsubTopicName;
1716

1817
public PublishActivity(
1918
ILogger<PublishActivity> logger,
@@ -28,14 +27,8 @@ public PublishActivity(
2827
throw new ArgumentNullException(nameof(parameters));
2928
}
3029

31-
_senderPubsubName = parameters.SenderPubSubName ??
32-
throw new ArgumentNullException(
33-
nameof(parameters.SenderPubSubName),
34-
"Parameter cannot be null.");
35-
_senderPubsubTopicName = parameters.SenderPubSubTopicName ??
36-
throw new ArgumentNullException(
37-
nameof(parameters.SenderPubSubTopicName),
38-
"Parameter cannot be null.");
30+
_senderPubsubName = parameters.ReceiverPubSubName;
31+
_senderPubsubTopicName = parameters.SenderPubSubTopicName;
3932

4033
_logger.LogTrace($"Entering: {nameof(PublishActivity)}");
4134
}

‎iotedge/Distributed.IoT.Edge/Distributed.IoT.Edge.WorkflowModule/Components/mqtt3-pub-sub.yaml

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
apiVersion: dapr.io/v1alpha1
2+
kind: Component
3+
metadata:
4+
name: local-pub-sub
5+
spec:
6+
type: pubsub.mqtt3
7+
version: v1
8+
metadata:
9+
- name: url
10+
value: mqtt://localhost:1883
11+
- name: retain
12+
value: "false"
13+
- name: cleanSession
14+
value: "false"
15+
- name: qos
16+
value: "1"
17+
- name: consumerID
18+
value: "workflow-module"

‎iotedge/Distributed.IoT.Edge/Distributed.IoT.Edge.WorkflowModule/Components/redis-statestore.yaml

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
apiVersion: dapr.io/v1alpha1
2+
kind: Component
3+
metadata:
4+
name: statestore
5+
spec:
6+
type: state.redis
7+
version: v1
8+
initTimeout: 1m
9+
metadata:
10+
- name: redisHost
11+
value: localhost:6379
12+
- name: redisPassword
13+
value: ""
14+
- name: actorStateStore
15+
value: "true"

‎iotedge/Distributed.IoT.Edge/Distributed.IoT.Edge.WorkflowModule/Distributed.IoT.Edge.WorkflowModule.csproj

Lines changed: 13 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
<Project Sdk="Microsoft.NET.Sdk.Web">
22
<PropertyGroup>
33
<OutputType>Exe</OutputType>
4-
<TargetFramework>net6.0</TargetFramework>
4+
<TargetFramework>net8.0</TargetFramework>
55
<ImplicitUsings>enable</ImplicitUsings>
66
<Nullable>enable</Nullable>
77
<DockerDefaultTargetOS>Linux</DockerDefaultTargetOS>
@@ -14,21 +14,23 @@
1414
<ItemGroup>
1515
<PackageReference Include="Azure.Messaging.EventHubs" Version="5.7.5" />
1616
<PackageReference Include="CommandLineParser" Version="2.9.1" />
17-
<PackageReference Include="Dapr.AspNetCore" Version="1.10.0" />
18-
<PackageReference Include="Dapr.Workflow" Version="1.10.0" />
19-
<PackageReference Include="Google.Api.CommonProtos" Version="2.8.0" />
20-
<PackageReference Include="Google.Protobuf" Version="3.22.0" />
21-
<PackageReference Include="Grpc.AspNetCore" Version="2.51.0" />
22-
<PackageReference Include="Grpc.Net.Client" Version="2.51.0" />
23-
<PackageReference Include="Microsoft.VisualStudio.Azure.Containers.Tools.Targets" Version="1.17.0" />
17+
<PackageReference Include="Dapr.AspNetCore" Version="1.13.*-*" />
18+
<PackageReference Include="Dapr.Workflow" Version="1.13.*-*" />
19+
<PackageReference Include="Google.Api.CommonProtos" Version="2.15.0" />
20+
<PackageReference Include="Google.Protobuf" Version="3.27.1" />
21+
<PackageReference Include="Grpc.AspNetCore" Version="2.63.0" />
22+
<PackageReference Include="Grpc.Net.Client" Version="2.63.0" />
23+
<PackageReference Include="Microsoft.VisualStudio.Azure.Containers.Tools.Targets"
24+
Version="1.20.1" />
2425
</ItemGroup>
2526

2627
<ItemGroup>
27-
<ProjectReference Include="..\Distributed.IoT.Edge.Common\Distributed.IoT.Edge.Common.csproj" />
28+
<ProjectReference
29+
Include="..\Distributed.IoT.Edge.Common\Distributed.IoT.Edge.Common.csproj" />
2830
</ItemGroup>
2931

3032
<ItemGroup>
31-
<PackageReference Update="Microsoft.CodeAnalysis.NetAnalyzers" Version="7.0.1 ">
33+
<PackageReference Update="Microsoft.CodeAnalysis.NetAnalyzers" Version="8.0.0">
3234
<PrivateAssets>all</PrivateAssets>
3335
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
3436
</PackageReference>
@@ -39,4 +41,4 @@
3941
</Content>
4042
</ItemGroup>
4143

42-
</Project>
44+
</Project>

‎iotedge/Distributed.IoT.Edge/Distributed.IoT.Edge.WorkflowModule/Program.cs

Lines changed: 9 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,15 @@
11
// Local run cmd line.
2-
// dapr run --app-id workflow-module --app-protocol grpc --app-port 5000 --resources-path=../../../deployment/helm/iot-edge-accelerator/templates/dapr -- dotnet run -- --receiverPubSubName "local-pub-sub" --receiverPubSubTopicName "telemetry" --senderPubSubName "local-pub-sub" --senderPubSubTopicName "enriched-telemetry"
2+
// dapr run --app-id workflow-module --app-protocol grpc --app-port 5002 --resources-path=./Components --dapr-grpc-port 5005 -- dotnet run -- --receiverPubSubName "local-pub-sub-2" --receiverPubSubTopicName "telemetry" --senderPubSubName "local-pub-sub" --senderPubSubTopicName "enriched-telemetry"
3+
// mosquitto_pub -h localhost -p 1883 -t telemetry -m '{"ambient":{"temperature":10}}'
4+
// mosquitto_sub -h localhost -p 1883 -t enriched-telemetry
5+
// dapr publish --publish-app-id workflow-module --pubsub local-pub-sub --topic telemetry --data '{"ambient":{"temperature":10}}' --metadata '{"rawPayload":"true"}'
36

4-
using System.Collections.Immutable;
57
using CommandLine;
6-
using Dapr.Client;
78
using Dapr.Workflow;
89
using Distributed.IoT.Edge.WorkflowModule;
910
using Distributed.IoT.Edge.WorkflowModule.Services;
1011
using Distributed.IoT.Edge.WorkflowModule.Workflows;
1112
using Microsoft.AspNetCore.Server.Kestrel.Core;
12-
using Microsoft.Extensions.DependencyInjection.Extensions;
1313
using WorkflowConsoleApp.Activities;
1414

1515
// Environment.SetEnvironmentVariable("DAPR_GRPC_PORT", "50001");
@@ -31,28 +31,27 @@
3131
builder.Services.AddSingleton<WorkflowParameters>(sp => parameters);
3232

3333
// Already registered by AddDaprWorkflow extension
34-
builder.Services.AddSingleton<DaprClient>(new DaprClientBuilder().Build());
34+
//builder.Services.AddSingleton<DaprClient>(new DaprClientBuilder().Build());
3535
builder.Services.AddTransient<SubscriptionService>(
3636
sp => new SubscriptionService(
3737
sp.GetRequiredService<ILogger<SubscriptionService>>(),
38-
sp.GetRequiredService<DaprClient>(),
39-
sp.GetRequiredService<WorkflowEngineClient>(),
38+
sp.GetRequiredService<DaprWorkflowClient>(),
4039
parameters));
4140
})
4241
.WithNotParsed(errors =>
4342
{
4443
Environment.Exit(1);
4544
});
4645

47-
// builder.WebHost.ConfigureKestrel(k => k.ListenLocalhost(5001, op => op.Protocols =
48-
// HttpProtocols.Http2));
46+
builder.WebHost.ConfigureKestrel(k => k.ListenLocalhost(5002, op => op.Protocols =
47+
HttpProtocols.Http2));
4948

5049
// Additional configuration is required to successfully run gRPC on macOS.
5150
// For instructions on how to configure Kestrel and gRPC clients on macOS, visit https://go.microsoft.com/fwlink/?linkid=2099682
5251
builder.Services.AddGrpc();
5352

5453
var app = builder.Build();
55-
54+
app.UseRouting();
5655
// Configure the HTTP request pipeline.
5756
app.MapGrpcService<SubscriptionService>();
5857

‎iotedge/Distributed.IoT.Edge/Distributed.IoT.Edge.WorkflowModule/Services/SubscriptionService.cs

Lines changed: 18 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -1,43 +1,33 @@
1-
using Dapr.AppCallback.Autogen.Grpc.v1;
2-
using Dapr.Client;
3-
using Dapr.Client.Autogen.Grpc.v1;
4-
using Dapr.Workflow;
5-
using Distributed.IoT.Edge.WorkflowModule.Workflows;
6-
using Google.Protobuf.WellKnownTypes;
7-
using Grpc.Core;
8-
91
namespace Distributed.IoT.Edge.WorkflowModule.Services
102
{
3+
using Dapr.AppCallback.Autogen.Grpc.v1;
4+
using Dapr.Workflow;
5+
using Distributed.IoT.Edge.WorkflowModule.Workflows;
6+
using Google.Protobuf.WellKnownTypes;
7+
using Grpc.Core;
8+
119
public class SubscriptionService : AppCallback.AppCallbackBase
1210
{
1311
private readonly ILogger<SubscriptionService> _logger;
14-
private readonly DaprClient _daprClient;
15-
private readonly WorkflowEngineClient _workflowClient;
12+
private readonly DaprWorkflowClient _workflowClient;
1613
private readonly string _receiverPubsubName;
1714
private readonly string _receiverPubsubTopicName;
1815

1916
public SubscriptionService(
2017
ILogger<SubscriptionService> logger,
21-
DaprClient daprClient,
22-
WorkflowEngineClient workflowClient,
18+
DaprWorkflowClient workflowClient,
2319
WorkflowParameters parameters)
2420
{
2521
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
26-
_daprClient = daprClient ?? throw new ArgumentNullException(nameof(daprClient));
27-
_workflowClient = workflowClient ?? throw new ArgumentNullException(nameof(workflowClient));
22+
_workflowClient = workflowClient ?? throw new ArgumentNullException(nameof(workflowClient));
2823
if (parameters == null)
2924
{
3025
throw new ArgumentNullException(nameof(parameters));
3126
}
3227

33-
_receiverPubsubName = parameters.ReceiverPubSubName ??
34-
throw new ArgumentNullException(
35-
nameof(parameters.ReceiverPubSubName),
36-
"Parameter cannot be null.");
37-
_receiverPubsubTopicName = parameters.ReceiverPubSubTopicName ??
38-
throw new ArgumentNullException(
39-
nameof(parameters.ReceiverPubSubTopicName),
40-
"Parameter cannot be null.");
28+
_receiverPubsubName = parameters.ReceiverPubSubName;
29+
_receiverPubsubTopicName = parameters.ReceiverPubSubTopicName;
30+
4131
}
4232

4333
public override Task<ListTopicSubscriptionsResponse> ListTopicSubscriptions(
@@ -68,30 +58,30 @@ public override async Task<TopicEventResponse> OnTopicEvent(
6858
throw new ArgumentNullException(nameof(request));
6959
}
7060

71-
var topicString = request.Data.ToStringUtf8();
72-
_logger.LogTrace($"requestPath: {request.Path}");
61+
var topicString = request.Extensions.ToString();
7362
_logger.LogTrace($"Sending event to workflow, object json: {topicString}");
7463

64+
var instanceId = Guid.NewGuid().ToString();
7565
// starting workflow to enrich and transform the data
7666
await _workflowClient.ScheduleNewWorkflowAsync(
7767
name: nameof(EnrichTelemetryWorkflow),
78-
instanceId: request.Id,
68+
instanceId: instanceId,
7969
input: topicString);
8070

8171
// Wait a second to allow workflow to start
8272
await Task.Delay(TimeSpan.FromSeconds(1));
8373
WorkflowState state = await _workflowClient.GetWorkflowStateAsync(
84-
instanceId: request.Id,
74+
instanceId: instanceId,
8575
getInputsAndOutputs: true);
8676

8777
_logger.LogTrace($"Your workflow {request.Id} has started. Here is the status of the workflow: {state.RuntimeStatus}");
8878
while (!state.IsWorkflowCompleted)
8979
{
9080
await Task.Delay(TimeSpan.FromSeconds(5));
9181
state = await _workflowClient.GetWorkflowStateAsync(
92-
instanceId: request.Id,
82+
instanceId: instanceId,
9383
getInputsAndOutputs: true);
94-
_logger.LogTrace($"State of workflow {request.Id} is: {state.RuntimeStatus}");
84+
_logger.LogTrace($"State of workflow {instanceId} is: {state.RuntimeStatus}");
9585
}
9686

9787
// Depending on the status return dapr side will either retry or drop the message from underlying pubsub.

‎iotedge/Distributed.IoT.Edge/Distributed.IoT.Edge.WorkflowModule/WorkflowParameters.cs

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
1-
using CommandLine;
2-
3-
namespace Distributed.IoT.Edge.WorkflowModule
1+
namespace Distributed.IoT.Edge.WorkflowModule
42
{
3+
using CommandLine;
4+
55
public class WorkflowParameters
66
{
77
[Option(

0 commit comments

Comments
 (0)