Support binding to EventData

extensions/Worker.Extensions.EventHubs/release_notes.md

@@ -4,6 +4,6 @@
 - My change description (#PR/#issue)
 -->
 
-### Microsoft.Azure.Functions.Worker.Extensions.EventHubs <version>
+### Microsoft.Azure.Functions.Worker.Extensions.EventHubs 5.4.0-preview1
 
-- Add `BindingCapabilities` attribute to EventHubTrigger to express function-level retry capabilities. (#1457)
+- Add support for binding to `EventData`(#1609)

extensions/Worker.Extensions.EventHubs/src/Constants.cs

@@ -0,0 +1,12 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+namespace Microsoft.Azure.Functions.Worker.Extensions.EventHubs
+{
+    internal static class Constants
+    {
+        internal const string BinaryContentType = "application/octet-stream";
+
+        internal const string BindingSource = "AzureEventHubsEventData";
+    }
+}

extensions/Worker.Extensions.EventHubs/src/EventDataConverter.cs

@@ -0,0 +1,58 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using Azure.Core.Amqp;
+using Microsoft.Azure.Functions.Worker.Converters;
+using Microsoft.Azure.Functions.Worker.Core;
+using Microsoft.Azure.Functions.Worker.Extensions.Abstractions;
+using Azure.Messaging.EventHubs;
+using Microsoft.Azure.Functions.Worker.Extensions.EventHubs;
+
+namespace Microsoft.Azure.Functions.Worker
+{
+    [SupportsDeferredBinding]
+    [SupportedConverterType(typeof(EventData))]
+    [SupportedConverterType(typeof(EventData[]))]
+    internal class EventDataConverter : IInputConverter
+    {
+        public ValueTask<ConversionResult> ConvertAsync(ConverterContext context)
+        {
+            try
+            {
+                ConversionResult result = context?.Source switch
+                {
+                    ModelBindingData binding => ConversionResult.Success(ConvertToEventData(binding)),
+                    // Only array collections are currently supported, which matches the behavior of the in-proc extension.
+                    CollectionModelBindingData collection => ConversionResult.Success(collection.ModelBindingDataArray
+                        .Select(ConvertToEventData).ToArray()),
+                    _ => ConversionResult.Unhandled()
+                };
+                return new ValueTask<ConversionResult>(result);
+            }
+            catch (Exception exception)
+            {
+                return new ValueTask<ConversionResult>(ConversionResult.Failed(exception));
+            }
+        }
+
+        private EventData ConvertToEventData(ModelBindingData binding)
+        {
+            if (binding?.Source is not Constants.BindingSource)
+            {
+                throw new InvalidOperationException(
+                    $"Unexpected binding source. Only '{Constants.BindingSource}' is supported.");
+            }
+
+            if (binding.ContentType != Constants.BinaryContentType)
+            {
+                throw new InvalidOperationException(
+                    $"Unexpected content-type. Only '{Constants.BinaryContentType}' is supported.");
+            }
+
+            return new EventData(AmqpAnnotatedMessage.FromBytes(binding.Content));
+        }
+    }
+}

extensions/Worker.Extensions.EventHubs/src/EventHubTriggerAttribute.cs

@@ -1,14 +1,17 @@
 // Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
-using Microsoft.Azure.Functions.Worker.Extensions.Abstractions;
+using Microsoft.Azure.Functions.Worker.Converters;
+using Microsoft.Azure.Functions.Worker.Extensions.Abstractions;
 
 namespace Microsoft.Azure.Functions.Worker
 {
     /// <summary>
     /// Attribute used to mark a function that should be triggered by Event Hubs messages.
     /// </summary>
     [BindingCapabilities(KnownBindingCapabilities.FunctionLevelRetry)]
+    [AllowConverterFallback(true)]
+    [InputConverter(typeof(EventDataConverter))]
     public sealed class EventHubTriggerAttribute : TriggerBindingAttribute, ISupportCardinality
     {
         // Batch by default

extensions/Worker.Extensions.EventHubs/src/Properties/AssemblyInfo.cs

@@ -1,6 +1,8 @@
 // Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
+using System.Runtime.CompilerServices;
 using Microsoft.Azure.Functions.Worker.Extensions.Abstractions;
 
-[assembly: ExtensionInformation("Microsoft.Azure.WebJobs.Extensions.EventHubs", "5.3.0")]
+[assembly: ExtensionInformation("Microsoft.Azure.WebJobs.Extensions.EventHubs", "5.4.0")]
+[assembly: InternalsVisibleTo("Microsoft.Azure.Functions.WorkerExtension.Tests, PublicKey=00240000048000009400000006020000002400005253413100040000010001005148be37ac1d9f58bd40a2e472c9d380d635b6048278f7d47480b08c928858f0f7fe17a6e4ce98da0e7a7f0b8c308aecd9e9b02d7e9680a5b5b75ac7773cec096fbbc64aebd429e77cb5f89a569a79b28e9c76426783f624b6b70327eb37341eb498a2c3918af97c4860db6cdca4732787150841e395a29cfacb959c1fd971c1")]

extensions/Worker.Extensions.EventHubs/src/Worker.Extensions.EventHubs.csproj

@@ -6,14 +6,21 @@
     <Description>Azure Event Hubs extensions for .NET isolated functions</Description>
 
     <!--Version information-->
-    <VersionPrefix>5.3.0</VersionPrefix>
+    <VersionPrefix>5.4.0</VersionPrefix>
+    <VersionSuffix>-preview1</VersionSuffix>
 
   </PropertyGroup>
 
   <Import Project="..\..\..\build\Extensions.props" />
 
   <ItemGroup>
+    <ProjectReference Include="..\..\..\src\DotNetWorker.Core\DotNetWorker.Core.csproj" />
     <ProjectReference Include="..\..\Worker.Extensions.Abstractions\src\Worker.Extensions.Abstractions.csproj" />
   </ItemGroup>
 
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Azure" Version="1.6.3" />
+    <PackageReference Include="Azure.Messaging.EventHubs" Version="5.9.2" />
+  </ItemGroup>
+
 </Project>

extensions/Worker.Extensions.ServiceBus/release_notes.md

@@ -4,6 +4,7 @@
 - My change description (#PR/#issue)
 -->
 
-### Microsoft.Azure.Functions.Worker.Extensions.ServiceBus <version>
+### Microsoft.Azure.Functions.Worker.Extensions.ServiceBus 5.10.0-preview3
 
-- <entry>
+- Make Constants class internal (#1609)
+- Return ConversionResult.Failed when the wrong content-type or binding source is used (#1609)

extensions/Worker.Extensions.ServiceBus/src/Constants.cs

@@ -1,11 +1,12 @@
 // Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
-
 namespace Microsoft.Azure.Functions.Worker.Extensions.ServiceBus
 {
-    public static class Constants
+    internal static class Constants
     {
         internal const string BinaryContentType = "application/octet-stream";
+
+        internal const string BindingSource = "AzureServiceBusReceivedMessage";
     }
 }

extensions/Worker.Extensions.ServiceBus/src/ServiceBusReceivedMessageConverter.cs

@@ -13,36 +13,48 @@
 
 namespace Microsoft.Azure.Functions.Worker
 {
-
     [SupportsDeferredBinding]
     [SupportedConverterType(typeof(ServiceBusReceivedMessage))]
     [SupportedConverterType(typeof(ServiceBusReceivedMessage[]))]
     internal class ServiceBusReceivedMessageConverter : IInputConverter
     {
         public ValueTask<ConversionResult> ConvertAsync(ConverterContext context)
         {
-            ConversionResult result = context?.Source switch
+            try
+            {
+                ConversionResult result = context?.Source switch
+                {
+                    ModelBindingData binding => ConversionResult.Success(ConvertToServiceBusReceivedMessage(binding)),
+                    // Only array collections are currently supported, which matches the behavior of the in-proc extension.
+                    CollectionModelBindingData collection => ConversionResult.Success(collection.ModelBindingDataArray
+                        .Select(ConvertToServiceBusReceivedMessage).ToArray()),
+                    _ => ConversionResult.Unhandled()
+                };
+                return new ValueTask<ConversionResult>(result);
+            }
+            catch (Exception exception)
             {
-                ModelBindingData binding => ConversionResult.Success(ConvertToServiceBusReceivedMessage(binding)),
-                // Only array collections are currently supported, which matches the behavior of the in-proc extension.
-                CollectionModelBindingData collection => ConversionResult.Success(collection.ModelBindingDataArray
-                    .Select(ConvertToServiceBusReceivedMessage).ToArray()),
-                _ => ConversionResult.Unhandled()
-            };
-            return new ValueTask<ConversionResult>(result);
+                return new ValueTask<ConversionResult>(ConversionResult.Failed(exception));
+            }
         }
 
         private ServiceBusReceivedMessage ConvertToServiceBusReceivedMessage(ModelBindingData binding)
         {
-            // The lock token is a 16 byte GUID
-            const int lockTokenLength = 16;
+            if (binding?.Source is not Constants.BindingSource)
+            {
+                throw new InvalidOperationException(
+                    $"Unexpected binding source. Only '{Constants.BindingSource}' is supported.");
+            }
 
             if (binding.ContentType != Constants.BinaryContentType)
             {
                 throw new InvalidOperationException(
                     $"Unexpected content-type. Only '{Constants.BinaryContentType}' is supported.");
             }
 
+            // The lock token is a 16 byte GUID
+            const int lockTokenLength = 16;
+
             ReadOnlyMemory<byte> bytes = binding.Content.ToMemory();
             ReadOnlyMemory<byte> lockTokenBytes = bytes.Slice(0, lockTokenLength);
             ReadOnlyMemory<byte> messageBytes = bytes.Slice(lockTokenLength, bytes.Length - lockTokenLength);

extensions/Worker.Extensions.ServiceBus/src/Worker.Extensions.ServiceBus.csproj

@@ -7,7 +7,7 @@
 
     <!--Version information-->
     <VersionPrefix>5.10.0</VersionPrefix>
-    <VersionSuffix>-preview2</VersionSuffix>
+    <VersionSuffix>-preview3</VersionSuffix>
 
     <!--Temporarily opting out of documentation. Pending documentation-->
     <GenerateDocumentationFile>false</GenerateDocumentationFile>

samples/WorkerBindingSamples/EventHubs/EventDataSamples.cs

@@ -0,0 +1,68 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using Azure.Messaging.EventHubs;
+using Microsoft.Azure.Functions.Worker;
+using Microsoft.Extensions.Logging;
+
+namespace SampleApp
+{
+    /// <summary>
+    /// Samples demonstrating binding to the <see cref="EventData"/> type.
+    /// </summary>
+    public class EventDataSamples
+    {
+        private readonly ILogger<EventDataSamples> _logger;
+
+        public EventDataSamples(ILogger<EventDataSamples> logger)
+        {
+            _logger = logger;
+        }
+
+        /// <summary>
+        /// This function demonstrates binding to a single <see cref="EventData"/>. Note that when doing so, you must also set the
+        /// <see cref="EventHubTriggerAttribute.IsBatched"/> property to <value>false</value> as the default value of this property is
+        /// <value>true</value>.
+        /// </summary>
+        [Function(nameof(EventDataFunctions))]
+        public void EventDataFunctions(
+            [EventHubTrigger("queue", Connection = "EventHubConnection", IsBatched = false)] EventData @event)
+        {
+            _logger.LogInformation("Event Body: {body}", @event.Body);
+            _logger.LogInformation("Event Content-Type: {contentType}", @event.ContentType);
+        }
+
+        /// <summary>
+        /// This function demonstrates binding to an array of <see cref="EventData"/>.
+        /// </summary>
+        [Function(nameof(EventDataBatchFunction))]
+        public void EventDataBatchFunction(
+            [EventHubTrigger("queue", Connection = "EventHubConnection")] EventData[] events)
+        {
+            foreach (EventData @event in events)
+            {
+                _logger.LogInformation("Event Body: {body}", @event.Body);
+                _logger.LogInformation("Event Content-Type: {contentType}", @event.ContentType);
+            }
+        }
+
+        /// <summary>
+        /// This functions demonstrates that it is possible to bind to both the <see cref="EventData"/> and any of the supported binding contract
+        /// properties at the same time. If attempting this, the <see cref="EventData"/> must be the first parameter. There is not
+        /// much benefit to doing this as all of the binding contract properties are available as properties on the <see cref="EventData"/>.
+        /// </summary>
+        [Function(nameof(EventDataWithStringPropertiesFunction))]
+        public void EventDataWithStringPropertiesFunction(
+            [EventHubTrigger("queue", Connection = "EventHubConnection")]
+            EventData @event, string contentType, long offset)
+        {
+            // The ContentType property and the contentType parameter are the same.
+            _logger.LogInformation("Event Content-Type: {contentType}", @event.ContentType);
+            _logger.LogInformation("Event Content-Type: {contentType}", contentType);
+
+            // Similarly the Offset property and the offset parameter are the same.
+            _logger.LogInformation("Event offset: {offset}", @event.Offset);
+            _logger.LogInformation("Event offset: {offset}", offset);
+        }
+    }
+}

samples/WorkerBindingSamples/WorkerBindingSamples.csproj

@@ -16,6 +16,7 @@
         <ProjectReference Include="..\..\extensions\Worker.Extensions.Abstractions\src\Worker.Extensions.Abstractions.csproj" />
         <ProjectReference Include="..\..\extensions\Worker.Extensions.CosmosDB\src\Worker.Extensions.CosmosDB.csproj" />
         <ProjectReference Include="..\..\extensions\Worker.Extensions.ServiceBus\src\Worker.Extensions.ServiceBus.csproj" />
+        <ProjectReference Include="..\..\extensions\Worker.Extensions.EventHubs\src\Worker.Extensions.EventHubs.csproj" />
         <ProjectReference Include="..\..\extensions\Worker.Extensions.Storage\src\Worker.Extensions.Storage.csproj" />
         <ProjectReference Include="..\..\src\DotNetWorker.ApplicationInsights\DotNetWorker.ApplicationInsights.csproj" />
         <ProjectReference Include="..\..\src\DotNetWorker\DotNetWorker.csproj" />