Add StorageClientProvider support to EventHubs extension

sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/CHANGELOG.md

@@ -8,6 +8,10 @@
 
 - The web proxy specified in configuration is now respected.
 
+#### New Features
+
+- Added support for specifying `accountName` or `blobServiceUri` for the checkpoint connection.
+
 ## 5.0.0-beta.4 (2021-04-06)
 
 ### Changes

sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/src/Config/EventHubClientFactory.cs

@@ -22,6 +22,7 @@ internal class EventHubClientFactory
         private readonly AzureComponentFactory _componentFactory;
         private readonly EventHubOptions _options;
         private readonly INameResolver _nameResolver;
+        private readonly CheckpointClientProvider _checkpointClientProvider;
         private readonly ConcurrentDictionary<string, EventHubProducerClient> _producerCache;
         private readonly ConcurrentDictionary<string, IEventHubConsumerClient> _consumerCache = new();
 
@@ -30,13 +31,15 @@ public EventHubClientFactory(
             AzureComponentFactory componentFactory,
             IOptions<EventHubOptions> options,
             INameResolver nameResolver,
-            AzureEventSourceLogForwarder forwarder)
+            AzureEventSourceLogForwarder forwarder,
+            CheckpointClientProvider checkpointClientProvider)
         {
             forwarder.Start();
             _configuration = configuration;
             _componentFactory = componentFactory;
             _options = options.Value;
             _nameResolver = nameResolver;
+            _checkpointClientProvider = checkpointClientProvider;
             _producerCache = new ConcurrentDictionary<string, EventHubProducerClient>();
         }
 
@@ -166,11 +169,7 @@ internal IEventHubConsumerClient GetEventHubConsumerClient(string eventHubName,
 
         internal BlobContainerClient GetCheckpointStoreClient()
         {
-            var section = _configuration.GetWebJobsConnectionStringSection(ConnectionStringNames.Storage);
-            var options = _componentFactory.CreateClientOptions(typeof(BlobClientOptions), null, section);
-            var credential = _componentFactory.CreateTokenCredential(section);
-            var client = (BlobServiceClient)_componentFactory.CreateClient(typeof(BlobServiceClient), section, credential, options);
-
+            var client =  _checkpointClientProvider.Get(ConnectionStringNames.Storage);
             return client.GetBlobContainerClient(_options.CheckpointContainer);
         }
 

sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/src/Config/EventHubWebJobsBuilderExtensions.cs

@@ -7,6 +7,7 @@
 using Azure.Messaging.EventHubs.Consumer;
 using Microsoft.Azure.WebJobs;
 using Microsoft.Azure.WebJobs.EventHubs;
+using Microsoft.Azure.WebJobs.EventHubs.Processor;
 using Microsoft.Extensions.Azure;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
@@ -83,6 +84,7 @@ public static IWebJobsBuilder AddEventHubs(this IWebJobsBuilder builder, Action<
 
             builder.Services.AddAzureClientsCore();
             builder.Services.AddSingleton<EventHubClientFactory>();
+            builder.Services.AddSingleton<CheckpointClientProvider>();
             builder.Services.Configure<EventHubOptions>(configure);
             builder.Services.PostConfigure<EventHubOptions>(ConfigureInitialOffsetOptions);
 

sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/src/Microsoft.Azure.WebJobs.Extensions.EventHubs.csproj

@@ -26,6 +26,7 @@
     <Compile Include="../../Azure.Messaging.EventHubs.Shared/src/Core/EventHubsRetryOptionsExtensions.cs" LinkBase="SharedSource" />
     <Compile Include="../../Azure.Messaging.EventHubs.Shared/src/BlobCheckpointStore/*.cs" LinkBase="SharedSource" />
     <Compile Include="../../../extensions/Microsoft.Azure.WebJobs.Extensions.Clients/src/Shared/WebJobsConfigurationExtensions.cs" LinkBase="SharedSource" />
+    <Compile Include="../../../extensions/Microsoft.Azure.WebJobs.Extensions.Clients/src/Shared/StorageClientProvider.cs" LinkBase="SharedSource" />
     <Compile Update="Resources.Designer.cs">
       <DesignTime>True</DesignTime>
       <AutoGen>True</AutoGen>

sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/src/Processor/CheckpointClientProvider.cs

@@ -0,0 +1,20 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using Azure.Storage.Blobs;
+using Microsoft.Azure.WebJobs.Extensions.Clients.Shared;
+using Microsoft.Extensions.Azure;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.Logging;
+
+namespace Microsoft.Azure.WebJobs.EventHubs.Processor
+{
+    internal class CheckpointClientProvider : StorageClientProvider<BlobServiceClient, BlobClientOptions>
+    {
+        public CheckpointClientProvider(IConfiguration configuration, AzureComponentFactory componentFactory, AzureEventSourceLogForwarder logForwarder, ILogger<BlobServiceClient> logger)
+            : base(configuration, componentFactory, logForwarder, logger) { }
+
+        /// <inheritdoc/>
+        protected override string ServiceUriSubDomain => "blob";
+    }
+}

sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/tests/EventHubEndToEndTests.cs

@@ -198,6 +198,42 @@ await AssertCanSendReceiveMessage(host =>
                     })));
         }
 
