Added support for AmqpWebSockets (#153)

* Added support for AmqpWebSockets

* EventHub + ServiceBus ver. 4.4

Author: snakefoot

src/NLog.Extensions.AzureEventHub/EventHubTarget.cs

@@ -120,6 +120,21 @@ public class EventHubTarget : AsyncTaskTarget
         /// </summary>
         public Layout AccessKey { get; set; }
 
+        /// <summary>
+        /// The connection uses the AMQP protocol over web sockets. See also <see cref="EventHubsTransportType.AmqpWebSockets"/>
+        /// </summary>
+        public Layout UseWebSockets { get; set; }
+
+        /// <summary>
+        /// The proxy to use for communication over web sockets.
+        /// </summary>
+        public Layout WebSocketProxyAddress { get; set; }
+
+        /// <summary>
+        /// Custom endpoint address that can be used when establishing the connection.
+        /// </summary>
+        public Layout CustomEndpointAddress { get; set; }
+
         /// <summary>
         /// Gets a list of user properties (aka custom properties) to add to the AMQP message
         /// </summary>
@@ -164,6 +179,9 @@ protected override void InitializeTarget()
             string storageAccountName = string.Empty;
             string storageAccountAccessKey = string.Empty;
             string eventHubName = string.Empty;
+            string useWebSockets = string.Empty;
+            string webSocketProxyAddress = string.Empty;
+            string customEndPointAddress = string.Empty;
 
             var defaultLogEvent = LogEventInfo.CreateNullEvent();
 
@@ -182,7 +200,15 @@ protected override void InitializeTarget()
                     storageAccountAccessKey = AccessKey?.Render(defaultLogEvent);
                 }
 
-                _eventHubService.Connect(connectionString, eventHubName, serviceUri, tenantIdentity, resourceIdentifier, clientIdentity, sharedAccessSignature, storageAccountName, storageAccountAccessKey);
+                useWebSockets = UseWebSockets?.Render(defaultLogEvent) ?? string.Empty;
+                if (!string.IsNullOrEmpty(useWebSockets) && (string.Equals(useWebSockets.Trim(), bool.TrueString, StringComparison.OrdinalIgnoreCase) || string.Equals(useWebSockets.Trim(), "1", StringComparison.OrdinalIgnoreCase)))
+                {
+                    useWebSockets = bool.TrueString;
+                }
+                customEndPointAddress = CustomEndpointAddress?.Render(defaultLogEvent) ?? string.Empty;
+                webSocketProxyAddress = WebSocketProxyAddress?.Render(defaultLogEvent) ?? string.Empty;
+
+                _eventHubService.Connect(connectionString, eventHubName, serviceUri, tenantIdentity, resourceIdentifier, clientIdentity, sharedAccessSignature, storageAccountName, storageAccountAccessKey, bool.TrueString == useWebSockets, webSocketProxyAddress, customEndPointAddress);
                 InternalLogger.Debug("AzureEventHubTarget(Name={0}): Initialized", Name);
             }
             catch (Exception ex)
@@ -470,33 +496,42 @@ private sealed class EventHubService : IEventHubService
 
             public string EventHubName { get; private set; }
 
