Attach metadata to each individual message in bulkpublish, if it exists (#1437)

* Instead of just attaching metadata to the full envelope, the metadata is attached to each individual message as well, if it exists at all.

Signed-off-by: Chris Walsh <[email protected]>
Co-authored-by: Whit Waldo <[email protected]>

Author: cwalsh2189

src/Dapr.Client/DaprClientGrpc.cs

@@ -215,6 +215,12 @@ events[counter] is CloudEvent
                         ? Constants.ContentTypeCloudEvent
                         : Constants.ContentTypeApplicationJson,
             };
+
+            if (metadata != null)
+            {
+                entry.Metadata.Add(metadata);
+            }
+
             envelope.Entries.Add(entry);
             entryMap.Add(counter.ToString(), new BulkPublishEntry<TValue>(
                 entry.EntryId, events[counter], entry.ContentType, entry.Metadata));

test/Dapr.Client.Test/BulkPublishEventApiTest.cs

@@ -87,30 +87,40 @@ await daprClient.BulkPublishEventAsync(TestPubsubName, TestTopicName, bulkPublis
         request.Dismiss();
 
         var envelope = await request.GetRequestEnvelopeAsync<Autogenerated.BulkPublishRequest>();
-            
+
         envelope.Entries.Count.ShouldBe(2);
         envelope.PubsubName.ShouldBe(TestPubsubName);
         envelope.Topic.ShouldBe(TestTopicName);
 
+        //Validate that the metadata is on the envelope
         envelope.Metadata.Count.ShouldBe(2);
         envelope.Metadata.Keys.Contains("key1").ShouldBeTrue();
         envelope.Metadata.Keys.Contains("key2").ShouldBeTrue();
         envelope.Metadata["key1"].ShouldBe("value1");
         envelope.Metadata["key2"].ShouldBe("value2");
 
+        //Validate that the metadata is also on each entry
         var firstEntry = envelope.Entries[0];
 
         firstEntry.EntryId.ShouldBe("0");
         firstEntry.ContentType.ShouldBe(TestContentType);
         firstEntry.Event.ToStringUtf8().ShouldBe(JsonSerializer.Serialize(bulkPublishData[0], client.InnerClient.JsonSerializerOptions));
-        firstEntry.Metadata.ShouldBeEmpty();
+        firstEntry.Metadata.Keys.Count.ShouldBe(2);
+        firstEntry.Metadata.ContainsKey("key1").ShouldBeTrue();
+        firstEntry.Metadata["key1"].ShouldBe("value1");
+        firstEntry.Metadata.ContainsKey("key2").ShouldBeTrue();
+        firstEntry.Metadata["key2"].ShouldBe("value2");
 
         var secondEntry = envelope.Entries[1];
 
         secondEntry.EntryId.ShouldBe("1");
         secondEntry.ContentType.ShouldBe(TestContentType);
         secondEntry.Event.ToStringUtf8().ShouldBe(JsonSerializer.Serialize(bulkPublishData[1], client.InnerClient.JsonSerializerOptions));
-        secondEntry.Metadata.ShouldBeEmpty();
+        secondEntry.Metadata.Keys.Count.ShouldBe(2);
+        secondEntry.Metadata.ContainsKey("key1").ShouldBeTrue();
+        secondEntry.Metadata["key1"].ShouldBe("value1");
+        secondEntry.Metadata.ContainsKey("key2").ShouldBeTrue();
+        secondEntry.Metadata["key2"].ShouldBe("value2");
 
         // Create Response & Respond
         var response = new Autogenerated.BulkPublishResponse
@@ -183,25 +193,35 @@ await daprClient.BulkPublishEventAsync(TestPubsubName, TestTopicName, bulkPublis
         envelope.PubsubName.ShouldBe(TestPubsubName);
         envelope.Topic.ShouldBe(TestTopicName);
 
+        //Validate the metadata on the envelope
         envelope.Metadata.Count.ShouldBe(2);
         envelope.Metadata.Keys.Contains("key1").ShouldBeTrue();
         envelope.Metadata.Keys.Contains("key2").ShouldBeTrue();
         envelope.Metadata["key1"].ShouldBe("value1");
         envelope.Metadata["key2"].ShouldBe("value2");
 
+        //Validate the metadata on each entry
         var firstEntry = envelope.Entries[0];
 
         firstEntry.EntryId.ShouldBe("0");
         firstEntry.ContentType.ShouldBe(TestContentType);
         firstEntry.Event.ToStringUtf8().ShouldBe(JsonSerializer.Serialize(bulkPublishData[0], client.InnerClient.JsonSerializerOptions));
-        firstEntry.Metadata.ShouldBeEmpty();
+        firstEntry.Metadata.Keys.Count.ShouldBe(2);
+        firstEntry.Metadata.ContainsKey("key1").ShouldBeTrue();
+        firstEntry.Metadata["key1"].ShouldBe("value1");
+        firstEntry.Metadata.ContainsKey("key2").ShouldBeTrue();
+        firstEntry.Metadata["key2"].ShouldBe("value2");
 
         var secondEntry = envelope.Entries[1];
 
         secondEntry.EntryId.ShouldBe("1");
         secondEntry.ContentType.ShouldBe(TestContentType);
         secondEntry.Event.ToStringUtf8().ShouldBe(JsonSerializer.Serialize(bulkPublishData[1], client.InnerClient.JsonSerializerOptions));
-        secondEntry.Metadata.ShouldBeEmpty();
+        secondEntry.Metadata.Keys.Count.ShouldBe(2);
+        secondEntry.Metadata.ContainsKey("key1").ShouldBeTrue();
+        secondEntry.Metadata["key1"].ShouldBe("value1");
+        secondEntry.Metadata.ContainsKey("key2").ShouldBeTrue();
+        secondEntry.Metadata["key2"].ShouldBe("value2");
 
         // Create Response & Respond
         var response = new Autogenerated.BulkPublishResponse
@@ -336,4 +356,4 @@ private class Widget
         public string Size { get; set; }
         public string Color { get; set; }
     }
-}
+}