Migrate Dapr programmatic pubsub functionality into Dapr.Messaging package

examples/Client/PublishSubscribe/StreamingSubscriptionExample/Program.cs

@@ -1,4 +1,5 @@
 using System.Text;
+using Dapr.Messaging.Clients.StreamingClient;
 using Dapr.Messaging.PublishSubscribe;
 using Dapr.Messaging.PublishSubscribe.Extensions;
 
@@ -21,12 +22,12 @@ Task<TopicResponseAction> HandleMessageAsync(TopicMessage message, CancellationT
     }
 }
 
-var messagingClient = app.Services.GetRequiredService<DaprPublishSubscribeClient>();
+var messagingClient = app.Services.GetRequiredService<DaprPubSubStreamingClient>();
 
 //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)),
+    new DaprStreamingSubscriptionOptions(new MessageHandlingPolicy(TimeSpan.FromSeconds(10), TopicResponseAction.Retry)),
     HandleMessageAsync, cancellationTokenSource.Token);
 
 await Task.Delay(TimeSpan.FromMinutes(1));

src/Dapr.Common/JsonConverters/RFC3389JsonConverter.cs

@@ -0,0 +1,80 @@
+// ------------------------------------------------------------------------
+//  Copyright 2025 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.Globalization;
+using System.Text.Json;
+using System.Text.Json.Serialization;
+
+namespace Dapr.Common.JsonConverters;
+
+/// <summary>
+/// Provides serialization and deserialization between a <see cref="DateTimeOffset"/> and its RFC3389 format.
+/// </summary>
+internal sealed class Rfc3389JsonConverter : JsonConverter<DateTimeOffset?>
+{
+    private const string Rfc3389Format = "yyyy-MM-dd'T'HH:mm:ss.fffK";
+
+    /// <summary>Reads and converts the JSON to type.</summary>
+    /// <param name="reader">The reader.</param>
+    /// <param name="typeToConvert">The type to convert.</param>
+    /// <param name="options">An object that specifies serialization options to use.</param>
+    /// <returns>The converted value.</returns>
+    public override DateTimeOffset? Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options)
+    {
+        var stringValue = reader.GetString();
+        if (stringValue is null)
+        {
+            return null;
+        }
+
+        if (string.IsNullOrWhiteSpace(stringValue))
+        {
+            throw new JsonException("The data string is empty or whitespace and cannot be converted to a DateTimeOffset.");
+        }
+
+        try
+        {
+            return DateTimeOffset.ParseExact(stringValue, Rfc3389Format, CultureInfo.InvariantCulture,
+                DateTimeStyles.AdjustToUniversal);
+        }
+        catch (FormatException ex)
+        {
+            throw new JsonException($"The date string '{stringValue}' is not in the expected RFC3389 format.", ex);
+        }
+    }
+
+    /// <summary>Writes a specified value as JSON.</summary>
+    /// <param name="writer">The writer to write to.</param>
+    /// <param name="value">The value to convert to JSON.</param>
+    /// <param name="options">An object that specifies serialization options to use.</param>
+    public override void Write(Utf8JsonWriter writer, DateTimeOffset? value, JsonSerializerOptions options)
+    {
+        try
+        {
+            if (value is null)
+            {
+                writer.WriteNullValue();
+            }
+            else
+            { 
+                var dateString = ((DateTimeOffset)value).ToString(Rfc3389Format, CultureInfo.InvariantCulture);
+                var targetValue = dateString.Replace("+00:00", "Z").Trim('"');
+                writer.WriteStringValue(targetValue);
+            }
+        }
+        catch (Exception ex)
+        {
+            throw new JsonException("An error occurred while writing the DateTimeOffset value.", ex);
+        }
+    }
+}

src/Dapr.Common/Serialization/TypeConverters.cs

@@ -0,0 +1,48 @@
+// ------------------------------------------------------------------------
+//  Copyright 2025 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.Text.Json;
+using Google.Protobuf;
+
+namespace Dapr.Common.Serialization;
+
+/// <summary>
+/// Converters used to serialize and deserialize data.
+/// </summary>
+internal static class TypeConverters
+{
+    /// <summary>
+    /// Converts an arbitrary type to a <see cref="System.Text.Json"/>-based <see cref="ByteString"/>. 
+    /// </summary>
+    /// <param name="data">The data to convert.</param>
+    /// <param name="options">The JSON serialization options.</param>
+    /// <typeparam name="T">The type of the given data.</typeparam>
+    /// <returns>The given data as a JSON-based byte string.</returns>
+    internal static ByteString ToJsonByteString<T>(T data, JsonSerializerOptions options)
+    {
+        var bytes = JsonSerializer.SerializeToUtf8Bytes(data, options);
+        return ByteString.CopyFrom(bytes);
+    }
+
+    /// <summary>
+    /// Deserializes a <see cref="System.Text.Json"/>-based <see cref="ByteString"/> to an arbitrary type. 
+    /// </summary>
+    /// <param name="data">The data to convert.</param>
+    /// <param name="options">The JSON serialization options.</param>
+    /// <typeparam name="T">The type of the data to deserialize to.</typeparam>
+    /// <returns>The strongly-typed deserialized data.</returns>
+    internal static T? FromJsonByteString<T>(ByteString data, JsonSerializerOptions options) where T : class
+    {
+        return data.Length == 0 ? null : JsonSerializer.Deserialize<T>(data.Span, options);
+    }
+}

