Merged PR 912267: Change functions binding client to amqp

Removed TransportType as an input and only use AMQP because it recovers when the connection is lost
Changed ModuleClientCache to have only one ModuleClient instance, multiple client connections can't be configured because it creates it uses CreateFromEnvironmentAsync.

edge-modules/functions/binding/src/Microsoft.Azure.Devices.Edge.Functions.Binding/EdgeHubAsyncCollector.cs

@@ -18,7 +18,6 @@ namespace Microsoft.Azure.Devices.Edge.Functions.Binding
     /// </summary>
     public class EdgeHubAsyncCollector : IAsyncCollector<Message>
     {
-        readonly TransportType transportType;
         readonly EdgeHubAttribute attribute;
         readonly int batchSize;
 
@@ -35,11 +34,9 @@ public class EdgeHubAsyncCollector : IAsyncCollector<Message>
         /// <summary>
         /// Create a sender around the given client. 
         /// </summary>
-        /// <param name="transportType">Device Client transport type. </param>
         /// <param name="attribute">Attributes used by EdgeHub when receiving a message from function.</param>
-        public EdgeHubAsyncCollector(TransportType transportType, EdgeHubAttribute attribute)
+        public EdgeHubAsyncCollector(EdgeHubAttribute attribute)
         {
-            this.transportType = transportType;
             this.attribute = attribute;
             this.batchSize = attribute.BatchSize > 0 ? (attribute.BatchSize > MaxBatchSize ? MaxBatchSize : attribute.BatchSize) : DefaultBatchSize;
         }
@@ -104,7 +101,7 @@ protected virtual async Task SendBatchAsync(IList<Message> batch)
                 return;
             }
 
-            ModuleClient client = await ModuleClientCache.Instance.GetOrCreateAsync(this.transportType).ConfigureAwait(false);
+            ModuleClient client = await ModuleClientCache.Instance.GetOrCreateAsync().ConfigureAwait(false);
 
             if (string.IsNullOrEmpty(this.attribute.OutputName))
             {

edge-modules/functions/binding/src/Microsoft.Azure.Devices.Edge.Functions.Binding/EdgeHubTriggerBindingProvider.cs

@@ -19,14 +19,8 @@ namespace Microsoft.Azure.Devices.Edge.Functions.Binding
     class EdgeHubTriggerBindingProvider : ITriggerBindingProvider
     {
         readonly ConcurrentDictionary<string, IList<EdgeHubMessageProcessor>> receivers = new ConcurrentDictionary<string, IList<EdgeHubMessageProcessor>>();
-        readonly TransportType transportType;
         ModuleClient moduleClient;
 
-        public EdgeHubTriggerBindingProvider(TransportType transportType)
-        {
-            this.transportType = transportType;
-        }
-
         public async Task<ITriggerBinding> TryCreateAsync(TriggerBindingProviderContext context)
         {
             if (context == null)
@@ -46,7 +40,7 @@ public async Task<ITriggerBinding> TryCreateAsync(TriggerBindingProviderContext
                 throw new InvalidOperationException($"Can't bind EdgeHubTriggerAttribute to type '{parameter.ParameterType}'.");
             }
 
-            await TrySetEventDefaultHandlerAsync().ConfigureAwait(false);
+            await this.TrySetEventDefaultHandlerAsync().ConfigureAwait(false);
 
             var messageProcessor = new EdgeHubMessageProcessor();
             var triggerBinding = new EdgeHubTriggerBinding(context.Parameter, messageProcessor);
@@ -77,7 +71,7 @@ async Task TrySetEventDefaultHandlerAsync()
                 return;
             }
 
-            this.moduleClient = await ModuleClientCache.Instance.GetOrCreateAsync(transportType).ConfigureAwait(false);
+            this.moduleClient = await ModuleClientCache.Instance.GetOrCreateAsync().ConfigureAwait(false);
             await this.moduleClient.SetMessageHandlerAsync(this.FunctionsMessageHandler, null).ConfigureAwait(false);
         }
 

edge-modules/functions/binding/src/Microsoft.Azure.Devices.Edge.Functions.Binding/ModuleClientCache.cs

@@ -3,21 +3,21 @@
 namespace Microsoft.Azure.Devices.Edge.Functions.Binding
 {
     using System;
-    using System.Collections.Concurrent;
     using System.Threading.Tasks;
     using Microsoft.Azure.Devices.Client;
-    using Microsoft.Azure.Devices.Client.Transport.Mqtt;
     using Microsoft.Azure.Devices.Edge.Util;
+    using Microsoft.Azure.Devices.Edge.Util.Concurrency;
     using Microsoft.Azure.Devices.Edge.Util.TransientFaultHandling;
-    using ExponentialBackoff = Microsoft.Azure.Devices.Edge.Util.TransientFaultHandling.ExponentialBackoff;
+    using ExponentialBackoff = Util.TransientFaultHandling.ExponentialBackoff;
 
     class ModuleClientCache
     {
         const int RetryCount = 5;
         static readonly ITransientErrorDetectionStrategy TimeoutErrorDetectionStrategy = new DelegateErrorDetectionStrategy(ex => ex.HasTimeoutException());
         static readonly RetryStrategy TransientRetryStrategy =
             new ExponentialBackoff(RetryCount, TimeSpan.FromSeconds(2), TimeSpan.FromSeconds(60), TimeSpan.FromSeconds(4));
-        readonly ConcurrentDictionary<string, Task<ModuleClient>> clients = new ConcurrentDictionary<string, Task<ModuleClient>>();
+        readonly AsyncLock asyncLock = new AsyncLock();
+        ModuleClient client;
 
         // Private constructor to ensure single instance
         ModuleClientCache()
@@ -26,10 +26,11 @@ class ModuleClientCache
 
         public static ModuleClientCache Instance { get; } = new ModuleClientCache();
 
-        public Task<ModuleClient> GetOrCreateAsync(TransportType transportType) =>
-            this.clients.GetOrAdd(
-                transportType.ToString(),
-                client =>
+        public async Task<ModuleClient> GetOrCreateAsync()
+        {
+            using (await this.asyncLock.LockAsync())
+            {
+                if (this.client == null)
                 {
                     var retryPolicy = new RetryPolicy(TimeoutErrorDetectionStrategy, TransientRetryStrategy);
                     retryPolicy.Retrying += (_, args) =>
@@ -40,16 +41,16 @@ public Task<ModuleClient> GetOrCreateAsync(TransportType transportType) =>
                             Console.WriteLine("Retrying...");
                         }
                     };
-                    return retryPolicy.ExecuteAsync(() => this.CreateModuleClient(transportType));
-                });
+                    this.client = await retryPolicy.ExecuteAsync(() => this.CreateModuleClient());
+                }
 
+                return this.client;
+            }
+        }
 
