add metadata option to publish api (#542)

src/Dapr.Client/DaprClient.cs

@@ -24,7 +24,28 @@ namespace Dapr.Client
         /// <param name="cancellationToken">A <see cref="CancellationToken" /> that can be used to cancel the operation.</param>
         /// <typeparam name="TData">The data type of the object that will be serialized.</typeparam>
         /// <returns>A <see cref="Task" /> that will complete when the operation has completed.</returns>
-        public abstract Task PublishEventAsync<TData>(string pubsubName, string topicName, TData data, CancellationToken cancellationToken = default);
+        public abstract Task PublishEventAsync<TData>(
+            string pubsubName,
+            string topicName,
+            TData data,
+            CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Publishes an event to the specified topic.
+        /// </summary>
+        /// <param name="pubsubName">The name of the pubsub component to use.</param>
+        /// <param name="topicName">The name of the topic the request should be published to.</param>
+        /// <param name="data">The  event data.</param>
+        /// <param name="metadata">A collection of metadata key-value pairs that will be provided to the binding. The valid metadata keys and values are determined by the type of binding used.</param>
+        /// <param name="cancellationToken">A <see cref="CancellationToken" /> that can be used to cancel the operation.</param>
+        /// <typeparam name="TData">The data type of the object that will be serialized.</typeparam>
+        /// <returns>A <see cref="Task" /> that will complete when the operation has completed.</returns>
+        public abstract Task PublishEventAsync<TData>(
+            string pubsubName,
+            string topicName,
+            TData data,
+            Dictionary<string, string> metadata,
+            CancellationToken cancellationToken = default);
 
         /// <summary>
         /// Publishes an event to the specified topic.
@@ -33,7 +54,24 @@ namespace Dapr.Client
         /// <param name="topicName">The name of the topic the request should be published to.</param>
         /// <param name="cancellationToken">A <see cref="CancellationToken" /> that can be used to cancel the operation.</param>
         /// <returns>A <see cref="Task" /> that will complete when the operation has completed.</returns>
-        public abstract Task PublishEventAsync(string pubsubName, string topicName, CancellationToken cancellationToken = default);
+        public abstract Task PublishEventAsync(
+            string pubsubName,
+            string topicName,
+            CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Publishes an event to the specified topic.
+        /// </summary>
+        /// <param name="pubsubName">The name of the pubsub component to use.</param>
+        /// <param name="topicName">The name of the topic the request should be published to.</param>
+        /// <param name="metadata">A collection of metadata key-value pairs that will be provided to the binding. The valid metadata keys and values are determined by the type of binding used.</param>
+        /// <param name="cancellationToken">A <see cref="CancellationToken" /> that can be used to cancel the operation.</param>
+        /// <returns>A <see cref="Task" /> that will complete when the operation has completed.</returns>
+        public abstract Task PublishEventAsync(
+            string pubsubName,
+            string topicName,
+            Dictionary<string, string> metadata,
+            CancellationToken cancellationToken = default);
 
         /// <summary>
         /// Invokes an output binding.

src/Dapr.Client/DaprClientGrpc.cs

@@ -60,23 +60,62 @@ namespace Dapr.Client
 
         #region Publish Apis
         /// <inheritdoc/>
-        public override Task PublishEventAsync<TData>(string pubsubName, string topicName, TData data, CancellationToken cancellationToken = default)
+
+        public override Task PublishEventAsync<TData>(
+            string pubsubName,
+            string topicName,
+            TData data,
+            CancellationToken cancellationToken = default)
         {
             ArgumentVerifier.ThrowIfNullOrEmpty(pubsubName, nameof(pubsubName));
             ArgumentVerifier.ThrowIfNullOrEmpty(topicName, nameof(topicName));
             ArgumentVerifier.ThrowIfNull(data, nameof(data));
-            return MakePublishRequest(pubsubName, topicName, data, cancellationToken);
+            return MakePublishRequest(pubsubName, topicName, data, null, cancellationToken);
         }
 
-        /// <inheritdoc/>
+        public override Task PublishEventAsync<TData>(
-        public override Task PublishEventAsync(string pubsubName, string topicName, CancellationToken cancellationToken = default)
+            string pubsubName,
+            string topicName,
+            TData data,
+            Dictionary<string, string> metadata,
+            CancellationToken cancellationToken = default)
         {
             ArgumentVerifier.ThrowIfNullOrEmpty(pubsubName, nameof(pubsubName));
             ArgumentVerifier.ThrowIfNullOrEmpty(topicName, nameof(topicName));
-            return MakePublishRequest(pubsubName, topicName, string.Empty, cancellationToken);
+            ArgumentVerifier.ThrowIfNull(data, nameof(data));
+            ArgumentVerifier.ThrowIfNull(metadata, nameof(metadata));
+            return MakePublishRequest(pubsubName, topicName, data, metadata, cancellationToken);
         }
 
-        private async Task MakePublishRequest<TContent>(string pubsubName, string topicName, TContent content, CancellationToken cancellationToken)
+        /// <inheritdoc/>
+        public override Task PublishEventAsync(
+            string pubsubName,
+            string topicName,
+            CancellationToken cancellationToken = default)
+        {
+            ArgumentVerifier.ThrowIfNullOrEmpty(pubsubName, nameof(pubsubName));
+            ArgumentVerifier.ThrowIfNullOrEmpty(topicName, nameof(topicName));
+            return MakePublishRequest(pubsubName, topicName, string.Empty, null, cancellationToken);
+        }
+
+        public override Task PublishEventAsync(
+            string pubsubName,
+            string topicName,
+            Dictionary<string, string> metadata,
+            CancellationToken cancellationToken = default)
+        {
+            ArgumentVerifier.ThrowIfNullOrEmpty(pubsubName, nameof(pubsubName));
+            ArgumentVerifier.ThrowIfNullOrEmpty(topicName, nameof(topicName));
+            ArgumentVerifier.ThrowIfNull(metadata, nameof(metadata));
+            return MakePublishRequest(pubsubName, topicName, string.Empty, metadata, cancellationToken);
+        }
+
+        private async Task MakePublishRequest<TContent>(
+            string pubsubName,
+            string topicName,
+            TContent content,
+            Dictionary<string, string> metadata,
+            CancellationToken cancellationToken)
         {
             // Create PublishEventEnvelope
             var envelope = new Autogenerated.PublishEventRequest()
@@ -90,6 +129,11 @@ namespace Dapr.Client
                 envelope.Data = TypeConverters.ToJsonByteString(content, this.jsonSerializerOptions);
             }
 
+            if (metadata != null)
+            {
+                envelope.Metadata.Add(metadata);
+            }
+
             await this.MakeGrpcCallHandleError(
                 options => client.PublishEventAsync(envelope, options),
                 cancellationToken);

test/Dapr.Client.Test/InvokeBindingApiTest.cs

@@ -19,6 +19,28 @@ namespace Dapr.Client.Test
     {
         [Fact]
         public async Task InvokeBindingAsync_ValidateRequest()
+        {
+            // Configure Client
+            var httpClient = new TestHttpClient();
+            var daprClient = new DaprClientBuilder()
+                .UseGrpcChannelOptions(new GrpcChannelOptions { HttpClient = httpClient })
+                .Build();
+
+            var invokeRequest = new InvokeRequest() { RequestParameter = "Hello " };
+            var task = daprClient.InvokeBindingAsync<InvokeRequest>("test", "create", invokeRequest);
+
+            // Get Request and validate                     
+            httpClient.Requests.TryDequeue(out var entry).Should().BeTrue();
+            var request = await GrpcUtils.GetRequestFromRequestMessageAsync<InvokeBindingRequest>(entry.Request);
+            request.Name.Should().Be("test");
+            request.Metadata.Count.Should().Be(0);
+            var json = request.Data.ToStringUtf8();
+            var typeFromRequest = JsonSerializer.Deserialize<InvokeRequest>(json);
+            typeFromRequest.RequestParameter.Should().Be("Hello ");
+        }
+
+        [Fact]
+        public async Task InvokeBindingAsync_ValidateRequest_WithMetadata()
         {
             // Configure Client
             var httpClient = new TestHttpClient();

test/Dapr.Client.Test/PublishEventApiTest.cs

@@ -6,6 +6,7 @@
 namespace Dapr.Client.Test
 {
     using System;
+    using System.Collections.Generic;
     using System.Text.Json;
     using System.Threading;
     using System.Threading.Tasks;
@@ -36,6 +37,39 @@ namespace Dapr.Client.Test
             request.PubsubName.Should().Be(TestPubsubName);
             request.Topic.Should().Be("test");
             jsonFromRequest.Should().Be(JsonSerializer.Serialize(publishData));
+            request.Metadata.Count.Should().Be(0);
+        }
+
+        [Fact]
+        public async Task PublishEventAsync_CanPublishTopicWithData_WithMetadata()
+        {
+            var httpClient = new TestHttpClient();
+            var daprClient = new DaprClientBuilder()
+                .UseGrpcChannelOptions(new GrpcChannelOptions{ HttpClient = httpClient })
+                .Build();
+
+            var metadata = new Dictionary<string, string>
+            {
+                { "key1", "value1" },
+                { "key2", "value2" }
+            };
+
+            var publishData = new PublishData() { PublishObjectParameter = "testparam" };
+            var task = daprClient.PublishEventAsync<PublishData>(TestPubsubName, "test", publishData, metadata);
+
+            httpClient.Requests.TryDequeue(out var entry).Should().BeTrue();
+            var request = await GrpcUtils.GetRequestFromRequestMessageAsync<PublishEventRequest>(entry.Request);
+            var jsonFromRequest = request.Data.ToStringUtf8();
+
+            request.PubsubName.Should().Be(TestPubsubName);
+            request.Topic.Should().Be("test");
+            jsonFromRequest.Should().Be(JsonSerializer.Serialize(publishData));
+
+            request.Metadata.Count.Should().Be(2);
+            request.Metadata.Keys.Contains("key1").Should().BeTrue();
+            request.Metadata.Keys.Contains("key2").Should().BeTrue();
+            request.Metadata["key1"].Should().Be("value1");
+            request.Metadata["key2"].Should().Be("value2");
         }
 
         [Fact]
@@ -46,7 +80,6 @@ namespace Dapr.Client.Test
                 .UseGrpcChannelOptions(new GrpcChannelOptions { HttpClient = httpClient })
                 .Build();
 
-
             var task = daprClient.PublishEventAsync(TestPubsubName, "test");
             httpClient.Requests.TryDequeue(out var entry).Should().BeTrue();
             var request = await GrpcUtils.GetRequestFromRequestMessageAsync<PublishEventRequest>(entry.Request);
@@ -55,6 +88,38 @@ namespace Dapr.Client.Test
             request.PubsubName.Should().Be(TestPubsubName);
             request.Topic.Should().Be("test");
             jsonFromRequest.Should().Be("\"\"");
+
+            request.Metadata.Count.Should().Be(0);
+        }
+
+        [Fact]
+        public async Task PublishEventAsync_CanPublishTopicWithNoContent_WithMetadata()
+        {
+            var httpClient = new TestHttpClient();
+            var daprClient = new DaprClientBuilder()
+                .UseGrpcChannelOptions(new GrpcChannelOptions { HttpClient = httpClient })
+                .Build();
+
+            var metadata = new Dictionary<string, string>
+            {
+                { "key1", "value1" },
+                { "key2", "value2" }
+            };
+
+            var task = daprClient.PublishEventAsync(TestPubsubName, "test", metadata);
+            httpClient.Requests.TryDequeue(out var entry).Should().BeTrue();
+            var request = await GrpcUtils.GetRequestFromRequestMessageAsync<PublishEventRequest>(entry.Request);
+            var jsonFromRequest = request.Data.ToStringUtf8();
+
+            request.PubsubName.Should().Be(TestPubsubName);
+            request.Topic.Should().Be("test");
+            jsonFromRequest.Should().Be("\"\"");
+
+            request.Metadata.Count.Should().Be(2);
+            request.Metadata.Keys.Contains("key1").Should().BeTrue();
+            request.Metadata.Keys.Contains("key2").Should().BeTrue();
+            request.Metadata["key1"].Should().Be("value1");
+            request.Metadata["key2"].Should().Be("value2");
         }
 
         [Fact]