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,10 @@
+// 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";
+    }
+}

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

@@ -0,0 +1,45 @@
+// 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)
+        {
+            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);
+        }
+
+        private EventData ConvertToEventData(ModelBindingData binding)
+        {
+            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/EventHubExtensionStartup.cs

@@ -0,0 +1,25 @@
+// 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 Microsoft.Azure.Functions.Worker;
+using Microsoft.Azure.Functions.Worker.Core;
+using Microsoft.Extensions.Azure;
+
+[assembly: WorkerExtensionStartup(typeof(EventHubExtensionStartup))]
+
+namespace Microsoft.Azure.Functions.Worker
+{
+    public class EventHubExtensionStartup : WorkerExtensionStartup
+    {
+        public override void Configure(IFunctionsWorkerApplicationBuilder applicationBuilder)
+        {
+            if (applicationBuilder == null)
+            {
+                throw new ArgumentNullException(nameof(applicationBuilder));
+            }
+
+            applicationBuilder.Services.AddAzureClientsCore(); // Adds AzureComponentFactory
+        }
+    }
+}

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,6 @@
 - 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)

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

@@ -1,10 +1,9 @@
 // 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";
     }

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" />

test/FunctionMetadataGeneratorTests/FunctionMetadataGeneratorTests.cs

@@ -11,6 +11,7 @@
 using System.Reflection;
 using System.Threading.Tasks;
 using Azure.Data.Tables;
+using Azure.Messaging.EventHubs;
 using Azure.Messaging.ServiceBus;
 using Azure.Storage.Blobs;
 using Microsoft.Azure.Functions.Tests;
@@ -971,6 +972,59 @@ void ValidateServiceBusBatchTrigger(ExpandoObject b)
             }
         }
 
+        [Fact]
+        public void EventHubs_SDKTypeBindings()
+        {
+            var generator = new FunctionMetadataGenerator();
+            var module = ModuleDefinition.ReadModule(_thisAssembly.Location);
+            var typeDef = TestUtility.GetTypeDefinition(typeof(SDKTypeBindings_EventHubs));
+            var functions = generator.GenerateFunctionMetadata(typeDef);
+            var extensions = generator.Extensions;
+
+            Assert.Equal(2, functions.Count());
+
+            AssertDictionary(extensions, new Dictionary<string, string>
+            {
+                { "Microsoft.Azure.WebJobs.Extensions.EventHubs", "5.4.0" },
+            });
+
+            var eventHubTriggerFunction = functions.Single(p => p.Name == nameof(SDKTypeBindings_EventHubs.EventHubTriggerFunction));
+
+            ValidateFunction(eventHubTriggerFunction, nameof(SDKTypeBindings_EventHubs.EventHubTriggerFunction), GetEntryPoint(nameof(SDKTypeBindings_EventHubs), nameof(SDKTypeBindings_EventHubs.EventHubTriggerFunction)),
+                ValidateEventHubTrigger);
+
+            var eventHubBatchTriggerFunction = functions.Single(p => p.Name == nameof(SDKTypeBindings_EventHubs.EventHubBatchTriggerFunction));
+
+            ValidateFunction(eventHubBatchTriggerFunction, nameof(SDKTypeBindings_EventHubs.EventHubBatchTriggerFunction), GetEntryPoint(nameof(SDKTypeBindings_EventHubs), nameof(SDKTypeBindings_EventHubs.EventHubBatchTriggerFunction)),
+                ValidateEventHubBatchTrigger);
+
+            void ValidateEventHubTrigger(ExpandoObject b)
+            {
+                AssertExpandoObject(b, new Dictionary<string, object>
+                {
+                    { "Name", "event" },
+                    { "Type", "eventHubTrigger" },
+                    { "Direction", "In" },
+                    { "eventHubName", "hub" },
+                    { "Cardinality", "One" },
+                    { "Properties", new Dictionary<string, object>( ) { { "SupportsDeferredBinding" , "True"} } }
+                });
+            }
+
+            void ValidateEventHubBatchTrigger(ExpandoObject b)
+            {
+                AssertExpandoObject(b, new Dictionary<string, object>
+                {
+                    { "Name", "events" },
+                    { "Type", "eventHubTrigger" },
+                    { "Direction", "In" },
+                    { "eventHubName", "hub" },
+                    { "Cardinality", "Many" },
+                    { "Properties", new Dictionary<string, object>( ) { { "SupportsDeferredBinding" , "True"} } }
+                });
+            }
+        }
+
         private class EventHubNotBatched
         {
             [Function("EventHubTrigger")]
@@ -1213,6 +1267,23 @@ public static void ServiceBusBatchTriggerFunction(
             }
         }
 
+        private class SDKTypeBindings_EventHubs
+        {
+            [Function(nameof(EventHubTriggerFunction))]
+            public static void EventHubTriggerFunction(
+                [EventHubTrigger("hub", IsBatched = false)] EventData @event)
+            {
+                throw new NotImplementedException();
+            }
+
+            [Function(nameof(EventHubBatchTriggerFunction))]
+            public static void EventHubBatchTriggerFunction(
+                [EventHubTrigger("hub")] EventData[] events)
+            {
+                throw new NotImplementedException();
+            }
+        }
+
         private class ExternalType_Return
         {
             public const string FunctionName = "BasicHttpWithExternalTypeReturn";