+        [Test]
+        public async Task CanSendAndReceive_BlobServiceUri_InConfiguration()
+        {
+            await AssertCanSendReceiveMessage(host =>
+                host.ConfigureAppConfiguration(configurationBuilder =>
+                    configurationBuilder.AddInMemoryCollection(new Dictionary<string, string>()
+                    {
+                        {"TestConnection:fullyQualifiedNamespace", EventHubsTestEnvironment.Instance.FullyQualifiedNamespace},
+                        {"TestConnection:clientId", EventHubsTestEnvironment.Instance.ClientId},
+                        {"TestConnection:clientSecret", EventHubsTestEnvironment.Instance.ClientSecret},
+                        {"TestConnection:tenantId", EventHubsTestEnvironment.Instance.TenantId},
+                        {"AzureWebJobsStorage:blobServiceUri", GetServiceUri()},
+                        {"AzureWebJobsStorage:clientId", EventHubsTestEnvironment.Instance.ClientId},
+                        {"AzureWebJobsStorage:clientSecret", EventHubsTestEnvironment.Instance.ClientSecret},
+                        {"AzureWebJobsStorage:tenantId", EventHubsTestEnvironment.Instance.TenantId},
+                    })));
+        }
+
+        [Test]
+        public async Task CanSendAndReceive_AccountName_InConfiguration()
+        {
+            await AssertCanSendReceiveMessage(host =>
+                host.ConfigureAppConfiguration(configurationBuilder =>
+                    configurationBuilder.AddInMemoryCollection(new Dictionary<string, string>()
+                    {
+                        {"TestConnection:fullyQualifiedNamespace", EventHubsTestEnvironment.Instance.FullyQualifiedNamespace},
+                        {"TestConnection:clientId", EventHubsTestEnvironment.Instance.ClientId},
+                        {"TestConnection:clientSecret", EventHubsTestEnvironment.Instance.ClientSecret},
+                        {"TestConnection:tenantId", EventHubsTestEnvironment.Instance.TenantId},
+                        {"AzureWebJobsStorage:accountName", StorageTestEnvironment.Instance.StorageAccountName},
+                        {"AzureWebJobsStorage:clientId", EventHubsTestEnvironment.Instance.ClientId},
+                        {"AzureWebJobsStorage:clientSecret", EventHubsTestEnvironment.Instance.ClientSecret},
+                        {"AzureWebJobsStorage:tenantId", EventHubsTestEnvironment.Instance.TenantId},
+                    })));
+        }
+
         [Test]
         public void ThrowsIfBindingToASingleEvent()
         {

sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/tests/EventHubsClientFactoryTests.cs

@@ -35,7 +35,8 @@ public void GetEventHubClient_AddsConnection(string expectedPathName, string con
             EventHubOptions options = new EventHubOptions();
             var configuration = CreateConfiguration(new KeyValuePair<string, string>("connection", connectionString));
 
-            var factory = new EventHubClientFactory(configuration, Mock.Of<AzureComponentFactory>(), Options.Create(options), new DefaultNameResolver(configuration), new AzureEventSourceLogForwarder(new NullLoggerFactory()));
+            var factory = new EventHubClientFactory(configuration, Mock.Of<AzureComponentFactory>(), Options.Create(options),
+                new DefaultNameResolver(configuration), new AzureEventSourceLogForwarder(new NullLoggerFactory()), Mock.Of<CheckpointClientProvider>());
 
             var client = factory.GetEventHubProducerClient(expectedPathName, "connection");
             Assert.AreEqual(expectedPathName, client.EventHubName);
@@ -47,7 +48,8 @@ public void CreatesClientsFromConfigWithConnectionString()
             EventHubOptions options = new EventHubOptions();
             var configuration = CreateConfiguration(new KeyValuePair<string, string>("connection", ConnectionString));
 
-            var factory = new EventHubClientFactory(configuration, Mock.Of<AzureComponentFactory>(), Options.Create(options), new DefaultNameResolver(configuration), new AzureEventSourceLogForwarder(new NullLoggerFactory()));
+            var factory = new EventHubClientFactory(configuration, Mock.Of<AzureComponentFactory>(), Options.Create(options),
+                new DefaultNameResolver(configuration), new AzureEventSourceLogForwarder(new NullLoggerFactory()), Mock.Of<CheckpointClientProvider>());
             var producer = factory.GetEventHubProducerClient("k1", "connection");
             var consumer = factory.GetEventHubConsumerClient("k1", "connection", null);
             var host = factory.GetEventProcessorHost("k1", "connection", null);
@@ -72,7 +74,8 @@ public void CreatesClientsFromConfigWithFullyQualifiedNamespace()
 
             var configuration = CreateConfiguration(new KeyValuePair<string, string>("connection:fullyQualifiedNamespace", "test89123-ns-x.servicebus.windows.net"));
 
-            var factory = new EventHubClientFactory(configuration, componentFactoryMock.Object, Options.Create(options), new DefaultNameResolver(configuration), new AzureEventSourceLogForwarder(new NullLoggerFactory()));
+            var factory = new EventHubClientFactory(configuration, componentFactoryMock.Object, Options.Create(options), new DefaultNameResolver(configuration),
+                new AzureEventSourceLogForwarder(new NullLoggerFactory()), Mock.Of<CheckpointClientProvider>());
             var producer = factory.GetEventHubProducerClient("k1", "connection");
             var consumer = factory.GetEventHubConsumerClient("k1", "connection", null);
             var host = factory.GetEventProcessorHost("k1", "connection", null);
@@ -92,7 +95,8 @@ public void ConsumersAndProducersAreCached()
             EventHubOptions options = new EventHubOptions();
             var configuration = CreateConfiguration(new KeyValuePair<string, string>("connection", ConnectionString));
 
-            var factory = new EventHubClientFactory(configuration, Mock.Of<AzureComponentFactory>(), Options.Create(options), new DefaultNameResolver(configuration), new AzureEventSourceLogForwarder(new NullLoggerFactory()));
+            var factory = new EventHubClientFactory(configuration, Mock.Of<AzureComponentFactory>(), Options.Create(options),
+                new DefaultNameResolver(configuration), new AzureEventSourceLogForwarder(new NullLoggerFactory()), Mock.Of<CheckpointClientProvider>());
             var producer = factory.GetEventHubProducerClient("k1", "connection");
             var consumer = factory.GetEventHubConsumerClient("k1", "connection", null);
             var producer2 = factory.GetEventHubProducerClient("k1", "connection");
@@ -116,7 +120,8 @@ public void UsesDefaultConnectionToStorageAccount()
                         null, null))
                 .Returns(new BlobServiceClient(configuration["AzureWebJobsStorage"]));
 
