mirror of https://github.com/dapr/dotnet-sdk.git
Add .NET client for pub/sub support - streaming subscriptions (#1381)
* Building out Dapr.Messaging and test project for streaming pubsub subscriptions Signed-off-by: Whit Waldo <whit.waldo@innovian.net> * Added copyright notices Signed-off-by: Whit Waldo <whit.waldo@innovian.net> * Minor stylistic updates Signed-off-by: Whit Waldo <whit.waldo@innovian.net> * Added generic client builder to support publish/subscribe client builder Signed-off-by: Whit Waldo <whit.waldo@innovian.net> * Tweaked XML comment Signed-off-by: Whit Waldo <whit.waldo@innovian.net> * Added several unit tests for the generic client builder Signed-off-by: Whit Waldo <whit.waldo@innovian.net> * Updated to include latest review changes: - Added lock so that while we guarantee the method is called only once, it should be thread-safe now - Marked PublishSubscribeReceiver as internal so its members aren't part of the public API - Updated TopicMessage to use IReadOnlyDictionary Signed-off-by: Whit Waldo <whit.waldo@innovian.net> * Switched to interlock exchange instead of lock to slightly simplify code Signed-off-by: Whit Waldo <whit.waldo@innovian.net> * Added sample project Signed-off-by: Whit Waldo <whit.waldo@innovian.net> * Minor changes to unit test Signed-off-by: Whit Waldo <whit.waldo@innovian.net> * Deleted protos folder Signed-off-by: Whit Waldo <whit.waldo@innovian.net> * Using lowercase protos dir name Signed-off-by: Whit Waldo <whit.waldo@innovian.net> * Added registration extension methods Signed-off-by: Whit Waldo <whit.waldo@innovian.net> * Updated example to use DI registration Signed-off-by: Whit Waldo <whit.waldo@innovian.net> * Added default cancellation token Signed-off-by: Whit Waldo <whit.waldo@innovian.net> * Passing stream into method instead of creating it twice Signed-off-by: Whit Waldo <whit.waldo@innovian.net> --------- Signed-off-by: Whit Waldo <whit.waldo@innovian.net>
This commit is contained in:
Whit Waldo
GitHub
parent
682df6fec9
commit
91ee78aff4
|
|
@ -1,51 +1,51 @@
|
|||
<Project>
|
||||
<PropertyGroup>
|
||||
<ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
|
||||
<CentralPackageTransitivePinningEnabled>true</CentralPackageTransitivePinningEnabled>
|
||||
</PropertyGroup>
|
||||
<ItemGroup>
|
||||
<PackageVersion Include="BenchmarkDotNet" Version="0.14.0" />
|
||||
<PackageVersion Include="coverlet.collector" Version="6.0.2" />
|
||||
<PackageVersion Include="coverlet.msbuild" Version="6.0.2" />
|
||||
<PackageVersion Include="FluentAssertions" Version="5.9.0" />
|
||||
<PackageVersion Include="GitHubActionsTestLogger" Version="1.1.2" />
|
||||
<PackageVersion Include="Google.Api.CommonProtos" Version="2.2.0" />
|
||||
<PackageVersion Include="Google.Protobuf" Version="3.28.2" />
|
||||
<PackageVersion Include="Grpc.AspNetCore" Version="2.66.0" />
|
||||
<PackageVersion Include="Grpc.Core.Testing" Version="2.46.6" />
|
||||
<PackageVersion Include="Grpc.Net.Client" Version="2.66.0" />
|
||||
<PackageVersion Include="Grpc.Net.ClientFactory" Version="2.66.0" />
|
||||
<PackageVersion Include="Grpc.Tools" Version="2.67.0" />
|
||||
<PackageVersion Include="Microsoft.AspNetCore.Mvc.Testing" Version="6.0.35" />
|
||||
<PackageVersion Include="Microsoft.AspNetCore.TestHost" Version="6.0.35" />
|
||||
<PackageVersion Include="Microsoft.CodeAnalysis.Analyzers" Version="3.3.4" />
|
||||
<PackageVersion Include="Microsoft.CodeAnalysis.Common" Version="4.8.0" />
|
||||
<PackageVersion Include="Microsoft.CodeAnalysis.CSharp" Version="4.8.0" />
|
||||
<PackageVersion Include="Microsoft.CodeAnalysis.CSharp.SourceGenerators.Testing" Version="1.1.2" />
|
||||
<PackageVersion Include="Microsoft.CodeAnalysis.CSharp.SourceGenerators.Testing.XUnit" Version="1.1.2" />
|
||||
<PackageVersion Include="Microsoft.CodeAnalysis.CSharp.Workspaces" Version="4.8.0" />
|
||||
<PackageVersion Include="Microsoft.DurableTask.Client.Grpc" Version="1.3.0" />
|
||||
<PackageVersion Include="Microsoft.DurableTask.Worker.Grpc" Version="1.3.0" />
|
||||
<PackageVersion Include="Microsoft.Extensions.Configuration" Version="6.0.1" />
|
||||
<PackageVersion Include="Microsoft.Extensions.Configuration.Abstractions" Version="6.0.0" />
|
||||
<PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="6.0.1" />
|
||||
<PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="6.0.0" />
|
||||
<PackageVersion Include="Microsoft.Extensions.Http" Version="6.0.0" />
|
||||
<PackageVersion Include="Microsoft.Extensions.Logging" Version="6.0.0" />
|
||||
<PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="6.0.0" />
|
||||
<PackageVersion Include="Microsoft.NET.Test.Sdk" Version="16.8.3" />
|
||||
<PackageVersion Include="Microsoft.SourceLink.GitHub" Version="1.1.1" />
|
||||
<PackageVersion Include="MinVer" Version="2.3.0" />
|
||||
<PackageVersion Include="Moq" Version="4.20.72" />
|
||||
<PackageVersion Include="Newtonsoft.Json" Version="13.0.3" />
|
||||
<PackageVersion Include="protobuf-net.Grpc.AspNetCore" Version="1.2.2" />
|
||||
<PackageVersion Include="Serilog.AspNetCore" Version="6.1.0" />
|
||||
<PackageVersion Include="Serilog.Sinks.Console" Version="4.1.0" />
|
||||
<PackageVersion Include="Serilog.Sinks.File" Version="5.0.0" />
|
||||
<PackageVersion Include="System.Formats.Asn1" Version="6.0.1" />
|
||||
<PackageVersion Include="System.Text.Json" Version="8.0.5" />
|
||||
<PackageVersion Include="xunit" Version="2.9.2" />
|
||||
<PackageVersion Include="xunit.extensibility.core" Version="2.9.2" />
|
||||
<PackageVersion Include="xunit.runner.visualstudio" Version="2.8.2" />
|
||||
</ItemGroup>
|
||||
<PropertyGroup>
|
||||
<ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
|
||||
<CentralPackageTransitivePinningEnabled>true</CentralPackageTransitivePinningEnabled>
|
||||
</PropertyGroup>
|
||||
<ItemGroup>
|
||||
<PackageVersion Include="BenchmarkDotNet" Version="0.14.0" />
|
||||
<PackageVersion Include="coverlet.collector" Version="6.0.2" />
|
||||
<PackageVersion Include="coverlet.msbuild" Version="6.0.2" />
|
||||
<PackageVersion Include="FluentAssertions" Version="5.9.0" />
|
||||
<PackageVersion Include="GitHubActionsTestLogger" Version="1.1.2" />
|
||||
<PackageVersion Include="Google.Api.CommonProtos" Version="2.2.0" />
|
||||
<PackageVersion Include="Google.Protobuf" Version="3.28.2" />
|
||||
<PackageVersion Include="Grpc.AspNetCore" Version="2.66.0" />
|
||||
<PackageVersion Include="Grpc.Core.Testing" Version="2.46.6" />
|
||||
<PackageVersion Include="Grpc.Net.Client" Version="2.66.0" />
|
||||
<PackageVersion Include="Grpc.Net.ClientFactory" Version="2.66.0" />
|
||||
<PackageVersion Include="Grpc.Tools" Version="2.67.0" />
|
||||
<PackageVersion Include="Microsoft.AspNetCore.Mvc.Testing" Version="6.0.35" />
|
||||
<PackageVersion Include="Microsoft.AspNetCore.TestHost" Version="6.0.35" />
|
||||
<PackageVersion Include="Microsoft.CodeAnalysis.Analyzers" Version="3.3.4" />
|
||||
<PackageVersion Include="Microsoft.CodeAnalysis.Common" Version="4.8.0" />
|
||||
<PackageVersion Include="Microsoft.CodeAnalysis.CSharp" Version="4.8.0" />
|
||||
<PackageVersion Include="Microsoft.CodeAnalysis.CSharp.SourceGenerators.Testing" Version="1.1.2" />
|
||||
<PackageVersion Include="Microsoft.CodeAnalysis.CSharp.SourceGenerators.Testing.XUnit" Version="1.1.2" />
|
||||
<PackageVersion Include="Microsoft.CodeAnalysis.CSharp.Workspaces" Version="4.8.0" />
|
||||
<PackageVersion Include="Microsoft.DurableTask.Client.Grpc" Version="1.3.0" />
|
||||
<PackageVersion Include="Microsoft.DurableTask.Worker.Grpc" Version="1.3.0" />
|
||||
<PackageVersion Include="Microsoft.Extensions.Configuration" Version="6.0.1" />
|
||||
<PackageVersion Include="Microsoft.Extensions.Configuration.Abstractions" Version="6.0.0" />
|
||||
<PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="6.0.1" />
|
||||
<PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="6.0.0" />
|
||||
<PackageVersion Include="Microsoft.Extensions.Logging" Version="6.0.0" />
|
||||
<PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="6.0.4" />
|
||||
<PackageVersion Include="Microsoft.Extensions.Http" Version="6.0.0" />
|
||||
<PackageVersion Include="Microsoft.NET.Test.Sdk" Version="16.8.3" />
|
||||
<PackageVersion Include="Microsoft.SourceLink.GitHub" Version="1.1.1" />
|
||||
<PackageVersion Include="MinVer" Version="2.3.0" />
|
||||
<PackageVersion Include="Moq" Version="4.20.72" />
|
||||
<PackageVersion Include="Newtonsoft.Json" Version="13.0.3" />
|
||||
<PackageVersion Include="protobuf-net.Grpc.AspNetCore" Version="1.2.2" />
|
||||
<PackageVersion Include="Serilog.AspNetCore" Version="6.1.0" />
|
||||
<PackageVersion Include="Serilog.Sinks.Console" Version="4.1.0" />
|
||||
<PackageVersion Include="Serilog.Sinks.File" Version="5.0.0" />
|
||||
<PackageVersion Include="System.Formats.Asn1" Version="6.0.1" />
|
||||
<PackageVersion Include="System.Text.Json" Version="6.0.10" />
|
||||
<PackageVersion Include="xunit" Version="2.9.2" />
|
||||
<PackageVersion Include="xunit.extensibility.core" Version="2.9.2" />
|
||||
<PackageVersion Include="xunit.runner.visualstudio" Version="2.8.2" />
|
||||
</ItemGroup>
|
||||
</Project>
|
||||
|
|
|
|||
20
all.sln
20
all.sln
|
|
@ -119,6 +119,11 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Dapr.Common", "src\Dapr.Com
|
|||
EndProject
|
||||
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Dapr.Common.Test", "test\Dapr.Common.Test\Dapr.Common.Test.csproj", "{CDB47863-BEBD-4841-A807-46D868962521}"
|
||||
EndProject
|
||||
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Dapr.Messaging.Test", "test\Dapr.Messaging.Test\Dapr.Messaging.Test.csproj", "{4E04EB35-7FD2-4FDB-B09A-F75CE24053B9}"
|
||||
EndProject
|
||||
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Dapr.Messaging", "src\Dapr.Messaging\Dapr.Messaging.csproj", "{0EAE36A1-B578-4F13-A113-7A477ECA1BDA}"
|
||||
EndProject
|
||||
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "StreamingSubscriptionExample", "examples\Client\PublishSubscribe\StreamingSubscriptionExample\StreamingSubscriptionExample.csproj", "{290D1278-F613-4DF3-9DF5-F37E38CDC363}"
|
||||
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Dapr.Jobs", "src\Dapr.Jobs\Dapr.Jobs.csproj", "{C8BB6A85-A7EA-40C0-893D-F36F317829B3}"
|
||||
EndProject
|
||||
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Dapr.Jobs.Test", "test\Dapr.Jobs.Test\Dapr.Jobs.Test.csproj", "{BF9828E9-5597-4D42-AA6E-6E6C12214204}"
|
||||
|
|
@ -311,6 +316,18 @@ Global
|
|||
{CDB47863-BEBD-4841-A807-46D868962521}.Debug|Any CPU.Build.0 = Debug|Any CPU
|
||||
{CDB47863-BEBD-4841-A807-46D868962521}.Release|Any CPU.ActiveCfg = Release|Any CPU
|
||||
{CDB47863-BEBD-4841-A807-46D868962521}.Release|Any CPU.Build.0 = Release|Any CPU
|
||||
{4E04EB35-7FD2-4FDB-B09A-F75CE24053B9}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
|
||||
{4E04EB35-7FD2-4FDB-B09A-F75CE24053B9}.Debug|Any CPU.Build.0 = Debug|Any CPU
|
||||
{4E04EB35-7FD2-4FDB-B09A-F75CE24053B9}.Release|Any CPU.ActiveCfg = Release|Any CPU
|
||||
{4E04EB35-7FD2-4FDB-B09A-F75CE24053B9}.Release|Any CPU.Build.0 = Release|Any CPU
|
||||
{0EAE36A1-B578-4F13-A113-7A477ECA1BDA}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
|
||||
{0EAE36A1-B578-4F13-A113-7A477ECA1BDA}.Debug|Any CPU.Build.0 = Debug|Any CPU
|
||||
{0EAE36A1-B578-4F13-A113-7A477ECA1BDA}.Release|Any CPU.ActiveCfg = Release|Any CPU
|
||||
{0EAE36A1-B578-4F13-A113-7A477ECA1BDA}.Release|Any CPU.Build.0 = Release|Any CPU
|
||||
{290D1278-F613-4DF3-9DF5-F37E38CDC363}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
|
||||
{290D1278-F613-4DF3-9DF5-F37E38CDC363}.Debug|Any CPU.Build.0 = Debug|Any CPU
|
||||
{290D1278-F613-4DF3-9DF5-F37E38CDC363}.Release|Any CPU.ActiveCfg = Release|Any CPU
|
||||
{290D1278-F613-4DF3-9DF5-F37E38CDC363}.Release|Any CPU.Build.0 = Release|Any CPU
|
||||
{C8BB6A85-A7EA-40C0-893D-F36F317829B3}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
|
||||
{C8BB6A85-A7EA-40C0-893D-F36F317829B3}.Debug|Any CPU.Build.0 = Debug|Any CPU
|
||||
{C8BB6A85-A7EA-40C0-893D-F36F317829B3}.Release|Any CPU.ActiveCfg = Release|Any CPU
|
||||
|
|
@ -379,6 +396,9 @@ Global
|
|||
{DFBABB04-50E9-42F6-B470-310E1B545638} = {27C5D71D-0721-4221-9286-B94AB07B58CF}
|
||||
{B445B19C-A925-4873-8CB7-8317898B6970} = {27C5D71D-0721-4221-9286-B94AB07B58CF}
|
||||
{CDB47863-BEBD-4841-A807-46D868962521} = {DD020B34-460F-455F-8D17-CF4A949F100B}
|
||||
{4E04EB35-7FD2-4FDB-B09A-F75CE24053B9} = {DD020B34-460F-455F-8D17-CF4A949F100B}
|
||||
{0EAE36A1-B578-4F13-A113-7A477ECA1BDA} = {27C5D71D-0721-4221-9286-B94AB07B58CF}
|
||||
{290D1278-F613-4DF3-9DF5-F37E38CDC363} = {0EF6EA64-D7C3-420D-9890-EAE8D54A57E6}
|
||||
{C8BB6A85-A7EA-40C0-893D-F36F317829B3} = {27C5D71D-0721-4221-9286-B94AB07B58CF}
|
||||
{BF9828E9-5597-4D42-AA6E-6E6C12214204} = {DD020B34-460F-455F-8D17-CF4A949F100B}
|
||||
{D9697361-232F-465D-A136-4561E0E88488} = {D687DDC4-66C5-4667-9E3A-FD8B78ECAA78}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,35 @@
|
|||
using System.Text;
|
||||
using Dapr.Messaging.PublishSubscribe;
|
||||
using Dapr.Messaging.PublishSubscribe.Extensions;
|
||||
|
||||
var builder = WebApplication.CreateBuilder(args);
|
||||
builder.Services.AddDaprPubSubClient();
|
||||
var app = builder.Build();
|
||||
|
||||
//Process each message returned from the subscription
|
||||
Task<TopicResponseAction> HandleMessageAsync(TopicMessage message, CancellationToken cancellationToken = default)
|
||||
{
|
||||
try
|
||||
{
|
||||
//Do something with the message
|
||||
Console.WriteLine(Encoding.UTF8.GetString(message.Data.Span));
|
||||
return Task.FromResult(TopicResponseAction.Success);
|
||||
}
|
||||
catch
|
||||
{
|
||||
return Task.FromResult(TopicResponseAction.Retry);
|
||||
}
|
||||
}
|
||||
|
||||
var messagingClient = app.Services.GetRequiredService<DaprPublishSubscribeClient>();
|
||||
|
||||
//Create a dynamic streaming subscription and subscribe with a timeout of 30 seconds and 10 seconds for message handling
|
||||
var cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
|
||||
var subscription = await messagingClient.SubscribeAsync("pubsub", "myTopic",
|
||||
new DaprSubscriptionOptions(new MessageHandlingPolicy(TimeSpan.FromSeconds(10), TopicResponseAction.Retry)),
|
||||
HandleMessageAsync, cancellationTokenSource.Token);
|
||||
|
||||
await Task.Delay(TimeSpan.FromMinutes(1));
|
||||
|
||||
//When you're done with the subscription, simply dispose of it
|
||||
await subscription.DisposeAsync();
|
||||
|
|
@ -0,0 +1,14 @@
|
|||
<Project Sdk="Microsoft.NET.Sdk.Web">
|
||||
|
||||
<PropertyGroup>
|
||||
<OutputType>Exe</OutputType>
|
||||
<TargetFramework>net6.0</TargetFramework>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
<Nullable>enable</Nullable>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\..\..\src\Dapr.Messaging\Dapr.Messaging.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
|
|
@ -0,0 +1,22 @@
|
|||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<PropertyGroup>
|
||||
<Description>This package contains the reference assemblies for developing messaging services using Dapr.</Description>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
<Nullable>enable</Nullable>
|
||||
<PackageId>Dapr.Messaging</PackageId>
|
||||
<Title>Dapr Messaging SDK</Title>
|
||||
<Description>Dapr Messaging SDK for building applications that utilize messaging components.</Description>
|
||||
<VersionSuffix>alpha</VersionSuffix>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Microsoft.Extensions.DependencyInjection" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\Dapr.Common\Dapr.Common.csproj" />
|
||||
<ProjectReference Include="..\Dapr.Protos\Dapr.Protos.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
|
|
@ -0,0 +1,31 @@
|
|||
// ------------------------------------------------------------------------
|
||||
// Copyright 2024 The Dapr Authors
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
// ------------------------------------------------------------------------
|
||||
|
||||
namespace Dapr.Messaging.PublishSubscribe;
|
||||
|
||||
/// <summary>
|
||||
/// The base implementation of a Dapr pub/sub client.
|
||||
/// </summary>
|
||||
public abstract class DaprPublishSubscribeClient
|
||||
{
|
||||
/// <summary>
|
||||
/// Dynamically subscribes to a Publish/Subscribe component and topic.
|
||||
/// </summary>
|
||||
/// <param name="pubSubName">The name of the Publish/Subscribe component.</param>
|
||||
/// <param name="topicName">The name of the topic to subscribe to.</param>
|
||||
/// <param name="options">Configuration options.</param>
|
||||
/// <param name="messageHandler">The delegate reflecting the action to take upon messages received by the subscription.</param>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
/// <returns></returns>
|
||||
public abstract Task<IAsyncDisposable> SubscribeAsync(string pubSubName, string topicName, DaprSubscriptionOptions options, TopicMessageHandler messageHandler, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
|
@ -0,0 +1,47 @@
|
|||
// ------------------------------------------------------------------------
|
||||
// Copyright 2024 The Dapr Authors
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
// ------------------------------------------------------------------------
|
||||
|
||||
using Dapr.Common;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Autogenerated = Dapr.Client.Autogen.Grpc.v1;
|
||||
|
||||
namespace Dapr.Messaging.PublishSubscribe;
|
||||
|
||||
/// <summary>
|
||||
/// Builds a <see cref="DaprPublishSubscribeClient"/>.
|
||||
/// </summary>
|
||||
public sealed class DaprPublishSubscribeClientBuilder : DaprGenericClientBuilder<DaprPublishSubscribeClient>
|
||||
{
|
||||
/// <summary>
|
||||
/// Used to initialize a new instance of the <see cref="DaprPublishSubscribeClientBuilder"/>.
|
||||
/// </summary>
|
||||
/// <param name="configuration">An optional instance of <see cref="IConfiguration"/>.</param>
|
||||
public DaprPublishSubscribeClientBuilder(IConfiguration? configuration = null) : base(configuration)
|
||||
{
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Builds the client instance from the properties of the builder.
|
||||
/// </summary>
|
||||
/// <returns>The Dapr client instance.</returns>
|
||||
/// <summary>
|
||||
/// Builds the client instance from the properties of the builder.
|
||||
/// </summary>
|
||||
public override DaprPublishSubscribeClient Build()
|
||||
{
|
||||
var daprClientDependencies = BuildDaprClientDependencies();
|
||||
var client = new Autogenerated.Dapr.DaprClient(daprClientDependencies.channel);
|
||||
|
||||
return new DaprPublishSubscribeGrpcClient(client);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,49 @@
|
|||
// ------------------------------------------------------------------------
|
||||
// Copyright 2024 The Dapr Authors
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
// ------------------------------------------------------------------------
|
||||
|
||||
using P = Dapr.Client.Autogen.Grpc.v1.Dapr;
|
||||
|
||||
namespace Dapr.Messaging.PublishSubscribe;
|
||||
|
||||
/// <summary>
|
||||
/// A client for interacting with the Dapr endpoints.
|
||||
/// </summary>
|
||||
internal sealed class DaprPublishSubscribeGrpcClient : DaprPublishSubscribeClient
|
||||
{
|
||||
private readonly P.DaprClient daprClient;
|
||||
|
||||
/// <summary>
|
||||
/// Creates a new instance of a <see cref="DaprPublishSubscribeGrpcClient"/>
|
||||
/// </summary>
|
||||
public DaprPublishSubscribeGrpcClient(P.DaprClient client)
|
||||
{
|
||||
daprClient = client;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Dynamically subscribes to a Publish/Subscribe component and topic.
|
||||
/// </summary>
|
||||
/// <param name="pubSubName">The name of the Publish/Subscribe component.</param>
|
||||
/// <param name="topicName">The name of the topic to subscribe to.</param>
|
||||
/// <param name="options">Configuration options.</param>
|
||||
/// <param name="messageHandler">The delegate reflecting the action to take upon messages received by the subscription.</param>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
/// <returns></returns>
|
||||
public override async Task<IAsyncDisposable> SubscribeAsync(string pubSubName, string topicName, DaprSubscriptionOptions options, TopicMessageHandler messageHandler, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var receiver = new PublishSubscribeReceiver(pubSubName, topicName, options, messageHandler, daprClient);
|
||||
await receiver.SubscribeAsync(cancellationToken);
|
||||
return receiver;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -0,0 +1,44 @@
|
|||
// ------------------------------------------------------------------------
|
||||
// Copyright 2024 The Dapr Authors
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
// ------------------------------------------------------------------------
|
||||
|
||||
namespace Dapr.Messaging.PublishSubscribe;
|
||||
|
||||
/// <summary>
|
||||
/// Options used to configure the dynamic Dapr subscription.
|
||||
/// </summary>
|
||||
/// <param name="MessageHandlingPolicy">Describes the policy to take on messages that have not been acknowledged within the timeout period.</param>
|
||||
public sealed record DaprSubscriptionOptions(MessageHandlingPolicy MessageHandlingPolicy)
|
||||
{
|
||||
/// <summary>
|
||||
/// Subscription metadata.
|
||||
/// </summary>
|
||||
public IReadOnlyDictionary<string, string> Metadata { get; init; } = new Dictionary<string, string>();
|
||||
|
||||
/// <summary>
|
||||
/// The optional name of the dead-letter topic to send unprocessed messages to.
|
||||
/// </summary>
|
||||
public string? DeadLetterTopic { get; init; }
|
||||
|
||||
/// <summary>
|
||||
/// If populated, this reflects the maximum number of messages that can be queued for processing on the replica. By default,
|
||||
/// no maximum boundary is enforced.
|
||||
/// </summary>
|
||||
public int? MaximumQueuedMessages { get; init; }
|
||||
|
||||
/// <summary>
|
||||
/// The maximum amount of time to take to dispose of acknowledgement messages after the cancellation token has
|
||||
/// been signaled.
|
||||
/// </summary>
|
||||
public TimeSpan MaximumCleanupTimeout { get; init; } = TimeSpan.FromSeconds(30);
|
||||
}
|
||||
|
||||
|
|
@ -0,0 +1,38 @@
|
|||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.DependencyInjection.Extensions;
|
||||
|
||||
namespace Dapr.Messaging.PublishSubscribe.Extensions;
|
||||
|
||||
/// <summary>
|
||||
/// Contains extension methods for using Dapr Publish/Subscribe with dependency injection.
|
||||
/// </summary>
|
||||
public static class PublishSubscribeServiceCollectionExtensions
|
||||
{
|
||||
/// <summary>
|
||||
/// Adds Dapr Publish/Subscribe support to the service collection.
|
||||
/// </summary>
|
||||
/// <param name="services">The <see cref="IServiceCollection"/>.</param>
|
||||
/// <param name="configure">Optionally allows greater configuration of the <see cref="DaprPublishSubscribeClient"/> using injected services.</param>
|
||||
/// <returns></returns>
|
||||
public static IServiceCollection AddDaprPubSubClient(this IServiceCollection services, Action<IServiceProvider, DaprPublishSubscribeClientBuilder>? configure = null)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(services, nameof(services));
|
||||
|
||||
//Register the IHttpClientFactory implementation
|
||||
services.AddHttpClient();
|
||||
|
||||
services.TryAddSingleton(serviceProvider =>
|
||||
{
|
||||
var httpClientFactory = serviceProvider.GetRequiredService<IHttpClientFactory>();
|
||||
|
||||
var builder = new DaprPublishSubscribeClientBuilder();
|
||||
builder.UseHttpClientFactory(httpClientFactory);
|
||||
|
||||
configure?.Invoke(serviceProvider, builder);
|
||||
|
||||
return builder.Build();
|
||||
});
|
||||
|
||||
return services;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,23 @@
|
|||
// ------------------------------------------------------------------------
|
||||
// Copyright 2024 The Dapr Authors
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
// ------------------------------------------------------------------------
|
||||
|
||||
|
||||
namespace Dapr.Messaging.PublishSubscribe;
|
||||
|
||||
/// <summary>
|
||||
/// Defines the policy for handling streaming message subscriptions, including retry logic and timeout settings.
|
||||
/// </summary>
|
||||
/// <param name="TimeoutDuration">The duration to wait before timing out a message handling operation.</param>
|
||||
/// <param name="DefaultResponseAction">The default action to take when a message handling operation times out.</param>
|
||||
public sealed record MessageHandlingPolicy(TimeSpan TimeoutDuration, TopicResponseAction DefaultResponseAction);
|
||||
|
||||
|
|
@ -0,0 +1,313 @@
|
|||
// ------------------------------------------------------------------------
|
||||
// Copyright 2024 The Dapr Authors
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
// ------------------------------------------------------------------------
|
||||
|
||||
using System.Threading.Channels;
|
||||
using Dapr.AppCallback.Autogen.Grpc.v1;
|
||||
using Grpc.Core;
|
||||
using P = Dapr.Client.Autogen.Grpc.v1;
|
||||
|
||||
namespace Dapr.Messaging.PublishSubscribe;
|
||||
|
||||
/// <summary>
|
||||
/// A thread-safe implementation of a receiver for messages from a specified Dapr publish/subscribe component and
|
||||
/// topic.
|
||||
/// </summary>
|
||||
internal sealed class PublishSubscribeReceiver : IAsyncDisposable
|
||||
{
|
||||
/// <summary>
|
||||
/// Provides options for the unbounded channel.
|
||||
/// </summary>
|
||||
private readonly static UnboundedChannelOptions UnboundedChannelOptions = new()
|
||||
{
|
||||
SingleWriter = true, SingleReader = true
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// The name of the Dapr publish/subscribe component.
|
||||
/// </summary>
|
||||
private readonly string pubSubName;
|
||||
/// <summary>
|
||||
/// The name of the topic to subscribe to.
|
||||
/// </summary>
|
||||
private readonly string topicName;
|
||||
/// <summary>
|
||||
/// Options allowing the behavior of the receiver to be configured.
|
||||
/// </summary>
|
||||
private readonly DaprSubscriptionOptions options;
|
||||
/// <summary>
|
||||
/// A channel used to decouple the messages received from the sidecar to their consumption.
|
||||
/// </summary>
|
||||
private readonly Channel<TopicMessage> topicMessagesChannel;
|
||||
/// <summary>
|
||||
/// Maintains the various acknowledgements for each message.
|
||||
/// </summary>
|
||||
private readonly Channel<TopicAcknowledgement> acknowledgementsChannel = Channel.CreateUnbounded<TopicAcknowledgement>(UnboundedChannelOptions);
|
||||
/// <summary>
|
||||
/// The stream connection between this instance and the Dapr sidecar.
|
||||
/// </summary>
|
||||
private AsyncDuplexStreamingCall<P.SubscribeTopicEventsRequestAlpha1, P.SubscribeTopicEventsResponseAlpha1>? clientStream;
|
||||
/// <summary>
|
||||
/// Used to ensure thread-safe operations against the stream.
|
||||
/// </summary>
|
||||
private readonly SemaphoreSlim semaphore = new(1, 1);
|
||||
/// <summary>
|
||||
/// The handler delegate responsible for processing the topic messages.
|
||||
/// </summary>
|
||||
private readonly TopicMessageHandler messageHandler;
|
||||
/// <summary>
|
||||
/// A reference to the DaprClient instance.
|
||||
/// </summary>
|
||||
private readonly P.Dapr.DaprClient client;
|
||||
/// <summary>
|
||||
/// Flag that prevents the developer from accidentally initializing the subscription more than once from the same receiver.
|
||||
/// </summary>
|
||||
private int hasInitialized;
|
||||
/// <summary>
|
||||
/// Flag that ensures the instance is only disposed a single time.
|
||||
/// </summary>
|
||||
private bool isDisposed;
|
||||
|
||||
/// <summary>
|
||||
/// Constructs a new instance of a <see cref="PublishSubscribeReceiver"/> instance.
|
||||
/// </summary>
|
||||
/// <param name="pubSubName">The name of the Dapr Publish/Subscribe component.</param>
|
||||
/// <param name="topicName">The name of the topic to subscribe to.</param>
|
||||
/// <param name="options">Options allowing the behavior of the receiver to be configured.</param>
|
||||
/// <param name="handler">The delegate reflecting the action to take upon messages received by the subscription.</param>
|
||||
/// <param name="client">A reference to the DaprClient instance.</param>
|
||||
internal PublishSubscribeReceiver(string pubSubName, string topicName, DaprSubscriptionOptions options, TopicMessageHandler handler, P.Dapr.DaprClient client)
|
||||
{
|
||||
this.client = client;
|
||||
this.pubSubName = pubSubName;
|
||||
this.topicName = topicName;
|
||||
this.options = options;
|
||||
this.messageHandler = handler;
|
||||
topicMessagesChannel = options.MaximumQueuedMessages is > 0
|
||||
? Channel.CreateBounded<TopicMessage>(new BoundedChannelOptions((int)options.MaximumQueuedMessages)
|
||||
{
|
||||
SingleReader = true, SingleWriter = true, FullMode = BoundedChannelFullMode.Wait
|
||||
})
|
||||
: Channel.CreateUnbounded<TopicMessage>(UnboundedChannelOptions);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Dynamically subscribes to messages on a PubSub topic provided by the Dapr sidecar.
|
||||
/// </summary>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
/// <returns>An <see cref="IAsyncEnumerable{TopicMessage}"/> containing messages provided by the sidecar.</returns>
|
||||
internal async Task SubscribeAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
//Prevents the receiver from performing the subscribe operation more than once (as the multiple initialization messages would cancel the stream).
|
||||
if (Interlocked.Exchange(ref hasInitialized, 1) == 1)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var stream = await GetStreamAsync(cancellationToken);
|
||||
|
||||
//Retrieve the messages from the sidecar and write to the messages channel
|
||||
var fetchMessagesTask = FetchDataFromSidecarAsync(stream, topicMessagesChannel.Writer, cancellationToken);
|
||||
|
||||
//Process the messages as they're written to either channel
|
||||
var acknowledgementProcessorTask = ProcessAcknowledgementChannelMessagesAsync(stream, cancellationToken);
|
||||
var topicMessageProcessorTask = ProcessTopicChannelMessagesAsync(cancellationToken);
|
||||
|
||||
try
|
||||
{
|
||||
await Task.WhenAll(fetchMessagesTask, acknowledgementProcessorTask, topicMessageProcessorTask);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Will be cleaned up during DisposeAsync
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Retrieves or creates the bidirectional stream to the DaprClient for transacting pub/sub subscriptions.
|
||||
/// </summary>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
/// <returns>The stream connection.</returns>
|
||||
private async Task<AsyncDuplexStreamingCall<P.SubscribeTopicEventsRequestAlpha1, P.SubscribeTopicEventsResponseAlpha1>> GetStreamAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
await semaphore.WaitAsync(cancellationToken);
|
||||
|
||||
try
|
||||
{
|
||||
return clientStream ??= client.SubscribeTopicEventsAlpha1(cancellationToken: cancellationToken);
|
||||
}
|
||||
finally
|
||||
{
|
||||
semaphore.Release();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Acknowledges the indicated message back to the Dapr sidecar with an indicated behavior to take on the message.
|
||||
/// </summary>
|
||||
/// <param name="messageId">The identifier of the message the behavior is in reference to.</param>
|
||||
/// <param name="behavior">The behavior to take on the message as indicated by either the message handler or timeout message handling configuration.</param>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
/// <returns></returns>
|
||||
private async Task AcknowledgeMessageAsync(string messageId, TopicResponseAction behavior, CancellationToken cancellationToken)
|
||||
{
|
||||
var action = behavior switch
|
||||
{
|
||||
TopicResponseAction.Success => TopicEventResponse.Types.TopicEventResponseStatus.Success,
|
||||
TopicResponseAction.Retry => TopicEventResponse.Types.TopicEventResponseStatus.Retry,
|
||||
TopicResponseAction.Drop => TopicEventResponse.Types.TopicEventResponseStatus.Drop,
|
||||
_ => throw new InvalidOperationException(
|
||||
$"Unrecognized topic acknowledgement action: {behavior}")
|
||||
};
|
||||
|
||||
var acknowledgement = new TopicAcknowledgement(messageId, action);
|
||||
await acknowledgementsChannel.Writer.WriteAsync(acknowledgement, cancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Processes each acknowledgement from the acknowledgement channel reader as it's populated.
|
||||
/// </summary>
|
||||
/// <param name="messageStream">The stream used to interact with the Dapr sidecar.</param>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
private async Task ProcessAcknowledgementChannelMessagesAsync(AsyncDuplexStreamingCall<P.SubscribeTopicEventsRequestAlpha1, P.SubscribeTopicEventsResponseAlpha1> messageStream, CancellationToken cancellationToken)
|
||||
{
|
||||
await foreach (var acknowledgement in acknowledgementsChannel.Reader.ReadAllAsync(cancellationToken))
|
||||
{
|
||||
await messageStream.RequestStream.WriteAsync(new P.SubscribeTopicEventsRequestAlpha1
|
||||
{
|
||||
EventProcessed = new P.SubscribeTopicEventsRequestProcessedAlpha1
|
||||
{
|
||||
Id = acknowledgement.MessageId,
|
||||
Status = new TopicEventResponse { Status = acknowledgement.Action }
|
||||
}
|
||||
}, cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Processes each topic messages from the channel as it's populated.
|
||||
/// </summary>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
private async Task ProcessTopicChannelMessagesAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
await foreach (var message in topicMessagesChannel.Reader.ReadAllAsync(cancellationToken))
|
||||
{
|
||||
using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
cts.CancelAfter(options.MessageHandlingPolicy.TimeoutDuration);
|
||||
|
||||
//Evaluate the message and return an acknowledgement result
|
||||
var messageAction = await messageHandler(message, cts.Token);
|
||||
|
||||
try
|
||||
{
|
||||
//Share the result with the sidecar
|
||||
await AcknowledgeMessageAsync(message.Id, messageAction, cancellationToken);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
//Acknowledge the message using the configured default response action
|
||||
await AcknowledgeMessageAsync(message.Id, options.MessageHandlingPolicy.DefaultResponseAction, cancellationToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Retrieves the subscription stream data from the Dapr sidecar.
|
||||
/// </summary>
|
||||
/// <param name="stream">The stream connection to and from the Dapr sidecar instance.</param>
|
||||
/// <param name="channelWriter">The channel writer instance.</param>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
private async Task FetchDataFromSidecarAsync(
|
||||
AsyncDuplexStreamingCall<P.SubscribeTopicEventsRequestAlpha1, P.SubscribeTopicEventsResponseAlpha1> stream,
|
||||
ChannelWriter<TopicMessage> channelWriter, CancellationToken cancellationToken)
|
||||
{
|
||||
//Build out the initial topic events request
|
||||
var initialRequest = new P.SubscribeTopicEventsRequestInitialAlpha1()
|
||||
{
|
||||
PubsubName = pubSubName, DeadLetterTopic = options.DeadLetterTopic ?? string.Empty, Topic = topicName
|
||||
};
|
||||
|
||||
if (options.Metadata.Count > 0)
|
||||
{
|
||||
foreach (var (key, value) in options.Metadata)
|
||||
{
|
||||
initialRequest.Metadata.Add(key, value);
|
||||
}
|
||||
}
|
||||
|
||||
//Send this request to the Dapr sidecar
|
||||
await stream.RequestStream.WriteAsync(
|
||||
new P.SubscribeTopicEventsRequestAlpha1 { InitialRequest = initialRequest }, cancellationToken);
|
||||
|
||||
//Each time a message is received from the stream, push it into the topic messages channel
|
||||
await foreach (var response in stream.ResponseStream.ReadAllAsync(cancellationToken))
|
||||
{
|
||||
var message =
|
||||
new TopicMessage(response.EventMessage.Id, response.EventMessage.Source, response.EventMessage.Type,
|
||||
response.EventMessage.SpecVersion, response.EventMessage.DataContentType,
|
||||
response.EventMessage.Topic, response.EventMessage.PubsubName)
|
||||
{
|
||||
Path = response.EventMessage.Path,
|
||||
Extensions = response.EventMessage.Extensions.Fields.ToDictionary(f => f.Key, kvp => kvp.Value)
|
||||
};
|
||||
|
||||
try
|
||||
{
|
||||
await channelWriter.WaitToWriteAsync(cancellationToken);
|
||||
await channelWriter.WriteAsync(message, cancellationToken);
|
||||
}
|
||||
catch (Exception)
|
||||
{
|
||||
// Handle being unable to write because the writer is completed due to an active DisposeAsync operation
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Disposes the various resources associated with the instance.
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
if (isDisposed)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
isDisposed = true;
|
||||
|
||||
//Stop processing new events - we'll leave any messages yet unseen as unprocessed and
|
||||
//Dapr will handle as necessary when they're not acknowledged
|
||||
topicMessagesChannel.Writer.Complete();
|
||||
|
||||
//Flush the remaining acknowledgements, but start by marking the writer as complete so it doesn't receive any other messages either
|
||||
acknowledgementsChannel.Writer.Complete();
|
||||
|
||||
try
|
||||
{
|
||||
//Process any remaining acknowledgements on the channel without exceeding the configured maximum clean up time
|
||||
await acknowledgementsChannel.Reader.Completion.WaitAsync(options.MaximumCleanupTimeout);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
//Handled
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Reflects the action to take on a given message identifier.
|
||||
/// </summary>
|
||||
/// <param name="MessageId">The identifier of the message.</param>
|
||||
/// <param name="Action">The action to take on the message in the acknowledgement request.</param>
|
||||
private sealed record TopicAcknowledgement(string MessageId, TopicEventResponse.Types.TopicEventResponseStatus Action);
|
||||
}
|
||||
|
||||
|
|
@ -0,0 +1,46 @@
|
|||
// ------------------------------------------------------------------------
|
||||
// Copyright 2024 The Dapr Authors
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
// ------------------------------------------------------------------------
|
||||
|
||||
using Google.Protobuf.WellKnownTypes;
|
||||
|
||||
namespace Dapr.Messaging.PublishSubscribe;
|
||||
|
||||
/// <summary>
|
||||
/// A message retrieved from a Dapr publish/subscribe topic.
|
||||
/// </summary>
|
||||
/// <param name="Id">The unique identifier of the topic message.</param>
|
||||
/// <param name="Source">Identifies the context in which an event happened, such as the organization publishing the
|
||||
/// event or the process that produced the event. The exact syntax and semantics behind the data
|
||||
/// encoded in the URI is defined by the event producer.</param>
|
||||
/// <param name="Type">The type of event related to the originating occurrence.</param>
|
||||
/// <param name="SpecVersion">The spec version of the CloudEvents specification.</param>
|
||||
/// <param name="DataContentType">The content type of the data.</param>
|
||||
/// <param name="Topic">The name of the topic.</param>
|
||||
/// <param name="PubSubName">The name of the Dapr publish/subscribe component.</param>
|
||||
public sealed record TopicMessage(string Id, string Source, string Type, string SpecVersion, string DataContentType, string Topic, string PubSubName)
|
||||
{
|
||||
/// <summary>
|
||||
/// The content of the event.
|
||||
/// </summary>
|
||||
public ReadOnlyMemory<byte> Data { get; init; }
|
||||
|
||||
/// <summary>
|
||||
/// The matching path from the topic subscription/routes (if specified) for this event.
|
||||
/// </summary>
|
||||
public string? Path { get; init; }
|
||||
|
||||
/// <summary>
|
||||
/// A map of additional custom properties sent to the app. These are considered to be CloudEvent extensions.
|
||||
/// </summary>
|
||||
public IReadOnlyDictionary<string, Value> Extensions { get; init; } = new Dictionary<string, Value>();
|
||||
}
|
||||
|
|
@ -0,0 +1,23 @@
|
|||
// ------------------------------------------------------------------------
|
||||
// Copyright 2024 The Dapr Authors
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
// ------------------------------------------------------------------------
|
||||
|
||||
namespace Dapr.Messaging.PublishSubscribe;
|
||||
|
||||
/// <summary>
|
||||
/// The handler delegate responsible for processing the topic message.
|
||||
/// </summary>
|
||||
/// <param name="request">The message request to process.</param>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
/// <returns>The acknowledgement behavior to report back to the pub/sub endpoint about the message.</returns>
|
||||
public delegate Task<TopicResponseAction> TopicMessageHandler(TopicMessage request,
|
||||
CancellationToken cancellationToken = default);
|
||||
|
|
@ -0,0 +1,34 @@
|
|||
// ------------------------------------------------------------------------
|
||||
// Copyright 2024 The Dapr Authors
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
// ------------------------------------------------------------------------
|
||||
|
||||
namespace Dapr.Messaging.PublishSubscribe;
|
||||
|
||||
/// <summary>
|
||||
/// Describes the various actions that can be taken on a topic message.
|
||||
/// </summary>
|
||||
public enum TopicResponseAction
|
||||
{
|
||||
/// <summary>
|
||||
/// Indicates the message was processed successfully and should be deleted from the pub/sub topic.
|
||||
/// </summary>
|
||||
Success,
|
||||
/// <summary>
|
||||
/// Indicates a failure while processing the message and that the message should be resent for a retry.
|
||||
/// </summary>
|
||||
Retry,
|
||||
/// <summary>
|
||||
/// Indicates a failure while processing the message and that the message should be dropped or sent to the
|
||||
/// dead-letter topic if specified.
|
||||
/// </summary>
|
||||
Drop
|
||||
}
|
||||
|
|
@ -0,0 +1,94 @@
|
|||
using System;
|
||||
using System.Text.Json;
|
||||
using Xunit;
|
||||
|
||||
namespace Dapr.Common.Test;
|
||||
|
||||
public class DaprGenericClientBuilderTest
|
||||
{
|
||||
[Fact]
|
||||
public void DaprGenericClientBuilder_ShouldUpdateHttpEndpoint_WhenHttpEndpointIsProvided()
|
||||
{
|
||||
// Arrange
|
||||
var builder = new SampleDaprGenericClientBuilder();
|
||||
const string endpointValue = "http://sample-endpoint";
|
||||
|
||||
// Act
|
||||
builder.UseHttpEndpoint(endpointValue);
|
||||
|
||||
// Assert
|
||||
Assert.Equal(endpointValue, builder.HttpEndpoint);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void DaprGenericClientBuilder_ShouldUpdateHttpEndpoint_WhenGrpcEndpointIsProvided()
|
||||
{
|
||||
// Arrange
|
||||
var builder = new SampleDaprGenericClientBuilder();
|
||||
const string endpointValue = "http://sample-endpoint";
|
||||
|
||||
// Act
|
||||
builder.UseGrpcEndpoint(endpointValue);
|
||||
|
||||
// Assert
|
||||
Assert.Equal(endpointValue, builder.GrpcEndpoint);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void DaprGenericClientBuilder_ShouldUpdateJsonSerializerOptions()
|
||||
{
|
||||
// Arrange
|
||||
const int maxDepth = 8;
|
||||
const bool writeIndented = true;
|
||||
var builder = new SampleDaprGenericClientBuilder();
|
||||
var options = new JsonSerializerOptions
|
||||
{
|
||||
WriteIndented = writeIndented,
|
||||
MaxDepth = maxDepth
|
||||
};
|
||||
|
||||
// Act
|
||||
builder.UseJsonSerializationOptions(options);
|
||||
|
||||
// Assert
|
||||
Assert.Equal(writeIndented, builder.JsonSerializerOptions.WriteIndented);
|
||||
Assert.Equal(maxDepth, builder.JsonSerializerOptions.MaxDepth);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void DaprGenericClientBuilder_ShouldUpdateDaprApiToken()
|
||||
{
|
||||
// Arrange
|
||||
const string apiToken = "abc123";
|
||||
var builder = new SampleDaprGenericClientBuilder();
|
||||
|
||||
// Act
|
||||
builder.UseDaprApiToken(apiToken);
|
||||
|
||||
// Assert
|
||||
Assert.Equal(apiToken, builder.DaprApiToken);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void DaprGenericClientBuilder_ShouldUpdateTimeout()
|
||||
{
|
||||
// Arrange
|
||||
var timeout = new TimeSpan(4, 2, 1, 2);
|
||||
var builder = new SampleDaprGenericClientBuilder();
|
||||
|
||||
// Act
|
||||
builder.UseTimeout(timeout);
|
||||
|
||||
// Assert
|
||||
Assert.Equal(timeout, builder.Timeout);
|
||||
}
|
||||
|
||||
private class SampleDaprGenericClientBuilder : DaprGenericClientBuilder<SampleDaprGenericClientBuilder>
|
||||
{
|
||||
public override SampleDaprGenericClientBuilder Build()
|
||||
{
|
||||
// Implementation
|
||||
throw new NotImplementedException();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,42 @@
|
|||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<PropertyGroup>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
<Nullable>enable</Nullable>
|
||||
<IsPackable>false</IsPackable>
|
||||
<IsTestProject>true</IsTestProject>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="coverlet.msbuild">
|
||||
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
|
||||
<PrivateAssets>all</PrivateAssets>
|
||||
</PackageReference>
|
||||
<PackageReference Include="coverlet.collector" />
|
||||
<PackageReference Include="Microsoft.NET.Test.Sdk" />
|
||||
<PackageReference Include="Moq" />
|
||||
<PackageReference Include="xunit" />
|
||||
<PackageReference Include="xunit.runner.visualstudio" >
|
||||
<PrivateAssets>all</PrivateAssets>
|
||||
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
|
||||
</PackageReference>
|
||||
<PackageReference Include="Google.Protobuf" />
|
||||
<PackageReference Include="Grpc.Core.Testing" />
|
||||
<PackageReference Include="Grpc.Net.Client" />
|
||||
<PackageReference Include="protobuf-net.Grpc.AspNetCore" />
|
||||
<PackageReference Include="Grpc.Tools" PrivateAssets="All" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<Using Include="Xunit"/>
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<Protobuf Include="protos\test.proto" ProtoRoot="protos" GrpcServices="Client" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\src\Dapr.Messaging\Dapr.Messaging.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
|
|
@ -0,0 +1,55 @@
|
|||
using Dapr.Messaging.PublishSubscribe;
|
||||
|
||||
namespace Dapr.Messaging.Test.PublishSubscribe
|
||||
{
|
||||
public class MessageHandlingPolicyTest
|
||||
{
|
||||
[Fact]
|
||||
public void Test_MessageHandlingPolicy_Constructor()
|
||||
{
|
||||
var timeoutDuration = TimeSpan.FromMilliseconds(2000);
|
||||
const TopicResponseAction defaultResponseAction = TopicResponseAction.Drop;
|
||||
|
||||
var policy = new MessageHandlingPolicy(timeoutDuration, defaultResponseAction);
|
||||
|
||||
Assert.Equal(timeoutDuration, policy.TimeoutDuration);
|
||||
Assert.Equal(defaultResponseAction, policy.DefaultResponseAction);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Test_MessageHandlingPolicy_Equality()
|
||||
{
|
||||
var timeSpan1 = TimeSpan.FromMilliseconds(1000);
|
||||
var timeSpan2 = TimeSpan.FromMilliseconds(2000);
|
||||
|
||||
var policy1 = new MessageHandlingPolicy(timeSpan1, TopicResponseAction.Success);
|
||||
var policy2 = new MessageHandlingPolicy(timeSpan1, TopicResponseAction.Success);
|
||||
var policy3 = new MessageHandlingPolicy(timeSpan2, TopicResponseAction.Retry);
|
||||
|
||||
Assert.Equal(policy1, policy2); // Value Equality
|
||||
Assert.NotEqual(policy1, policy3); // Different values
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Test_MessageHandlingPolicy_Immutability()
|
||||
{
|
||||
var timeoutDuration = TimeSpan.FromMilliseconds(2000);
|
||||
const TopicResponseAction defaultResponseAction = TopicResponseAction.Drop;
|
||||
|
||||
var policy1 = new MessageHandlingPolicy(timeoutDuration, defaultResponseAction);
|
||||
|
||||
var newTimeoutDuration = TimeSpan.FromMilliseconds(3000);
|
||||
const TopicResponseAction newDefaultResponseAction = TopicResponseAction.Retry;
|
||||
|
||||
// Creating a new policy with different values.
|
||||
var policy2 = policy1 with
|
||||
{
|
||||
TimeoutDuration = newTimeoutDuration, DefaultResponseAction = newDefaultResponseAction
|
||||
};
|
||||
|
||||
// Asserting that original policy is unaffected by changes made to new policy.
|
||||
Assert.Equal(timeoutDuration, policy1.TimeoutDuration);
|
||||
Assert.Equal(defaultResponseAction, policy1.DefaultResponseAction);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,32 @@
|
|||
// ------------------------------------------------------------------------
|
||||
// Copyright 2021 The Dapr Authors
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
// ------------------------------------------------------------------------
|
||||
|
||||
syntax = "proto3";
|
||||
|
||||
option csharp_namespace = "Dapr.Client.Autogen.Test.Grpc.v1";
|
||||
|
||||
message TestRun {
|
||||
repeated TestCase tests = 1;
|
||||
}
|
||||
|
||||
message TestCase {
|
||||
string name = 1;
|
||||
}
|
||||
|
||||
message Request {
|
||||
string RequestParameter = 1;
|
||||
}
|
||||
|
||||
message Response {
|
||||
string Name = 1;
|
||||
}
|
||||
Loading…
Reference in New Issue