-            public void Connect(string connectionString, string eventHubName, string serviceUri, string tenantIdentity, string resourceIdentifier, string clientIdentity, string sharedAccessSignature, string storageAccountName, string storageAccountAccessKey)
+            public void Connect(string connectionString, string eventHubName, string serviceUri, string tenantIdentity, string resourceIdentifier, string clientIdentity, string sharedAccessSignature, string storageAccountName, string storageAccountAccessKey, bool useWebSockets, string webSocketsProxyAddress, string endPointAddress)
             {
                 EventHubName = eventHubName;
 
+                Azure.Messaging.EventHubs.Producer.EventHubProducerClientOptions options = default;
+                if (useWebSockets || !string.IsNullOrEmpty(webSocketsProxyAddress) || !string.IsNullOrEmpty(endPointAddress))
+                {
+                    options = new Azure.Messaging.EventHubs.Producer.EventHubProducerClientOptions();
+                    options.ConnectionOptions.TransportType = useWebSockets ? EventHubsTransportType.AmqpWebSockets : options.ConnectionOptions.TransportType;
+                    options.ConnectionOptions.Proxy = !string.IsNullOrEmpty(webSocketsProxyAddress) ? new System.Net.WebProxy(webSocketsProxyAddress, true) : options.ConnectionOptions.Proxy;
+                    options.ConnectionOptions.CustomEndpointAddress = !string.IsNullOrEmpty(endPointAddress) ? new Uri(endPointAddress) : options.ConnectionOptions.CustomEndpointAddress;
+                }
+
                 if (string.IsNullOrWhiteSpace(serviceUri))
                 {
                     if (string.IsNullOrWhiteSpace(eventHubName))
                     {
-                        _client = new Azure.Messaging.EventHubs.Producer.EventHubProducerClient(connectionString);
+                        _client = new Azure.Messaging.EventHubs.Producer.EventHubProducerClient(connectionString, options);
                     }
                     else
                     {
-                        _client = new Azure.Messaging.EventHubs.Producer.EventHubProducerClient(connectionString, eventHubName);
+                        _client = new Azure.Messaging.EventHubs.Producer.EventHubProducerClient(connectionString, eventHubName, options);
                     }
                 }
                 else if (!string.IsNullOrWhiteSpace(sharedAccessSignature))
                 {
-                    _client = new Azure.Messaging.EventHubs.Producer.EventHubProducerClient(serviceUri, eventHubName, new Azure.AzureSasCredential(sharedAccessSignature));
+                    _client = new Azure.Messaging.EventHubs.Producer.EventHubProducerClient(serviceUri, eventHubName, new Azure.AzureSasCredential(sharedAccessSignature), options);
                 }
                 else if (!string.IsNullOrWhiteSpace(storageAccountName))
                 {
-                    _client = new Azure.Messaging.EventHubs.Producer.EventHubProducerClient(serviceUri, eventHubName, new Azure.AzureNamedKeyCredential(storageAccountName, storageAccountAccessKey));
+                    _client = new Azure.Messaging.EventHubs.Producer.EventHubProducerClient(serviceUri, eventHubName, new Azure.AzureNamedKeyCredential(storageAccountName, storageAccountAccessKey), options);
                 }
                 else
                 {
                     var tokenCredentials = AzureCredentialHelpers.CreateTokenCredentials(clientIdentity, tenantIdentity, resourceIdentifier);
-                    _client = new Azure.Messaging.EventHubs.Producer.EventHubProducerClient(serviceUri, eventHubName, tokenCredentials);
+                    _client = new Azure.Messaging.EventHubs.Producer.EventHubProducerClient(serviceUri, eventHubName, tokenCredentials, options);
                 }
             }
 

src/NLog.Extensions.AzureEventHub/IEventHubService.cs

@@ -8,7 +8,7 @@ namespace NLog.Extensions.AzureStorage
     internal interface IEventHubService
     {
         string EventHubName { get; }
-        void Connect(string connectionString, string eventHubName, string serviceUri, string tenantIdentity, string resourceIdentifier, string clientIdentity, string sharedAccessSignature, string storageAccountName, string storageAccountAccessKey);
+        void Connect(string connectionString, string eventHubName, string serviceUri, string tenantIdentity, string resourceIdentifier, string clientIdentity, string sharedAccessSignature, string storageAccountName, string storageAccountAccessKey, bool useWebSockets, string webSocketsProxyAddress, string endPointAddress);
         Task CloseAsync();
         Task SendAsync(IEnumerable<EventData> eventDataBatch, string partitionKey, CancellationToken cancellationToken);
     }

src/NLog.Extensions.AzureEventHub/NLog.Extensions.AzureEventHub.csproj

@@ -3,9 +3,8 @@
   <PropertyGroup>
     <TargetFrameworks>netstandard2.0</TargetFrameworks>
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
-    <DefineConstants>$(DefineConstants);NETCORE</DefineConstants>
 
-    <Version>4.3.1</Version>
+    <Version>4.4.0</Version>
 
     <Description>NLog EventHubTarget for writing to Azure Event Hubs datastreams</Description>
     <Authors>jdetmar</Authors>