src/Dapr.Messaging/PublishSubscribe/IDaprPubSubBuilder.cs → src/Dapr.Messaging/Clients/IDaprPubSubBuilder.cs

@@ -11,7 +11,7 @@
 //  limitations under the License.
 //  ------------------------------------------------------------------------
 
-namespace Dapr.Messaging.PublishSubscribe;
+namespace Dapr.Messaging.Clients.StreamingClient;
 
 /// <summary>
 /// Provides a Dapr client builder specific for Publish/Subscribe operations.

src/Dapr.Messaging/Clients/ProgrammaticClient/CloudEvent.cs

@@ -0,0 +1,70 @@
+// ------------------------------------------------------------------------
+//  Copyright 2025 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.Text.Json.Serialization;
+using Dapr.Common.JsonConverters;
+using Dapr.Messaging.JsonConverters;
+
+namespace Dapr.Messaging.PublishSubscribe;
+
+/// <summary>
+/// Represents a CloudEvent without data.
+/// </summary>
+/// <param name="Source">The context in which an event happened, e.g. the type of an event source.</param>
+/// <param name="Type">Describes the type of event related to the originating occurrence.</param>
+public record CloudEvent(
+    [property: JsonPropertyName("source")] Uri Source, 
+    [property: JsonPropertyName("type")] string Type)
+{
+    /// <summary>
+    /// The subject of the event in the context of the event producer (identified by <see cref="Source"/>).
+    /// </summary>
+    [JsonPropertyName("subject")]
+    public string? Subject { get; init; }
+
+    /// <summary>
+    /// The version of the CloudEvents specification which the event uses.
+    /// </summary>
+    /// <remarks>
+    /// While this SDK implements specification 1.0.2, this value only has the major and minor values included allowing
+    /// for "patch" changes that don't change this property's value in the serialization.
+    /// </remarks>
+    [JsonPropertyName("specversion")]
+    public string SpecVersion => "1.0";
+
+    /// <summary>
+    /// The timestamp of when the occurrence happened.
+    /// </summary>
+    [JsonPropertyName("time")]
+    [JsonConverter(typeof(Rfc3389JsonConverter))]
+    public DateTimeOffset? Time { get; init; } = null;
+}
+
+/// <summary>
+/// Represents a CloudEvent with typed data.
+/// </summary>
+/// <param name="Source">The context in which an event happened, e.g. the type of an event source.</param>
+/// <param name="Type">Describes the type of event related to the originating occurrence.</param>
+/// <param name="Data">Domain-specific information about the event occurrence.</param>
+[JsonConverter(typeof(CloudEventDataJsonSerializer<>))]
+public record CloudEvent<TData>(
+    Uri Source, 
+    string Type, 
+    [property: JsonPropertyName("data")] TData Data) : CloudEvent(Source, Type)
+{
+    /// <summary>
+    /// Content type of the data value.
+    /// </summary>
+    [JsonPropertyName("datacontenttype")]
+    public string DataContentType { get; init; } = "application/json";
+}

src/Dapr.Messaging/Clients/ProgrammaticClient/DaprBulkPublishRequest.cs

@@ -0,0 +1,22 @@
+// ------------------------------------------------------------------------
+//  Copyright 2025 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.Clients.ProgrammaticClient;
+
+/// <summary>
+/// Information about the type being published via the bulk publish operation. 
+/// </summary>
+/// <param name="Payload">The data to serialize in the event.</param>
+/// <param name="DataContentType">The optional data content type. This defaults to "application/json" is not set.</param>
+/// <typeparam name="TValue">The type to serialize.</typeparam>
+public sealed record DaprBulkPublishRequest<TValue>(TValue Payload, string DataContentType = "application/json");

src/Dapr.Messaging/Clients/ProgrammaticClient/DaprBulkPublishResponse.cs

@@ -0,0 +1,22 @@
+// ------------------------------------------------------------------------
+//  Copyright 2025 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.Messaging.PublishSubscribe;
+
+namespace Dapr.Messaging.Clients.ProgrammaticClient;
+
+/// <summary>
+/// Represents the responses returned for failed bulk publishing events.
+/// </summary>
+/// <param name="FailedEntries">The list of entries that failed to be published.</param>
+public record DaprBulkPublishResponse(IReadOnlyList<DaprBulkPublishResponseFailedEntry> FailedEntries);

src/Dapr.Messaging/Clients/ProgrammaticClient/DaprBulkPublishResponseFailedEntry.cs

@@ -0,0 +1,23 @@
+// ------------------------------------------------------------------------
+//  Copyright 2025 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.Client.Autogen.Grpc.v1;
+
+namespace Dapr.Messaging.PublishSubscribe;
+
+/// <summary>
+/// Represents the status of each event that was published during BulkPublishRequest.
+/// </summary>
+/// <param name="Entry">The entry that failed to be published.</param>
+/// <param name="ErrorMessage">The error message stating why the entry failed to publish.</param>
+public record DaprBulkPublishResponseFailedEntry(BulkPublishRequestEntry Entry, string ErrorMessage);