Add client and model readmes and fix partition context (Azure#19114)

pakrym · azure-sdk · web-flow · commit f0cd63f587ac · 2021-03-01T18:43:14.000-08:00
Co-authored-by: azure-sdk &lt;azuresdk@microsoft.com&gt;
diff --git a/sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/README.md b/sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/README.md
@@ -98,7 +98,6 @@ public static string Run([TimerTrigger("0 */5 * * * *")] TimerInfo myTimer)
 
 To send multiple events from a single Azure Function invocation you can apply the `EventHubAttribute` to the `IAsyncCollector<string>` or `IAsyncCollector<EventData>` parameter.
 
-
 ```C# Snippet:BindingToCollector
 [FunctionName("BindingToCollector")]
 public static async Task Run(
@@ -111,6 +110,39 @@ public static async Task Run(
 }
 ```
 
+### Using binding to strongly-typed models
+
+To use strongly-typed model classes with the EventHub binding apply the `EventHubAttribute` to the model parameter.
+
+```C# Snippet:TriggerSingleModel
+[FunctionName("TriggerSingleModel")]
+public static void Run(
+    [EventHubTrigger("<event_hub_name>", Connection = "<connection_name>")] Dog dog,
+    ILogger logger)
+{
+    logger.LogInformation($"Who's a good dog? {dog.Name} is!");
+}
+```
+
+### Sending multiple events using EventHubProducerClient
+
+You can also bind to the `EventHubProducerClient` directly to have the most control over the event sending.
+
+```C# Snippet:BindingToProducerClient
+[FunctionName("BindingToProducerClient")]
+public static async Task Run(
+    [TimerTrigger("0 */5 * * * *")] TimerInfo myTimer,
+    [EventHub("<event_hub_name>", Connection = "<connection_name>")] EventHubProducerClient eventHubProducerClient)
+{
+    // IAsyncCollector allows sending multiple events in a single function invocation
+    await eventHubProducerClient.SendAsync(new[]
+    {
+        new EventData(new BinaryData($"Event 1 added at: {DateTime.Now}")),
+        new EventData(new BinaryData($"Event 2 added at: {DateTime.Now}"))
+    });
+}
+```
+
 ### Per-event triggers
 
 To run a function every time an event is sent to Event Hub apply the `EventHubTriggerAttribute` to a `string` or `EventData` parameter.
diff --git a/sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/src/Listeners/EventHubListener.cs b/sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/src/Listeners/EventHubListener.cs
@@ -141,7 +141,7 @@ public async Task ProcessEventsAsync(EventProcessorHostPartition context, IEnume
                 var triggerInput = new EventHubTriggerInput
                 {
                     Events = events,
-                    PartitionContext = context
+                    ProcessorPartition = context
                 };
 
                 TriggeredFunctionData input = null;
@@ -271,9 +271,9 @@ private static string GetOperationDetails(EventProcessorHostPartition context, s
                     {
                         writer.WritePropertyName("runtimeInformation");
                         writer.WriteStartObject();
-                        WritePropertyIfNotNull(writer, "lastEnqueuedOffset", context.LastEnqueuedEventProperties.Value.Offset?.ToString(CultureInfo.InvariantCulture));
-                        WritePropertyIfNotNull(writer, "lastSequenceNumber", context.LastEnqueuedEventProperties.Value.SequenceNumber?.ToString(CultureInfo.InvariantCulture));
-                        WritePropertyIfNotNull(writer, "lastEnqueuedTimeUtc", context.LastEnqueuedEventProperties.Value.EnqueuedTime?.ToString("o", CultureInfo.InvariantCulture));
+                        WritePropertyIfNotNull(writer, "lastEnqueuedOffset", context.LastEnqueuedEventProperties.Offset?.ToString(CultureInfo.InvariantCulture));
+                        WritePropertyIfNotNull(writer, "lastSequenceNumber", context.LastEnqueuedEventProperties.SequenceNumber?.ToString(CultureInfo.InvariantCulture));
+                        WritePropertyIfNotNull(writer, "lastEnqueuedTimeUtc", context.LastEnqueuedEventProperties.EnqueuedTime?.ToString("o", CultureInfo.InvariantCulture));
                         writer.WriteEndObject();
                     }
                     writer.WriteEndObject();
diff --git a/sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/src/Processor/EventProcessorHostPartition.cs b/sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/src/Processor/EventProcessorHostPartition.cs
@@ -11,6 +11,8 @@ namespace Microsoft.Azure.WebJobs.EventHubs.Processor
 {
     internal class EventProcessorHostPartition : EventProcessorPartition
     {
+        private PartitionContext _partitionContext;
+
         public EventProcessorHostPartition()
         {
         }
@@ -20,17 +22,21 @@ public EventProcessorHostPartition(string partitionId)
             PartitionId = partitionId;
         }
 
+        public PartitionContext PartitionContext => _partitionContext ??= new EventProcessorHostPartitionContext(this);
+
         public string Owner => ProcessorHost.Identifier;
         public string EventHubPath => ProcessorHost.EventHubName;
         public CheckpointInfo? Checkpoint { get; set; }
 
-        public LastEnqueuedEventProperties? LastEnqueuedEventProperties
+        public LastEnqueuedEventProperties LastEnqueuedEventProperties
         {
             get
             {
+                if (ReadLastEnqueuedEventPropertiesFunc == null) return default;
+
                 try
                 {
-                    return ReadLastEnqueuedEventPropertiesFunc?.Invoke(PartitionId);
+                    return ReadLastEnqueuedEventPropertiesFunc.Invoke(PartitionId);
                 }
                 catch (EventHubsException e) when (e.Reason == EventHubsException.FailureReason.ClientClosed)
                 {
@@ -49,5 +55,17 @@ public async Task CheckpointAsync(EventData checkpointEvent)
             await ProcessorHost.CheckpointAsync(PartitionId, checkpointEvent).ConfigureAwait(false);
             Checkpoint = new CheckpointInfo(checkpointEvent.Offset, checkpointEvent.SequenceNumber);
         }
+
+        private class EventProcessorHostPartitionContext : PartitionContext
+        {
+            private readonly EventProcessorHostPartition _hostPartition;
+
+            public EventProcessorHostPartitionContext(EventProcessorHostPartition hostPartition) : base(hostPartition.PartitionId)
+            {
+                _hostPartition = hostPartition;
+            }
+
+            public override LastEnqueuedEventProperties ReadLastEnqueuedEventProperties() => _hostPartition.LastEnqueuedEventProperties;
+        }
     }
 }
diff --git a/sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/src/Triggers/EventHubTriggerBindingStrategy.cs b/sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/src/Triggers/EventHubTriggerBindingStrategy.cs
@@ -78,7 +78,7 @@ public Dictionary<string, object> GetBindingData(EventHubTriggerInput value)
             }
 
             var bindingData = new Dictionary<string, object>(StringComparer.OrdinalIgnoreCase);
-            SafeAddValue(() => bindingData.Add(nameof(value.PartitionContext), value.PartitionContext));
+            SafeAddValue(() => bindingData.Add("PartitionContext", value.ProcessorPartition?.PartitionContext));
 
             if (value.IsSingleDispatch)
             {
diff --git a/sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/src/Triggers/EventHubTriggerInput.cs b/sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/src/Triggers/EventHubTriggerInput.cs
@@ -4,7 +4,9 @@
 using Azure.Messaging.EventHubs;
 using System.Collections.Generic;
 using System.Globalization;
+using Azure.Messaging.EventHubs.Consumer;
 using Azure.Messaging.EventHubs.Primitives;
+using Microsoft.Azure.WebJobs.EventHubs.Processor;
 
 namespace Microsoft.Azure.WebJobs.EventHubs
 {
@@ -17,7 +19,7 @@ internal sealed class EventHubTriggerInput
 
         internal EventData[] Events { get; set; }
 
-        internal EventProcessorPartition PartitionContext { get; set; }
+        internal EventProcessorHostPartition ProcessorPartition { get; set; }
 
         public bool IsSingleDispatch
         {
@@ -31,7 +33,7 @@ public static EventHubTriggerInput New(EventData eventData)
         {
             return new EventHubTriggerInput
             {
-                PartitionContext = null,
+                ProcessorPartition = null,
                 Events = new EventData[]
                 {
                       eventData
@@ -45,7 +47,7 @@ public EventHubTriggerInput GetSingleEventTriggerInput(int idx)
             return new EventHubTriggerInput
             {
                 Events = this.Events,
-                PartitionContext = this.PartitionContext,
+                ProcessorPartition = this.ProcessorPartition,
                 _selector = idx
             };
         }
diff --git a/sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/tests/EventHubEndToEndTests.cs b/sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/tests/EventHubEndToEndTests.cs
@@ -49,7 +49,7 @@ public async Task EventHub_PocoBinding()
             var (jobHost, host) = BuildHost<EventHubTestBindToPocoJobs>();
             using (jobHost)
             {
-                await jobHost.CallAsync(nameof(EventHubTestBindToPocoJobs.SendEvent_TestHub), new { input = "{ Name: 'foo', Value: 'data' }" });
+                await jobHost.CallAsync(nameof(EventHubTestBindToPocoJobs.SendEvent_TestHub));
 
                 bool result = _eventWait.WaitOne(Timeout);
                 Assert.True(result);
@@ -131,6 +131,19 @@ public async Task EventHub_SingleDispatch_BinaryData()
             AssertSingleDispatchLogs(host);
         }
 
+        [Test]
+        public async Task EventHub_ProducerClient()
+        {
+            var (jobHost, host) = BuildHost<EventHubTestClientDispatch>();
+            using (jobHost)
+            {
+                await jobHost.CallAsync(nameof(EventHubTestClientDispatch.SendEvents));
+
+                bool result = _eventWait.WaitOne(Timeout);
+                Assert.True(result);
+            }
+        }
+
         private static void AssertSingleDispatchLogs(IHost host)
         {
             IEnumerable<LogMessage> logMessages = host.GetTestLoggerProvider()
@@ -385,13 +398,34 @@ public static void SendEvent_TestHub(string input, [EventHub(TestHubName, Connec
 
             public static void ProcessSingleEvent([EventHubTrigger(TestHubName, Connection = TestHubName)] string evt,
                 string partitionKey, DateTime enqueuedTimeUtc, IDictionary<string, object> properties,
-                IDictionary<string, object> systemProperties)
+                IDictionary<string, object> systemProperties,
+                PartitionContext partitionContext)
             {
                 Assert.True((DateTime.Now - enqueuedTimeUtc).TotalSeconds < 30);
 
                 Assert.AreEqual("value1", properties["TestProp1"]);
                 Assert.AreEqual("value2", properties["TestProp2"]);
 
+                Assert.NotNull(partitionContext.PartitionId);
+                Assert.NotNull(partitionContext.ReadLastEnqueuedEventProperties());
+
+                _eventWait.Set();
+            }
+        }
+
+        public class EventHubTestClientDispatch
+        {
+            public static async Task SendEvents([EventHub(TestHubName, Connection = TestHubName)] EventHubProducerClient producer)
+            {
+                await producer.SendAsync(new[]
+                {
+                    new EventData(new BinaryData("Event 1")),
+                });
+            }
+
+            public static void ProcessSingleEvent([EventHubTrigger(TestHubName, Connection = TestHubName)] EventData eventData)
+            {
+                Assert.AreEqual(eventData.EventBody.ToString(), "Event 1");
                 _eventWait.Set();
             }
         }
@@ -429,16 +463,16 @@ public static void ProcessSingleEvent([EventHubTrigger(TestHubName, Connection =
 
         public class EventHubTestBindToPocoJobs
         {
-            public static void SendEvent_TestHub(string input, [EventHub(TestHubName, Connection = TestHubName)] out EventData evt)
+            public static void SendEvent_TestHub([EventHub(TestHubName, Connection = TestHubName)] out TestPoco evt)
             {
-                evt = new EventData(Encoding.UTF8.GetBytes(input));
+                evt = new TestPoco() {Value = "data", Name = "foo"};
             }
 
-            public static void BindToPoco([EventHubTrigger(TestHubName, Connection = TestHubName)] TestPoco input, string value, string name, ILogger logger)
+            public static void BindToPoco([EventHubTrigger(TestHubName, Connection = TestHubName)] TestPoco input, ILogger logger)
             {
-                Assert.AreEqual(input.Value, value);
-                Assert.AreEqual(input.Name, name);
-                logger.LogInformation($"PocoValues({name},{value})");
+                Assert.AreEqual(input.Value, "data");
+                Assert.AreEqual(input.Name, "foo");
+                logger.LogInformation($"PocoValues(foo,data)");
                 _eventWait.Set();
             }
         }
diff --git a/sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/tests/EventHubTests.cs b/sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/tests/EventHubTests.cs
@@ -73,13 +73,13 @@ public void GetBindingData_SingleDispatch_ReturnsExpectedValue()
             var evt = GetSystemProperties(new byte[] { });
 
             var input = EventHubTriggerInput.New(evt);
-            input.PartitionContext = GetPartitionContext();
+            input.ProcessorPartition = GetPartitionContext();
 
             var strategy = new EventHubTriggerBindingStrategy();
             var bindingData = strategy.GetBindingData(input);
 
             Assert.AreEqual(7, bindingData.Count);
-            Assert.AreSame(input.PartitionContext, bindingData["PartitionContext"]);
+            Assert.AreSame(input.ProcessorPartition.PartitionContext, bindingData["PartitionContext"]);
             Assert.AreEqual(evt.PartitionKey, bindingData["PartitionKey"]);
             Assert.AreEqual(evt.Offset, bindingData["Offset"]);
             Assert.AreEqual(evt.SequenceNumber, bindingData["SequenceNumber"]);
@@ -118,13 +118,13 @@ public void GetBindingData_MultipleDispatch_ReturnsExpectedValue()
             var input = new EventHubTriggerInput
             {
                 Events = events,
-                PartitionContext = GetPartitionContext(),
+                ProcessorPartition = GetPartitionContext(),
             };
             var strategy = new EventHubTriggerBindingStrategy();
             var bindingData = strategy.GetBindingData(input);
 
             Assert.AreEqual(7, bindingData.Count);
-            Assert.AreSame(input.PartitionContext, bindingData["PartitionContext"]);
+            Assert.AreSame(input.ProcessorPartition.PartitionContext, bindingData["PartitionContext"]);
 
             // verify an array was created for each binding data type
             Assert.AreEqual(events.Length, ((string[])bindingData["PartitionKeyArray"]).Length);
@@ -145,8 +145,8 @@ public void TriggerStrategy()
             string data = "123";
 
             var strategy = new EventHubTriggerBindingStrategy();
-            EventHubTriggerInput triggerInput = strategy.ConvertFromString(data);
 
+            EventHubTriggerInput triggerInput = strategy.ConvertFromString(data);
             var contract = strategy.GetBindingData(triggerInput);
 
             EventData single = strategy.BindSingle(triggerInput, null);
diff --git a/sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/tests/Samples/BindingToProducerClient.cs b/sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/tests/Samples/BindingToProducerClient.cs
@@ -0,0 +1,28 @@
+// 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.Threading.Tasks;
+using Azure.Messaging.EventHubs;
+using Azure.Messaging.EventHubs.Producer;
+
+namespace Microsoft.Azure.WebJobs.Extensions.EventHubs.Tests.Samples
+{
+    public static class BindingToProducerClient
+    {
+        #region Snippet:BindingToProducerClient
+        [FunctionName("BindingToProducerClient")]
+        public static async Task Run(
+            [TimerTrigger("0 */5 * * * *")] TimerInfo myTimer,
+            [EventHub("<event_hub_name>", Connection = "<connection_name>")] EventHubProducerClient eventHubProducerClient)
+        {
+            // IAsyncCollector allows sending multiple events in a single function invocation
+            await eventHubProducerClient.SendAsync(new[]
+            {
+                new EventData(new BinaryData($"Event 1 added at: {DateTime.Now}")),
+                new EventData(new BinaryData($"Event 2 added at: {DateTime.Now}"))
+            });
+        }
+        #endregion
+    }
+}
diff --git a/sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/tests/Samples/TriggerSingleModel.cs b/sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/tests/Samples/TriggerSingleModel.cs
@@ -0,0 +1,27 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using Microsoft.Extensions.Logging;
+
+namespace Microsoft.Azure.WebJobs.Extensions.EventHubs.Tests.Samples
+{
+    public static class TriggerSingleModel
+    {
+        public class Dog
+        {
+            public string Name { get; set; }
+            public string Breed { get; set; }
+            public int Age { get; set; }
+        }
+
+        #region Snippet:TriggerSingleModel
+        [FunctionName("TriggerSingleModel")]
+        public static void Run(
+            [EventHubTrigger("<event_hub_name>", Connection = "<connection_name>")] Dog dog,
+            ILogger logger)
+        {
+            logger.LogInformation($"Who's a good dog? {dog.Name} is!");
+        }
+        #endregion
+    }
+}

Original file line number	Diff line number	Diff line change
`@@ -11,6 +11,8 @@ namespace Microsoft.Azure.WebJobs.EventHubs.Processor`
`11`	`11`	`{`
`12`	`12`	`internal class EventProcessorHostPartition : EventProcessorPartition`
`13`	`13`	`{`
	`14`	`+ private PartitionContext _partitionContext;`
	`15`	`+`
`14`	`16`	`public EventProcessorHostPartition()`
`15`	`17`	`{`
`16`	`18`	`}`
`@@ -20,17 +22,21 @@ public EventProcessorHostPartition(string partitionId)`
`20`	`22`	`PartitionId = partitionId;`
`21`	`23`	`}`
`22`	`24`
	`25`	`+ public PartitionContext PartitionContext => _partitionContext ??= new EventProcessorHostPartitionContext(this);`
	`26`	`+`
`23`	`27`	`public string Owner => ProcessorHost.Identifier;`
`24`	`28`	`public string EventHubPath => ProcessorHost.EventHubName;`
`25`	`29`	`public CheckpointInfo? Checkpoint { get; set; }`
`26`	`30`
`27`		`- public LastEnqueuedEventProperties? LastEnqueuedEventProperties`
	`31`	`+ public LastEnqueuedEventProperties LastEnqueuedEventProperties`
`28`	`32`	`{`
`29`	`33`	`get`
`30`	`34`	`{`
	`35`	`+ if (ReadLastEnqueuedEventPropertiesFunc == null) return default;`
	`36`	`+`
`31`	`37`	`try`
`32`	`38`	`{`
`33`		`- return ReadLastEnqueuedEventPropertiesFunc?.Invoke(PartitionId);`
	`39`	`+ return ReadLastEnqueuedEventPropertiesFunc.Invoke(PartitionId);`
`34`	`40`	`}`
`35`	`41`	`catch (EventHubsException e) when (e.Reason == EventHubsException.FailureReason.ClientClosed)`
`36`	`42`	`{`
`@@ -49,5 +55,17 @@ public async Task CheckpointAsync(EventData checkpointEvent)`
`49`	`55`	`await ProcessorHost.CheckpointAsync(PartitionId, checkpointEvent).ConfigureAwait(false);`
`50`	`56`	`Checkpoint = new CheckpointInfo(checkpointEvent.Offset, checkpointEvent.SequenceNumber);`
`51`	`57`	`}`
	`58`	`+`
	`59`	`+ private class EventProcessorHostPartitionContext : PartitionContext`
	`60`	`+ {`
	`61`	`+ private readonly EventProcessorHostPartition _hostPartition;`
	`62`	`+`
	`63`	`+ public EventProcessorHostPartitionContext(EventProcessorHostPartition hostPartition) : base(hostPartition.PartitionId)`
	`64`	`+ {`
	`65`	`+ _hostPartition = hostPartition;`
	`66`	`+ }`
	`67`	`+`
	`68`	`+ public override LastEnqueuedEventProperties ReadLastEnqueuedEventProperties() => _hostPartition.LastEnqueuedEventProperties;`
	`69`	`+ }`
`52`	`70`	`}`
`53`	`71`	`}`
Original file line number	Diff line number	Diff line change
`@@ -78,7 +78,7 @@ public Dictionary<string, object> GetBindingData(EventHubTriggerInput value)`
`78`	`78`	`}`
`79`	`79`
`80`	`80`	`var bindingData = new Dictionary<string, object>(StringComparer.OrdinalIgnoreCase);`
`81`		`- SafeAddValue(() => bindingData.Add(nameof(value.PartitionContext), value.PartitionContext));`
	`81`	`+ SafeAddValue(() => bindingData.Add("PartitionContext", value.ProcessorPartition?.PartitionContext));`
`82`	`82`
`83`	`83`	`if (value.IsSingleDispatch)`
`84`	`84`	`{`
Original file line number	Diff line number	Diff line change
`@@ -4,7 +4,9 @@`
`4`	`4`	`using Azure.Messaging.EventHubs;`
`5`	`5`	`using System.Collections.Generic;`
`6`	`6`	`using System.Globalization;`
	`7`	`+using Azure.Messaging.EventHubs.Consumer;`
`7`	`8`	`using Azure.Messaging.EventHubs.Primitives;`
	`9`	`+using Microsoft.Azure.WebJobs.EventHubs.Processor;`
`8`	`10`
`9`	`11`	`namespace Microsoft.Azure.WebJobs.EventHubs`
`10`	`12`	`{`
`@@ -17,7 +19,7 @@ internal sealed class EventHubTriggerInput`
`17`	`19`
`18`	`20`	`internal EventData[] Events { get; set; }`
`19`	`21`
`20`		`- internal EventProcessorPartition PartitionContext { get; set; }`
	`22`	`+ internal EventProcessorHostPartition ProcessorPartition { get; set; }`
`21`	`23`
`22`	`24`	`public bool IsSingleDispatch`
`23`	`25`	`{`
`@@ -31,7 +33,7 @@ public static EventHubTriggerInput New(EventData eventData)`
`31`	`33`	`{`
`32`	`34`	`return new EventHubTriggerInput`
`33`	`35`	`{`
`34`		`- PartitionContext = null,`
	`36`	`+ ProcessorPartition = null,`
`35`	`37`	`Events = new EventData[]`
`36`	`38`	`{`
`37`	`39`	`eventData`
`@@ -45,7 +47,7 @@ public EventHubTriggerInput GetSingleEventTriggerInput(int idx)`
`45`	`47`	`return new EventHubTriggerInput`
`46`	`48`	`{`
`47`	`49`	`Events = this.Events,`
`48`		`- PartitionContext = this.PartitionContext,`
	`50`	`+ ProcessorPartition = this.ProcessorPartition,`
`49`	`51`	`_selector = idx`
`50`	`52`	`};`
`51`	`53`	`}`