@@ -19,8 +18,9 @@
     <RepositoryUrl>https://github.com/JDetmar/NLog.Extensions.AzureStorage.git</RepositoryUrl>
     <PackageLicenseExpression>MIT</PackageLicenseExpression>
     <PackageReleaseNotes>
-- Updated dependency Azure.Messaging.EventHubs v5.9.3
-- Updated dependency Azure.Identity v1.10.3
+- Added option UseWebSockets (true / false)
+- Added option WebSocketProxyAddress (Uri)
+- Added option CustomEndpointAddress (Uri)
 
 Docs: https://github.com/JDetmar/NLog.Extensions.AzureStorage/blob/master/src/NLog.Extensions.AzureEventHub/README.md
     </PackageReleaseNotes>

src/NLog.Extensions.AzureEventHub/README.md

@@ -52,6 +52,12 @@ _messageId_ - EventData MessageId. [Layout](https://github.com/NLog/NLog/wiki/La
 
 _correlationId_ - EventData Correlationid. [Layout](https://github.com/NLog/NLog/wiki/Layouts)
 
+_useWebSockets_ - Enable AmqpWebSockets. Ex. true/false (optional)
+
+_webSocketProxyAddress_ - Custom WebProxy address for WebSockets (optional)
+
+_customEndpointAddress_ - Custom endpoint address that can be used when establishing the connection (optional)
+
 _serviceUri_ - Alternative to ConnectionString, where Managed Identiy is applied from DefaultAzureCredential for User delegation SAS.
 
 _tenantIdentity_ - Alternative to ConnectionString. Used together with ServiceUri. Input for DefaultAzureCredential.

src/NLog.Extensions.AzureServiceBus/ICloudServiceBus.cs

@@ -10,7 +10,7 @@ internal interface ICloudServiceBus
     {
         string EntityPath { get; }
         TimeSpan? DefaultTimeToLive { get; }
-        void Connect(string connectionString, string queueOrTopicName, string serviceUri, string tenantIdentity, string resourceIdentifier, string clientIdentity, string sharedAccessSignature, string storageAccountName, string storageAccountAccessKey, TimeSpan? timeToLive);
+        void Connect(string connectionString, string queueOrTopicName, string serviceUri, string tenantIdentity, string resourceIdentifier, string clientIdentity, string sharedAccessSignature, string storageAccountName, string storageAccountAccessKey, bool useWebSockets, string webSocketProxyAddress, string endPointAddress, TimeSpan? timeToLive);
         Task SendAsync(IEnumerable<ServiceBusMessage> messages, CancellationToken cancellationToken);
         Task CloseAsync();
     }

src/NLog.Extensions.AzureServiceBus/NLog.Extensions.AzureServiceBus.csproj

@@ -4,7 +4,7 @@
     <TargetFramework>netstandard2.0</TargetFramework>
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
 
-    <Version>4.3.1</Version>
+    <Version>4.4.0</Version>
 
     <Description>NLog ServiceBusTarget for writing to Azure Service Bus Topic or Queue</Description>
     <Authors>jdetmar</Authors>
@@ -18,8 +18,10 @@
     <RepositoryUrl>https://github.com/JDetmar/NLog.Extensions.AzureStorage.git</RepositoryUrl>
     <PackageLicenseExpression>MIT</PackageLicenseExpression>
     <PackageReleaseNotes>
-- Updated dependency Azure.Messaging.ServiceBus v7.16.1
-- Updated dependency Azure.Identity v1.10.3
+- Added option UseWebSockets (true / false)
+- Added option WebSocketProxyAddress (Uri)
+- Added option CustomEndpointAddress (Uri)
+- Updated dependency Azure.Messaging.ServiceBus v7.16.2
 
 Docs: https://github.com/JDetmar/NLog.Extensions.AzureStorage/blob/master/src/NLog.Extensions.AzureServiceBus/README.md
     </PackageReleaseNotes>

src/NLog.Extensions.AzureServiceBus/README.md

@@ -65,6 +65,12 @@ _timeToLiveSeconds_ - Default Time-To-Live (TTL) for ServiceBus messages in seco
 
 _timeToLiveDays_ - Default Time-To-Live (TTL) for ServiceBus messages in days (Optional)
 
+_useWebSockets_ - Enable AmqpWebSockets. Ex. true/false (optional)
+
+_webSocketProxyAddress_ - Custom WebProxy address for WebSockets (optional)
+
+_customEndpointAddress_ - Custom endpoint address that can be used when establishing the connection (optional)
+
 _serviceUri_ - Alternative to ConnectionString, where Managed Identiy is applied from DefaultAzureCredential for User delegation SAS.
 
 _tenantIdentity_ - Alternative to ConnectionString. Used together with ServiceUri. Input for DefaultAzureCredential.

src/NLog.Extensions.AzureServiceBus/ServiceBusTarget.cs

@@ -141,6 +141,21 @@ public sealed class ServiceBusTarget : AsyncTaskTarget
         /// </summary>
         public Layout AccessKey { get; set; }
 
+        /// <summary>
+        /// The connection uses the AMQP protocol over web sockets. See also <see cref="ServiceBusTransportType.AmqpWebSockets"/>
+        /// </summary>
+        public Layout UseWebSockets { get; set; }
+
+        /// <summary>
+        /// The proxy to use for communication over web sockets.
+        /// </summary>
+        public Layout WebSocketProxyAddress { get; set; }
+
+        /// <summary>
+        /// Custom endpoint address that can be used when establishing the connection.
+        /// </summary>
+        public Layout CustomEndpointAddress { get; set; }
+
         /// <summary>
         /// Gets a list of user properties (aka custom application properties) to add to the AMQP message
         /// </summary>
@@ -187,6 +202,9 @@ protected override void InitializeTarget()
             string storageAccountName = string.Empty;
             string storageAccountAccessKey = string.Empty;
             string queueOrTopicName = string.Empty;
+            string useWebSockets = string.Empty;
+            string webSocketProxyAddress = string.Empty;
+            string customEndPointAddress = string.Empty;
 
             var defaultLogEvent = LogEventInfo.CreateNullEvent();
 
@@ -212,13 +230,21 @@ protected override void InitializeTarget()
                     storageAccountAccessKey = AccessKey?.Render(defaultLogEvent);
                 }
 
+                useWebSockets = UseWebSockets?.Render(defaultLogEvent) ?? string.Empty;
+                if (!string.IsNullOrEmpty(useWebSockets) && (string.Equals(useWebSockets.Trim(), bool.TrueString, StringComparison.OrdinalIgnoreCase) || string.Equals(useWebSockets.Trim(), "1", StringComparison.OrdinalIgnoreCase)))
+                {
+                    useWebSockets = bool.TrueString;
+                }
+                webSocketProxyAddress = WebSocketProxyAddress?.Render(defaultLogEvent) ?? string.Empty;
+                customEndPointAddress = CustomEndpointAddress?.Render(defaultLogEvent) ?? string.Empty;
+
                 var timeToLive = RenderDefaultTimeToLive();
                 if (timeToLive <= TimeSpan.Zero)
                 {
                     timeToLive = default(TimeSpan?);
                 }
 
-                _cloudServiceBus.Connect(connectionString, queueOrTopicName, serviceUri, tenantIdentity, resourceIdentifier, clientIdentity, sharedAccessSignature, storageAccountName, storageAccountAccessKey, timeToLive);
+                _cloudServiceBus.Connect(connectionString, queueOrTopicName, serviceUri, tenantIdentity, resourceIdentifier, clientIdentity, sharedAccessSignature, storageAccountName, storageAccountAccessKey, bool.TrueString == useWebSockets, webSocketProxyAddress, customEndPointAddress, timeToLive);
                 InternalLogger.Debug("AzureServiceBusTarget(Name={0}): Initialized", Name);
             }
             catch (Exception ex)
