Add .NET client for dynamic pubsub subscriptions

README.md

@@ -44,7 +44,10 @@ This repo builds the following packages:
 - Dapr.AspNetCore
 - Dapr.Actors
 - Dapr.Actors.AspNetCore
+- Dapr.Common
 - Dapr.Extensions.Configuration
+- Dapr.Messaging
+- Dapr.Protos
 - Dapr.Workflow
 
 ### Prerequisites

all.sln

@@ -118,6 +118,16 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Dapr.E2E.Test.Actors.Genera
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Cryptography", "examples\Client\Cryptography\Cryptography.csproj", "{C74FBA78-13E8-407F-A173-4555AEE41FF3}"
 EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Dapr.Messaging", "src\Dapr.Messaging\Dapr.Messaging.csproj", "{250F0236-2014-4DD8-A688-CD25EE299FA3}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Dapr.Protos", "src\Dapr.Protos\Dapr.Protos.csproj", "{CE506C30-5701-47C9-A86E-39D796B8DF35}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Dapr.Common", "src\Dapr.Common\Dapr.Common.csproj", "{9AB7EB9D-82CB-4BC6-B895-4F52F8DC489F}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Dapr.Messaging.Test", "test\Dapr.Messaging.Test\Dapr.Messaging.Test.csproj", "{93C6ABAF-F4B7-4CA2-8734-565EF847668A}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "StreamingSubscriptionExample", "examples\Client\PublishSubscribe\StreamingSubscriptionExample\StreamingSubscriptionExample.csproj", "{E748C988-1F5F-4A34-9A5C-2EE2B6CD1BA2}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -290,6 +300,26 @@ Global
 		{C74FBA78-13E8-407F-A173-4555AEE41FF3}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{C74FBA78-13E8-407F-A173-4555AEE41FF3}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{C74FBA78-13E8-407F-A173-4555AEE41FF3}.Release|Any CPU.Build.0 = Release|Any CPU
+		{250F0236-2014-4DD8-A688-CD25EE299FA3}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{250F0236-2014-4DD8-A688-CD25EE299FA3}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{250F0236-2014-4DD8-A688-CD25EE299FA3}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{250F0236-2014-4DD8-A688-CD25EE299FA3}.Release|Any CPU.Build.0 = Release|Any CPU
+		{CE506C30-5701-47C9-A86E-39D796B8DF35}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{CE506C30-5701-47C9-A86E-39D796B8DF35}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{CE506C30-5701-47C9-A86E-39D796B8DF35}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{CE506C30-5701-47C9-A86E-39D796B8DF35}.Release|Any CPU.Build.0 = Release|Any CPU
+		{9AB7EB9D-82CB-4BC6-B895-4F52F8DC489F}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{9AB7EB9D-82CB-4BC6-B895-4F52F8DC489F}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{9AB7EB9D-82CB-4BC6-B895-4F52F8DC489F}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{9AB7EB9D-82CB-4BC6-B895-4F52F8DC489F}.Release|Any CPU.Build.0 = Release|Any CPU
+		{93C6ABAF-F4B7-4CA2-8734-565EF847668A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{93C6ABAF-F4B7-4CA2-8734-565EF847668A}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{93C6ABAF-F4B7-4CA2-8734-565EF847668A}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{93C6ABAF-F4B7-4CA2-8734-565EF847668A}.Release|Any CPU.Build.0 = Release|Any CPU
+		{E748C988-1F5F-4A34-9A5C-2EE2B6CD1BA2}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{E748C988-1F5F-4A34-9A5C-2EE2B6CD1BA2}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{E748C988-1F5F-4A34-9A5C-2EE2B6CD1BA2}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{E748C988-1F5F-4A34-9A5C-2EE2B6CD1BA2}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -343,6 +373,11 @@ Global
 		{AF89083D-4715-42E6-93E9-38497D12A8A6} = {DD020B34-460F-455F-8D17-CF4A949F100B}
 		{B5CDB0DC-B26D-48F1-B934-FE5C1C991940} = {DD020B34-460F-455F-8D17-CF4A949F100B}
 		{C74FBA78-13E8-407F-A173-4555AEE41FF3} = {A7F41094-8648-446B-AECD-DCC2CC871F73}
+		{250F0236-2014-4DD8-A688-CD25EE299FA3} = {27C5D71D-0721-4221-9286-B94AB07B58CF}
+		{CE506C30-5701-47C9-A86E-39D796B8DF35} = {27C5D71D-0721-4221-9286-B94AB07B58CF}
+		{9AB7EB9D-82CB-4BC6-B895-4F52F8DC489F} = {27C5D71D-0721-4221-9286-B94AB07B58CF}
+		{93C6ABAF-F4B7-4CA2-8734-565EF847668A} = {DD020B34-460F-455F-8D17-CF4A949F100B}
+		{E748C988-1F5F-4A34-9A5C-2EE2B6CD1BA2} = {0EF6EA64-D7C3-420D-9890-EAE8D54A57E6}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {65220BF2-EAE1-4CB2-AA58-EBE80768CB40}

examples/Client/PublishSubscribe/StreamingSubscriptionExample/Program.cs

@@ -0,0 +1,34 @@
+using System.Text;
+using Dapr.Messaging.PublishSubscribe;
+
+var daprMessagingClientBuilder = new DaprPublishSubscribeClientBuilder();
+var daprMessagingClient = daprMessagingClientBuilder.Build();
+
+//Processor for each of the messages returned from the subscription
+async Task<TopicResponseAction> HandleMessage(TopicMessage message, CancellationToken cancellationToken = default)
+{
+    try
+    {
+        //Do something with the message
+        Console.WriteLine(Encoding.UTF8.GetString(message.Data.Span));
+        return await Task.FromResult(TopicResponseAction.Success);
+    }
+    catch
+    {
+        return await Task.FromResult(TopicResponseAction.Retry);
+    }
+}
+
+//Create a dynamic streaming subscription
+var subscription = daprMessagingClient.Register("pubsub", "myTopic",
+    new DaprSubscriptionOptions(new MessageHandlingPolicy(TimeSpan.FromSeconds(15), TopicResponseAction.Retry)),
+    HandleMessage, CancellationToken.None);
+
+//Subscribe to messages on it with a timeout of 30 seconds
+var cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+await subscription.SubscribeAsync(cancellationTokenSource.Token);
+
+await Task.Delay(TimeSpan.FromMinutes(1));
+
+//When you're done with the subscription, simply dispose of it
+await subscription.DisposeAsync();

examples/Client/PublishSubscribe/StreamingSubscriptionExample/StreamingSubscriptionExample.csproj

@@ -0,0 +1,14 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net6</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\..\..\src\Dapr.Messaging\Dapr.Messaging.csproj" />
+  </ItemGroup>
+
+</Project>

properties/IsExternalInit.cs

@@ -13,5 +13,5 @@ namespace System.Runtime.CompilerServices
     internal static class IsExternalInit
     {
     }
-    
+
 }

