implementation

Author: m-redding

sdk/eventgrid/Azure.Messaging.EventGrid.Namespaces/src/Customization/CloudEventRequestContent.cs

@@ -0,0 +1,91 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+using Azure.Core;
+
+namespace Azure.Messaging.EventGrid.Namespaces
+{
+    internal class CloudEventRequestContent : RequestContent
+    {
+        private readonly IEnumerable<CloudEvent> _cloudEvents;
+        private const string TraceParentHeaderName = "traceparent";
+        private const string TraceStateHeaderName = "tracestate";
+        private readonly bool _isDistributedTracingEnabled;
+        private RequestContent _serializedContent;
+
+        public CloudEventRequestContent(IEnumerable<CloudEvent> cloudEvents, bool isDistributedTracingEnabled)
+        {
+            _cloudEvents = cloudEvents;
+            _isDistributedTracingEnabled = isDistributedTracingEnabled;
+        }
+
+        public CloudEventRequestContent(CloudEvent cloudEvent, bool isDistributedTracingEnabled)
+        {
+            _cloudEvents = [ cloudEvent ];
+            _isDistributedTracingEnabled = isDistributedTracingEnabled;
+        }
+
+        public override void Dispose()
+        {
+        }
+
+        public override bool TryComputeLength(out long length)
+        {
+            EnsureSerialized();
+            return _serializedContent.TryComputeLength(out length);
+        }
+
+        public override void WriteTo(Stream stream, CancellationToken cancellationToken)
+        {
+            EnsureSerialized();
+            _serializedContent.WriteTo(stream, cancellationToken);
+        }
+
+        public override async Task WriteToAsync(Stream stream, CancellationToken cancellationToken)
+        {
+            EnsureSerialized();
+            await _serializedContent.WriteToAsync(stream, cancellationToken).ConfigureAwait(false);
+        }
+
+        private void EnsureSerialized()
+        {
+            if (_serializedContent != null)
+            {
+                return;
+            }
+
+            if (_isDistributedTracingEnabled)
+            {
+                string currentActivityId = null;
+                string traceState = null;
+                Activity currentActivity = Activity.Current;
+                if (currentActivity != null && (currentActivity.IdFormat == ActivityIdFormat.W3C))
+                {
+                    currentActivityId = currentActivity.Id;
+                    traceState = currentActivity.TraceStateString;
+                }
+
+                foreach (CloudEvent cloudEvent in _cloudEvents)
+                {
+                    if (currentActivityId != null &&
+                        !cloudEvent.ExtensionAttributes.ContainsKey(TraceParentHeaderName) &&
+                        !cloudEvent.ExtensionAttributes.ContainsKey(TraceStateHeaderName))
+                    {
+                        cloudEvent.ExtensionAttributes.Add(TraceParentHeaderName, currentActivityId);
+                        if (traceState != null)
+                        {
+                            cloudEvent.ExtensionAttributes.Add(TraceStateHeaderName, traceState);
+                        }
+                    }
+                }
+            }
+
+            _serializedContent = RequestContent.Create(_cloudEvents);
+        }
+    }
+}

sdk/eventgrid/Azure.Messaging.EventGrid.Namespaces/src/Customization/EventGridSenderClient.cs

@@ -18,6 +18,7 @@ namespace Azure.Messaging.EventGrid.Namespaces
     public partial class EventGridSenderClient
     {
         private readonly string _topicName;
+        private readonly bool _isDistributedTracingEnabled;
 
         /// <summary> Initializes a new instance of EventGridSenderClient. </summary>
         /// <param name="endpoint"> The host name of the namespace, e.g. namespaceName1.westus-1.eventgrid.azure.net. </param>
@@ -66,6 +67,7 @@ public EventGridSenderClient(Uri endpoint,
             _endpoint = endpoint;
             _apiVersion = options.Version;
             _topicName = topicName;
+            _isDistributedTracingEnabled = options.Diagnostics.IsDistributedTracingEnabled;
         }
 
         /// <summary> Initializes a new instance of EventGridSenderClient. </summary>
@@ -93,6 +95,7 @@ public EventGridSenderClient(Uri endpoint,
             _endpoint = endpoint;
             _apiVersion = options.Version;
             _topicName = topicName;
+            _isDistributedTracingEnabled = options.Diagnostics.IsDistributedTracingEnabled;
         }
 
         /// <summary> Publish Single Cloud Event to namespace topic. In case of success, the server responds with an HTTP 200 status code with an empty JSON object in response. Otherwise, the server can return various error codes. For example, 401: which indicates authorization failure, 403: which indicates quota exceeded or message is too large, 410: which indicates that specific topic is not found, 400: for bad request, and 500: for internal server error. </summary>
@@ -104,7 +107,7 @@ public virtual Response Send(CloudEvent cloudEvent, CancellationToken cancellati
             Argument.AssertNotNull(cloudEvent, nameof(cloudEvent));
 
             RequestContext context = FromCancellationToken(cancellationToken);
-            return Send(_topicName, RequestContent.Create(cloudEvent), context);
+            return Send(_topicName, new CloudEventRequestContent(cloudEvent, _isDistributedTracingEnabled), context);
         }
 
         /// <summary> Publish Single Cloud Event to namespace topic. In case of success, the server responds with an HTTP 200 status code with an empty JSON object in response. Otherwise, the server can return various error codes. For example, 401: which indicates authorization failure, 403: which indicates quota exceeded or message is too large, 410: which indicates that specific topic is not found, 400: for bad request, and 500: for internal server error. </summary>
@@ -118,7 +121,7 @@ public virtual async Task<Response> SendAsync(
             Argument.AssertNotNull(cloudEvent, nameof(cloudEvent));
 
             RequestContext context = FromCancellationToken(cancellationToken);
-            return await SendAsync(_topicName, RequestContent.Create(cloudEvent), context).ConfigureAwait(false);
+            return await SendAsync(_topicName, new CloudEventRequestContent(cloudEvent, _isDistributedTracingEnabled), context).ConfigureAwait(false);
         }
 
         /// <summary> Publish Batch Cloud Event to namespace topic. In case of success, the server responds with an HTTP 200 status code with an empty JSON object in response. Otherwise, the server can return various error codes. For example, 401: which indicates authorization failure, 403: which indicates quota exceeded or message is too large, 410: which indicates that specific topic is not found, 400: for bad request, and 500: for internal server error. </summary>
@@ -136,7 +139,7 @@ public virtual async Task<Response> SendAsync(IEnumerable<CloudEvent> cloudEvent
             scope.Start();
             try
             {
-                return await SendEventsAsync(_topicName, RequestContent.Create(cloudEvents), context).ConfigureAwait(false);
+                return await SendEventsAsync(_topicName, new CloudEventRequestContent(cloudEvents, _isDistributedTracingEnabled), context).ConfigureAwait(false);
             }
             catch (Exception e)
             {
@@ -160,7 +163,7 @@ public virtual Response Send(IEnumerable<CloudEvent> cloudEvents, CancellationTo
             scope.Start();
             try
             {
-                return SendEvents(_topicName, RequestContent.Create(cloudEvents), context);
+                return SendEvents(_topicName, new CloudEventRequestContent(cloudEvents, _isDistributedTracingEnabled), context);
             }
             catch (Exception e)
             {