ServiceBus MessageFactory caching (#881)

src/Microsoft.Azure.WebJobs.ServiceBus/MessagingProvider.cs

@@ -2,8 +2,8 @@
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
 using System;
+using System.Collections.Concurrent;
 using System.Globalization;
-using System.Threading.Tasks;
 using Microsoft.Azure.WebJobs.Host;
 using Microsoft.ServiceBus;
 using Microsoft.ServiceBus.Messaging;
@@ -18,6 +18,11 @@ public class MessagingProvider
     {
         private readonly ServiceBusConfiguration _config;
 
+        /// <summary>
+        /// Cache of <see cref="MessagingFactory"/> instances by connection string.
+        /// </summary>
+        private readonly ConcurrentDictionary<string, MessagingFactory> _messagingFactoryCache = new ConcurrentDictionary<string, MessagingFactory>();
+
         /// <summary>
         /// Constructs a new instance.
         /// </summary>
@@ -51,6 +56,7 @@ public virtual NamespaceManager CreateNamespaceManager(string connectionStringNa
         /// <param name="connectionStringName">Optional connection string name indicating the connection string to use.
         /// If null, the default connection string on the <see cref="ServiceBusConfiguration"/> will be used.</param>
         /// <returns>A <see cref="MessagingFactory"/>.</returns>
+        /// <remarks>For performance reasons, factories are cached per connection string.</remarks>
         public virtual MessagingFactory CreateMessagingFactory(string entityPath, string connectionStringName = null)
         {
             if (string.IsNullOrEmpty(entityPath))
@@ -60,7 +66,15 @@ public virtual MessagingFactory CreateMessagingFactory(string entityPath, string
 
             string connectionString = GetConnectionString(connectionStringName);
 
-            return MessagingFactory.CreateFromConnectionString(connectionString);
+            // We cache messaging factories per connection, in accordance with ServiceBus
+            // performance guidelines.
+            // https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-performance-improvements
+            var messagingFactory = _messagingFactoryCache.GetOrAdd(connectionString, (c) =>
+            {
+                return MessagingFactory.CreateFromConnectionString(c);
+            });
+
+            return messagingFactory;
         }
 
         /// <summary>

test/Microsoft.Azure.WebJobs.Host.EndToEndTests/ServiceBusEndToEndTests.cs

@@ -2,9 +2,11 @@
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
 using System;
+using System.Collections.Generic;
 using System.Diagnostics;
 using System.IO;
 using System.Linq;
+using System.Reflection;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Azure.WebJobs.Host.Listeners;
@@ -22,6 +24,7 @@ public class ServiceBusEndToEndTests
         private const int SBTimeout = 60 * 1000;
         private const string QueueNamePrefix = PrefixForAll + "queue-";
         private const string StartQueueName = QueueNamePrefix + "start";
+        private const string BinderQueueName = QueueNamePrefix + "binder";
 
         private const string TopicName = PrefixForAll + "topic";
 
@@ -137,6 +140,24 @@ await TestHelpers.Await(() =>
             }
         }
 
+        [Fact]
+        public async Task ServiceBusBinderTest()
+        {
+            var hostType = typeof(ServiceBusTestJobs);
+            var host = CreateHost(hostType);
+            var method = typeof(ServiceBusTestJobs).GetMethod("ServiceBusBinderTest");
+
+            int numMessages = 10;
+            var args = new { message = "Test Message", numMessages = numMessages };
+            await host.CallAsync(method, args);
+            await host.CallAsync(method, args);
+            await host.CallAsync(method, args);
+
+            var queueName = ResolveName(BinderQueueName);
+            var queueDescription = await _namespaceManager.GetQueueAsync(queueName);
+            Assert.Equal(numMessages * 3, queueDescription.MessageCount);
+        }
+
         [Fact]
         public async Task CustomMessageProcessorTest()
         {
@@ -231,6 +252,9 @@ private void Cleanup()
                 _secondaryNamespaceManager.DeleteQueue(elementName);
             }
 
+            elementName = ResolveName(BinderQueueName);
+            CleanupQueue(elementName);
+
             elementName = ResolveName(QueueNamePrefix + "1");
             CleanupQueue(elementName);
 
@@ -449,6 +473,26 @@ public static void MultipleAccounts(
             {
                 output = input;
             }
+
+            [NoAutomaticTrigger]
+            public static async Task ServiceBusBinderTest(
+                string message,
+                int numMessages,
+                Binder binder)
+            {
+                var attribute = new ServiceBusAttribute(BinderQueueName)
+                {
+                    EntityType = EntityType.Queue
+                };
+                var collector = await binder.BindAsync<IAsyncCollector<string>>(attribute);
+
+                for (int i = 0; i < numMessages; i++)
+                {
+                    await collector.AddAsync(message + i);
+                }
+
+                await collector.FlushAsync();
+            }
         }
 
         /// <summary>

test/Microsoft.Azure.WebJobs.ServiceBus.UnitTests/MessagingProviderTests.cs

@@ -63,6 +63,23 @@ public void CreateMessageReceiver_ReturnsExpectedReceiver()
             Assert.Equal(100, receiver.PrefetchCount);
         }
 
+        [Fact]
+        public void CreateMessagingFactory_CachesPerConnection()
+        {
+            var factory = _provider.CreateMessagingFactory("test");
+
+            Assert.Same(factory, _provider.CreateMessagingFactory("test"));
+            Assert.Same(factory, _provider.CreateMessagingFactory("test"));
+            Assert.Same(factory, _provider.CreateMessagingFactory("test"));
+
+            var factory2 = _provider.CreateMessagingFactory("test", "ServiceBusOverride");
+            Assert.NotSame(factory, factory2);
+
+            Assert.Same(factory2, _provider.CreateMessagingFactory("test", "ServiceBusOverride"));
+            Assert.Same(factory2, _provider.CreateMessagingFactory("test", "ServiceBusOverride"));
+            Assert.Same(factory2, _provider.CreateMessagingFactory("test", "ServiceBusOverride"));
+        }
+
         [Fact]
         public void GetConnectionString_ThrowsIfConnectionStringNullOrEmpty()
         {