-        async Task<ModuleClient> CreateModuleClient(TransportType transportType)
+        async Task<ModuleClient> CreateModuleClient()
         {
-            var mqttSetting = new MqttTransportSettings(transportType);
-
-            ITransportSettings[] settings = { mqttSetting };
-            ModuleClient moduleClient = await ModuleClient.CreateFromEnvironmentAsync(settings).ConfigureAwait(false);
+            ModuleClient moduleClient = await ModuleClient.CreateFromEnvironmentAsync().ConfigureAwait(false);
 
             moduleClient.ProductInfo = "Microsoft.Azure.Devices.Edge.Functions.Binding";
             return moduleClient;

edge-modules/functions/binding/src/Microsoft.Azure.Devices.Edge.Functions.Binding/Utils.cs

@@ -3,7 +3,6 @@
 namespace Microsoft.Azure.Devices.Edge.Functions.Binding
 {
     using System.Collections.Generic;
-    using System.ComponentModel;
     using Microsoft.Azure.Devices.Client;
 
     static class Utils
@@ -19,22 +18,5 @@ public static Message GetMessageCopy(byte[] payload, Message message)
 
             return copy;
         }
-
-        public static TransportType ToTransportType(string transportName, TransportType defaultTransport)
-        {
-            TransportType transportType;
-
-            if (transportName == null)
-            {
-                transportType = defaultTransport;
-            }
-            else
-            {
-                TypeConverter converter = TypeDescriptor.GetConverter(typeof(TransportType));
-                transportType = (TransportType)converter.ConvertFromInvariantString(transportName);
-            }
-
-            return transportType;
-        }
     }
 }

edge-modules/functions/binding/src/Microsoft.Azure.Devices.Edge.Functions.Binding/bindings/EdgeHubCollectorBuilder.cs

@@ -7,16 +7,9 @@ namespace Microsoft.Azure.Devices.Edge.Functions.Binding.Bindings
 
     class EdgeHubCollectorBuilder : IConverter<EdgeHubAttribute, IAsyncCollector<Message>>
     {
-        readonly TransportType transportType;
-
-        public EdgeHubCollectorBuilder(TransportType transportType)
-        {
-            this.transportType = transportType;
-        }
-
         public IAsyncCollector<Message> Convert(EdgeHubAttribute attribute)
         {
-            return new EdgeHubAsyncCollector(this.transportType, attribute);
+            return new EdgeHubAsyncCollector(attribute);
         }
     }
 }

edge-modules/functions/binding/src/Microsoft.Azure.Devices.Edge.Functions.Binding/config/EdgeHubExtensionConfigProvider.cs

@@ -5,7 +5,6 @@ namespace Microsoft.Azure.Devices.Edge.Functions.Binding.Config
     using System;
     using Microsoft.Azure.Devices.Client;
     using Microsoft.Azure.Devices.Edge.Functions.Binding.Bindings;
-    using Microsoft.Azure.WebJobs;
     using Microsoft.Azure.WebJobs.Host;
     using Microsoft.Azure.WebJobs.Host.Config;
     using Microsoft.Azure.WebJobs.Host.Triggers;
@@ -18,26 +17,20 @@ public class EdgeHubExtensionConfigProvider : IExtensionConfigProvider
     {
         public void Initialize(ExtensionConfigContext context)
         {
-            const string ClientTransportType = "ClientTransportType";
-
             if (context == null)
             {
                 throw new ArgumentNullException(nameof(context));
             }
 
             var extensions = context.Config.GetService<IExtensionRegistry>();
-            var nameResolver = context.Config.GetService<INameResolver>();
-
-            TransportType transportType = Utils.ToTransportType(
-                nameResolver.Resolve(ClientTransportType), TransportType.Mqtt_Tcp_Only);
 
             // register trigger binding provider
-            var triggerBindingProvider = new EdgeHubTriggerBindingProvider(transportType);
+            var triggerBindingProvider = new EdgeHubTriggerBindingProvider();
             extensions.RegisterExtension<ITriggerBindingProvider>(triggerBindingProvider);
 
             extensions.RegisterBindingRules<EdgeHubAttribute>();
             FluentBindingRule<EdgeHubAttribute> rule = context.AddBindingRule<EdgeHubAttribute>();
-            rule.BindToCollector<Message>(typeof(EdgeHubCollectorBuilder), transportType);
+            rule.BindToCollector<Message>(typeof(EdgeHubCollectorBuilder));
 
             context.AddConverter<Message, string>(this.MessageConverter);
             context.AddConverter<string, Message>(this.ConvertToMessage);