src/Dapr.Common/Dapr.Common.csproj

@@ -0,0 +1,18 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <Nullable>enable</Nullable>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <Compile Include="..\Shared\DaprDefaults.cs" />
+        <Compile Include="..\Shared\ArgumentVerifier.cs" />
+    </ItemGroup>
+
+    <ItemGroup>
+        <PackageReference Include="Grpc.Net.Client" Version="2.52.0" />
+        <PackageReference Include="Microsoft.Extensions.Http" Version="3.1.32"  />
+    </ItemGroup>
+
+</Project>

src/Dapr.Common/DaprException.cs

@@ -0,0 +1,55 @@
+// ------------------------------------------------------------------------
+// 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.Runtime.Serialization;
+
+namespace Dapr
+{
+    /// <summary>
+    /// The base type of exceptions thrown by the Dapr .NET SDK.
+    /// </summary>
+    [Serializable]
+    public class DaprException : Exception
+    {
+        /// <summary>
+        /// Initializes a new instance of <see cref="DaprException" /> with the provided <paramref name="message" />.
+        /// </summary>
+        /// <param name="message">The exception message.</param>
+        public DaprException(string message)
+            : base(message)
+        {
+        }
+        /// <summary>
+        /// Initializes a new instance of <see cref="DaprException" /> with the provided 
+        /// <paramref name="message" /> and <paramref name="innerException" />.
+        /// </summary>
+        /// <param name="message">The exception message.</param>
+        /// <param name="innerException">The inner exception.</param>
+        public DaprException(string message, Exception innerException)
+            : base(message, innerException)
+        {
+        }
+        /// <summary>
+        /// Initializes a new instance of the <see cref="DaprException"/> class with a specified context.
+        /// </summary>
+        /// <param name="info">The <see cref="SerializationInfo" /> object that contains serialized object data of the exception being thrown.</param>
+        /// <param name="context">The <see cref="StreamingContext" /> object that contains contextual information about the source or destination. The context parameter is reserved for future use and can be null.</param>
+#if NET8_0_OR_GREATER
+        [Obsolete(DiagnosticId = "SYSLIB0051")] // add this attribute to GetObjectData
+#endif
+        protected DaprException(SerializationInfo info, StreamingContext context)
+            : base(info, context)
+        {
+        }
+    }
+}