-            var factory = new EventHubClientFactory(configuration, factoryMock.Object, Options.Create(options), new DefaultNameResolver(configuration), new AzureEventSourceLogForwarder(new NullLoggerFactory()));
+            var factory = new EventHubClientFactory(configuration, factoryMock.Object, Options.Create(options), new DefaultNameResolver(configuration),
+                new AzureEventSourceLogForwarder(new NullLoggerFactory()), Mock.Of<CheckpointClientProvider>());
 
             var client = factory.GetCheckpointStoreClient();
             Assert.AreEqual("azure-webjobs-eventhub", client.Name);
@@ -138,7 +143,8 @@ public void RespectsConnectionOptionsForProducer(string expectedPathName, string
             };
 
             var configuration = CreateConfiguration(new KeyValuePair<string, string>("connection", connectionString));
-            var factory = new EventHubClientFactory(configuration, Mock.Of<AzureComponentFactory>(), Options.Create(options), new DefaultNameResolver(configuration), new AzureEventSourceLogForwarder(new NullLoggerFactory()));
+            var factory = new EventHubClientFactory(configuration, Mock.Of<AzureComponentFactory>(), Options.Create(options),
+                new DefaultNameResolver(configuration), new AzureEventSourceLogForwarder(new NullLoggerFactory()), Mock.Of<CheckpointClientProvider>());
 
             var producer = factory.GetEventHubProducerClient(expectedPathName, "connection");
             EventHubConnection connection = (EventHubConnection)typeof(EventHubProducerClient).GetProperty("Connection", BindingFlags.NonPublic | BindingFlags.Instance)
