CloudNative.CloudEvents bridge library

eng/Packages.Data.props

@@ -32,6 +32,7 @@
     <PackageReference Update="Azure.Storage.Blobs.ChangeFeed" Version="12.0.0-preview.1" />
     <PackageReference Update="BenchmarkDotNet" Version="0.11.5" />
     <PackageReference Update="Castle.Core" Version="4.4.0" />
+    <PackageReference Update="CloudNative.CloudEvents" Version="2.0.0-beta.1" />
     <PackageReference Update="coverlet.collector" Version="1.3.0" />
     <PackageReference Update="FluentAssertions" Version="5.10.3" />
     <PackageReference Update="FsCheck.Xunit" Version="2.14.0" />

sdk/eventgrid/Azure.Messaging.EventGrid/api/Azure.Messaging.EventGrid.netstandard2.0.cs

@@ -49,6 +49,10 @@ public EventGridPublisherClient(System.Uri endpoint, Azure.AzureKeyCredential cr
         public EventGridPublisherClient(System.Uri endpoint, Azure.Messaging.EventGrid.EventGridSharedAccessSignatureCredential credential) { }
         public EventGridPublisherClient(System.Uri endpoint, Azure.Messaging.EventGrid.EventGridSharedAccessSignatureCredential credential, Azure.Messaging.EventGrid.EventGridPublisherClientOptions options) { }
         public static string BuildSharedAccessSignature(System.Uri endpoint, System.DateTimeOffset expirationUtc, Azure.AzureKeyCredential key, Azure.Messaging.EventGrid.EventGridPublisherClientOptions.ServiceVersion apiVersion = Azure.Messaging.EventGrid.EventGridPublisherClientOptions.ServiceVersion.V2018_01_01) { throw null; }
+        [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
+        public virtual Azure.Response SendEncodedCloudEvents(System.ReadOnlyMemory<byte> cloudEvents, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
+        [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
+        public virtual System.Threading.Tasks.Task<Azure.Response> SendEncodedCloudEventsAsync(System.ReadOnlyMemory<byte> cloudEvents, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         public virtual Azure.Response SendEvents(System.Collections.Generic.IEnumerable<Azure.Messaging.EventGrid.CloudEvent> events, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         public virtual Azure.Response SendEvents(System.Collections.Generic.IEnumerable<Azure.Messaging.EventGrid.EventGridEvent> events, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         public virtual Azure.Response SendEvents(System.Collections.Generic.IEnumerable<object> events, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }

sdk/eventgrid/Microsoft.Azure.CloudNative.CloudEvents.EventGrid/CHANGELOG.md

@@ -0,0 +1,7 @@
+# Release History
+
+## 1.0.0-preview.1 (Unreleased)
+
+### Added
+
+- Bridge library to allow sending CloudNative CloudEvents using Azure Event Grid.

sdk/eventgrid/Microsoft.Azure.CloudNative.CloudEvents.EventGrid/Directory.Build.props

@@ -0,0 +1,7 @@
+<Project ToolsVersion="15.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
+  <PropertyGroup>
+    <IsClientLibrary>true</IsClientLibrary>
+  </PropertyGroup>
+
+  <Import Project="$([MSBuild]::GetDirectoryNameOfFileAbove($(MSBuildThisFileDirectory).., Directory.Build.props))\Directory.Build.props" />
+</Project>

sdk/eventgrid/Microsoft.Azure.CloudNative.CloudEvents.EventGrid/README.md

@@ -0,0 +1,49 @@
+# CloudNative CloudEvent support for Azure.Messaging.EventGrid library for .NET
+
+TODO
+
+## Getting started
+
+TODO
+
+### Install the package
+
+TODO
+
+### Prerequisites
+
+TODO
+
+### Authenticate the client
+
+TODO
+
+## Key concepts
+
+TODO
+
+## Examples
+
+TODO
+
+## Troubleshooting
+
+TODO
+
+## Next steps
+
+TODO
+
+## Contributing
+
+This project welcomes contributions and suggestions. Most contributions require you to agree to a Contributor License Agreement (CLA) declaring that you have the right to, and actually do, grant us the rights to use your contribution. For details, visit <https://cla.microsoft.com>.
+
+When you submit a pull request, a CLA-bot will automatically determine whether you need to provide a CLA and decorate the PR appropriately (e.g., label, comment). Simply follow the instructions provided by the bot. You will only need to do this once across all repos using our CLA.
+
+This project has adopted the [Microsoft Open Source Code of Conduct][code_of_conduct]. For more information see the [Code of Conduct FAQ][code_of_conduct_faq] or contact opencode@microsoft.com with any additional questions or comments.
+
+![Impressions](https://azure-sdk-impressions.azurewebsites.net/api/impressions/azure-sdk-for-net%2Fsdk%2Fcore%2FMicrosoft.Azure.Core.NewtonsoftJson%2FREADME.png)
+
+[code_of_conduct]: https://opensource.microsoft.com/codeofconduct
+[code_of_conduct_faq]: https://opensource.microsoft.com/codeofconduct/faq/
+[microsoft_spatial_package]: https://www.nuget.org/packages/Microsoft.Spatial/

sdk/eventgrid/Microsoft.Azure.CloudNative.CloudEvents.EventGrid/api/Microsoft.Azure.CloudNative.CloudEvents.EventGrid.netstandard2.0.cs

@@ -0,0 +1,8 @@
+namespace Microsoft.Azure.CloudNative.CloudEvents.EventGrid
+{
+    public static partial class EventGridPublisherClientExtensions
+    {
+        public static Azure.Response SendCloudEvents(this Azure.Messaging.EventGrid.EventGridPublisherClient client, System.Collections.Generic.IEnumerable<CloudNative.CloudEvents.CloudEvent> cloudEvents, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
+        public static System.Threading.Tasks.Task<Azure.Response> SendCloudEventsAsync(this Azure.Messaging.EventGrid.EventGridPublisherClient client, System.Collections.Generic.IEnumerable<CloudNative.CloudEvents.CloudEvent> cloudEvents, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
+    }
+}

sdk/eventgrid/Microsoft.Azure.CloudNative.CloudEvents.EventGrid/src/EventGridPublisherClientExtensions.cs

@@ -0,0 +1,101 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using Azure;
+using Azure.Core;
+using Azure.Core.Pipeline;
+using Azure.Messaging.EventGrid;
+using CloudNative.CloudEvents;
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.IO;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using CloudEvent = CloudNative.CloudEvents.CloudEvent;
+
+#pragma warning disable AZC0001 // Use one of the following pre-approved namespace groups (https://azure.github.io/azure-sdk/registered_namespaces.html): Azure.AI, Azure.Analytics, Azure.Communication, Azure.Data, Azure.DigitalTwins, Azure.Iot, Azure.Learn, Azure.Media, Azure.Management, Azure.Messaging, Azure.Search, Azure.Security, Azure.Storage, Azure.Template, Azure.Identity, Microsoft.Extensions.Azure
+namespace Microsoft.Azure.CloudNative.CloudEvents.EventGrid
+#pragma warning restore AZC0001 // Use one of the following pre-approved namespace groups (https://azure.github.io/azure-sdk/registered_namespaces.html): Azure.AI, Azure.Analytics, Azure.Communication, Azure.Data, Azure.DigitalTwins, Azure.Iot, Azure.Learn, Azure.Media, Azure.Management, Azure.Messaging, Azure.Search, Azure.Security, Azure.Storage, Azure.Template, Azure.Identity, Microsoft.Extensions.Azure
+{
+    /// <summary>
+    /// This class contains extension methods to enable usage of the CloudNative.CloudEvent
+    /// library with the Azure Event Grid library.
+    /// </summary>
+    public static class EventGridPublisherClientExtensions
+    {
+        private static readonly JsonEventFormatter s_eventFormatter = new JsonEventFormatter();
+        private const string TraceParentHeaderName = "traceparent";
+        private const string TraceStateHeaderName = "tracestate";
+
+        /// <summary> Publishes a set of CloudEvents to an Event Grid topic. </summary>
+        /// <param name="client">The <see cref="EventGridPublisherClient"/> instance to extend.</param>
+        /// <param name="cloudEvents"> The set of events to be published to Event Grid. </param>
+        /// <param name="cancellationToken"> An optional cancellation token instance to signal the request to cancel the operation.</param>
+        public static Response SendCloudEvents(
+            this EventGridPublisherClient client,
+            IEnumerable<CloudEvent> cloudEvents,
+            CancellationToken cancellationToken = default) =>
+            SendCloudEventsInternalAsync(client, cloudEvents, false, cancellationToken).EnsureCompleted();
+
+        /// <summary> Publishes a set of CloudEvents to an Event Grid topic. </summary>
+        /// <param name="client">The <see cref="EventGridPublisherClient"/> instance to extend.</param>
+        /// <param name="cloudEvents"> The set of events to be published to Event Grid. </param>
+        /// <param name="cancellationToken"> An optional cancellation token instance to signal the request to cancel the operation.</param>
+        public static async Task<Response> SendCloudEventsAsync(
+            this EventGridPublisherClient client,
+            IEnumerable<CloudEvent> cloudEvents,
+            CancellationToken cancellationToken = default) =>
+            await SendCloudEventsInternalAsync(client, cloudEvents, true, cancellationToken).ConfigureAwait(false);
+
+        private static async Task<Response> SendCloudEventsInternalAsync(
+            EventGridPublisherClient client,
+            IEnumerable<CloudEvent> cloudEvents,
+            bool async,
+            CancellationToken cancellationToken = default)
+        {
+            Argument.AssertNotNull(client, nameof(client));
+            string activityId = null;
+            string traceState = null;
+            Activity currentActivity = Activity.Current;
+            if (currentActivity != null && currentActivity.IsW3CFormat())
+            {
+                activityId = currentActivity.Id;
+                currentActivity.TryGetTraceState(out traceState);
+            }
+
+            using var stream = new MemoryStream();
+            var writer = new Utf8JsonWriter(stream);
+            writer.WriteStartArray();
+            foreach (var cloudEvent in cloudEvents)
+            {
+                var attributes = cloudEvent.GetAttributes();
+                if (activityId != null &&
+                        !attributes.ContainsKey(TraceParentHeaderName) &&
+                        !attributes.ContainsKey(TraceStateHeaderName))
+                {
+                    attributes.Add(TraceParentHeaderName, activityId);
+                    if (traceState != null)
+                    {
+                        attributes.Add(TraceStateHeaderName, traceState);
+                    }
+                }
+
+                byte[] bytes = s_eventFormatter.EncodeStructuredEvent(cloudEvent, out var _);
+                using JsonDocument document = JsonDocument.Parse(bytes);
+                document.RootElement.WriteTo(writer);
+            }
+            writer.WriteEndArray();
+            writer.Flush();
+            if (async)
+            {
+                return await client.SendEncodedCloudEventsAsync(stream.GetBuffer().AsMemory(0, (int)stream.Position), cancellationToken).ConfigureAwait(false);
+            }
+            else
+            {
+                return client.SendEncodedCloudEvents(stream.GetBuffer().AsMemory(0, (int)stream.Position), cancellationToken);
+            }
+        }
+    }
+}

sdk/eventgrid/Microsoft.Azure.CloudNative.CloudEvents.EventGrid/src/Microsoft.Azure.CloudNative.CloudEvents.EventGrid.csproj

@@ -0,0 +1,21 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <AssemblyTitle>CloudNative CloudEvents support for Azure.Messaging.EventGrid library</AssemblyTitle>
+    <Description>This library allows the CloudEvent model from CloudNative.CloudEvents to be published using the Azure Event Grid client library.</Description>
+    <Version>1.0.0-preview.1</Version>
+    <PackageTags>Microsoft Azure SDK CloudNative CloudEvents</PackageTags>
+    <TargetFrameworks>$(RequiredTargetFrameworks)</TargetFrameworks>
+  </PropertyGroup>
+
+  <!-- Pull in Shared Source from Azure.Core -->
+  <ItemGroup>
+    <Compile Include="$(AzureCoreSharedSources)Argument.cs" Link="Shared\%(RecursiveDir)\%(Filename)%(Extension)" />
+    <Compile Include="$(AzureCoreSharedSources)DiagnosticScope.cs" Link="Shared\%(RecursiveDir)\%(Filename)%(Extension)" />
+    <Compile Include="$(AzureCoreSharedSources)TaskExtensions.cs" Link="Shared\%(RecursiveDir)\%(Filename)%(Extension)" />
+    <PackageReference Include="CloudNative.CloudEvents" />
+    <PackageReference Include="System.Text.Json" />
+  </ItemGroup>
+  <ItemGroup>
+    <ProjectReference Include="..\..\..\eventgrid\Azure.Messaging.EventGrid\src\Azure.Messaging.EventGrid.csproj" />
+  </ItemGroup>
+</Project>

sdk/eventgrid/Microsoft.Azure.CloudNative.CloudEvents.EventGrid/tests/CloudEventTests.cs

@@ -0,0 +1,137 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.IO;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using Azure;
+using Azure.Core;
+using Azure.Core.Pipeline;
+using Azure.Core.TestFramework;
+using Azure.Messaging.EventGrid;
+using CloudNative.CloudEvents;
+using NUnit.Framework;
+using CloudEvent = CloudNative.CloudEvents.CloudEvent;
+
+namespace Microsoft.Azure.CloudNative.CloudEvents.EventGrid.Tests
+{
+    public class CloudEventTests
+    {
+
+        private const string TraceParentHeaderName = "traceparent";
+        private const string TraceStateHeaderName = "tracestate";
+
+        private static readonly JsonEventFormatter s_eventFormatter = new JsonEventFormatter();
+
+
+        [Test]
+        [TestCase(false, false)]
+        [TestCase(true, true)]
+        [TestCase(true, false)]
+        [TestCase(false, true)]
+        public async Task SetsTraceParentExtension(bool inclTraceparent, bool inclTracestate)
+        {
+            var mockTransport = new MockTransport(new MockResponse(200));
+            var options = new EventGridPublisherClientOptions
+            {
+                Transport = mockTransport
+            };
+            EventGridPublisherClient client =
+               new EventGridPublisherClient(
+                   new Uri("http://localHost"),
+                   new AzureKeyCredential("fakeKey"),
+                   options);
+            var activity = new Activity($"{nameof(EventGridPublisherClient)}.{nameof(EventGridPublisherClient.SendEvents)}");
+            activity.SetW3CFormat();
+            activity.Start();
+            List<CloudEvent> eventsList = new List<CloudEvent>();
+            for (int i = 0; i < 10; i++)
+            {
+                var cloudEvent =
+                    new CloudEvent(
+                        "record",
+                        new Uri("http://localHost"),
+                        Guid.NewGuid().ToString(),
+                        DateTime.Now);
+
+                if (inclTraceparent && inclTracestate && i % 2 == 0)
+                {
+                    cloudEvent.GetAttributes().Add("traceparent", "traceparentValue");
+                }
+                if (inclTracestate && i % 2 == 0)
+                {
+                    cloudEvent.GetAttributes().Add("tracestate", "param:value");
+                }
+                eventsList.Add(cloudEvent);
+            }
+            await client.SendCloudEventsAsync(eventsList);
+
+            activity.Stop();
+            List<CloudEvent> cloudEvents = DeserializeRequest(mockTransport.SingleRequest);
+            IEnumerator<CloudEvent> cloudEnum = eventsList.GetEnumerator();
+            foreach (CloudEvent cloudEvent in cloudEvents)
+            {
+                cloudEnum.MoveNext();
+                IDictionary<string, object> cloudEventAttr = cloudEnum.Current.GetAttributes();
+                if (cloudEventAttr.ContainsKey(TraceParentHeaderName) &&
+                    cloudEventAttr.ContainsKey(TraceStateHeaderName))
+                {
+                    Assert.AreEqual(
+                        cloudEventAttr[TraceParentHeaderName],
+                        cloudEvent.GetAttributes()[TraceParentHeaderName]);
+
+                    Assert.AreEqual(
+                        cloudEventAttr[TraceStateHeaderName],
+                        cloudEvent.GetAttributes()[TraceStateHeaderName]);
+                }
+                else if (cloudEventAttr.ContainsKey(TraceParentHeaderName))
+                {
+                    Assert.AreEqual(
+                        cloudEventAttr[TraceParentHeaderName],
+                        cloudEvent.GetAttributes()[TraceParentHeaderName]);
+                }
+                else if (cloudEventAttr.ContainsKey(TraceStateHeaderName))
+                {
+                    Assert.AreEqual(
+                       cloudEventAttr[TraceStateHeaderName],
+                       cloudEvent.GetAttributes()[TraceStateHeaderName]);
+                }
+                else
+                {
+                    Assert.AreEqual(
+                       activity.Id,
+                       cloudEvent.GetAttributes()[TraceParentHeaderName]);
+                }
+            }
+        }
+
+        private static List<CloudEvent> DeserializeRequest(Request request)
+        {
+            var content = request.Content;
+            var stream = new MemoryStream();
+            content.WriteTo(stream, CancellationToken.None);
+            stream.Position = 0;
+            JsonDocument requestDocument = JsonDocument.Parse(stream);
+            var cloudEvents = new List<CloudEvent>();
+            // Parse JsonElement into separate events, deserialize event envelope properties
+            if (requestDocument.RootElement.ValueKind == JsonValueKind.Object)
+            {
+                var bytes = JsonSerializer.SerializeToUtf8Bytes(requestDocument.RootElement, typeof(JsonElement));
+                cloudEvents.Add(s_eventFormatter.DecodeStructuredEvent(bytes));
+            }
+            else if (requestDocument.RootElement.ValueKind == JsonValueKind.Array)
+            {
+                foreach (JsonElement property in requestDocument.RootElement.EnumerateArray())
+                {
+                    var bytes = JsonSerializer.SerializeToUtf8Bytes(property, typeof(JsonElement));
+                    cloudEvents.Add(s_eventFormatter.DecodeStructuredEvent(bytes));
+                }
+            }
+            return cloudEvents;
+        }
+    }
+}