@@ -571,27 +597,36 @@ private sealed class CloudServiceBus : ICloudServiceBus
 
             public string EntityPath { get; private set; }
 
-            public void Connect(string connectionString, string queueOrTopicName, string serviceUri, string tenantIdentity, string resourceIdentifier, string clientIdentity, string sharedAccessSignature, string storageAccountName, string storageAccountAccessKey, TimeSpan? timeToLive)
+            public void Connect(string connectionString, string queueOrTopicName, string serviceUri, string tenantIdentity, string resourceIdentifier, string clientIdentity, string sharedAccessSignature, string storageAccountName, string storageAccountAccessKey, bool useWebSockets, string webSocketProxyAddress, string endPointAddress, TimeSpan? timeToLive)
             {
                 EntityPath = queueOrTopicName;
                 DefaultTimeToLive = timeToLive;
 
+                Azure.Messaging.ServiceBus.ServiceBusClientOptions options = default;
+                if (useWebSockets || !string.IsNullOrEmpty(webSocketProxyAddress) || !string.IsNullOrEmpty(endPointAddress))
+                {
+                    options = new Azure.Messaging.ServiceBus.ServiceBusClientOptions();
+                    options.TransportType = useWebSockets ? ServiceBusTransportType.AmqpWebSockets : options.TransportType;
+                    options.WebProxy = !string.IsNullOrEmpty(webSocketProxyAddress) ? new System.Net.WebProxy(webSocketProxyAddress, true) : options.WebProxy;
+                    options.CustomEndpointAddress = !string.IsNullOrEmpty(endPointAddress) ? new Uri(endPointAddress) : options.CustomEndpointAddress;
+                }
+
                 if (string.IsNullOrEmpty(serviceUri))
                 {
-                    _client = new ServiceBusClient(connectionString);
+                    _client = new ServiceBusClient(connectionString, options);
                 }
                 else if (!string.IsNullOrWhiteSpace(sharedAccessSignature))
                 {
-                    _client = new ServiceBusClient(serviceUri, new Azure.AzureSasCredential(sharedAccessSignature));
+                    _client = new ServiceBusClient(serviceUri, new Azure.AzureSasCredential(sharedAccessSignature), options);
                 }
                 else if (!string.IsNullOrWhiteSpace(storageAccountName))
                 {
-                    _client = new ServiceBusClient(serviceUri, new Azure.AzureNamedKeyCredential(storageAccountName, storageAccountAccessKey));
+                    _client = new ServiceBusClient(serviceUri, new Azure.AzureNamedKeyCredential(storageAccountName, storageAccountAccessKey), options);
                 }
                 else
                 {
                     var tokenCredentials = AzureCredentialHelpers.CreateTokenCredentials(clientIdentity, tenantIdentity, resourceIdentifier);
-                    _client = new ServiceBusClient(serviceUri, tokenCredentials);
+                    _client = new ServiceBusClient(serviceUri, tokenCredentials, options);
                 }
 
                 _sender = _client.CreateSender(queueOrTopicName);

test/NLog.Extensions.AzureEventHub.Tests/EventHubServiceMock.cs

@@ -23,7 +23,7 @@ public async Task CloseAsync()
                 EventDataSent.Clear();
         }
 