@@ -168,7 +174,8 @@ public void RespectsConnectionOptionsForConsumer(string expectedPathName, string
             };
 
             var configuration = CreateConfiguration(new KeyValuePair<string, string>("connection", connectionString));
-            var factory = new EventHubClientFactory(configuration, Mock.Of<AzureComponentFactory>(), Options.Create(options), new DefaultNameResolver(configuration), new AzureEventSourceLogForwarder(new NullLoggerFactory()));
+            var factory = new EventHubClientFactory(configuration, Mock.Of<AzureComponentFactory>(), Options.Create(options),
+                new DefaultNameResolver(configuration), new AzureEventSourceLogForwarder(new NullLoggerFactory()), Mock.Of<CheckpointClientProvider>());
 
             var consumer = factory.GetEventHubConsumerClient(expectedPathName, "connection", "consumer");
             var consumerClient = (EventHubConsumerClient)typeof(EventHubConsumerClientImpl)
@@ -212,7 +219,8 @@ public void RespectsConnectionOptionsForProcessor(string expectedPathName, strin
             };
 
             var configuration = CreateConfiguration(new KeyValuePair<string, string>("connection", connectionString));
-            var factory = new EventHubClientFactory(configuration, Mock.Of<AzureComponentFactory>(), Options.Create(options), new DefaultNameResolver(configuration), new AzureEventSourceLogForwarder(new NullLoggerFactory()));
+            var factory = new EventHubClientFactory(configuration, Mock.Of<AzureComponentFactory>(), Options.Create(options),
+                new DefaultNameResolver(configuration), new AzureEventSourceLogForwarder(new NullLoggerFactory()), Mock.Of<CheckpointClientProvider>());
 
             var processor = factory.GetEventProcessorHost(expectedPathName, "connection", "consumer");
             EventProcessorOptions processorOptions = (EventProcessorOptions)typeof(EventProcessor<EventProcessorHostPartition>)
@@ -231,7 +239,8 @@ public void DefaultStrategyIsGreedy()
             EventHubOptions options = new EventHubOptions();
 
             var configuration = CreateConfiguration(new KeyValuePair<string, string>("connection", ConnectionString));
-            var factory = new EventHubClientFactory(configuration, Mock.Of<AzureComponentFactory>(), Options.Create(options), new DefaultNameResolver(configuration), new AzureEventSourceLogForwarder(new NullLoggerFactory()));
+            var factory = new EventHubClientFactory(configuration, Mock.Of<AzureComponentFactory>(), Options.Create(options),
+                new DefaultNameResolver(configuration), new AzureEventSourceLogForwarder(new NullLoggerFactory()), Mock.Of<CheckpointClientProvider>());
 
             var processor = factory.GetEventProcessorHost("connection", "connection", "consumer");
             EventProcessorOptions processorOptions = (EventProcessorOptions)typeof(EventProcessor<EventProcessorHostPartition>)