-        public void Connect(string connectionString, string eventHubName, string serviceUri, string tenantIdentity, string resourceIdentifier, string clientIdentity, string sharedAccessSignature, string storageAccountName, string storageAccountAccessKey)
+        public void Connect(string connectionString, string eventHubName, string serviceUri, string tenantIdentity, string resourceIdentifier, string clientIdentity, string sharedAccessSignature, string storageAccountName, string storageAccountAccessKey, bool useWebSockets, string webSocketsProxyAddress, string endPointAddress)
         {
             ConnectionString = connectionString;
             EventHubName = eventHubName;

test/NLog.Extensions.AzureServiceBus.Tests/ServiceBusMock.cs

@@ -31,7 +31,7 @@ public string PeekLastMessageBody()
             return null;
         }
 
-        public void Connect(string connectionString, string queueOrTopicName, string serviceUri, string tenantIdentity, string resourceIdentifier, string clientIdentity, string sharedAccessSignature, string storageAccountName, string storageAccountAccessKey, TimeSpan? timeToLive)
+        public void Connect(string connectionString, string queueOrTopicName, string serviceUri, string tenantIdentity, string resourceIdentifier, string clientIdentity, string sharedAccessSignature, string storageAccountName, string storageAccountAccessKey, bool useWebSockets, string webProxy, string endPointAddress, TimeSpan? timeToLive)
         {
             ConnectionString = connectionString;
             EntityPath = queueOrTopicName;