Add extension library for WebSocket4Net usage.

Source/MQTTnet.Extensions.WebSocket4Net/MQTTnet.Extensions.WebSocket4Net.csproj

@@ -0,0 +1,34 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFrameworks>netstandard1.3;netstandard2.0</TargetFrameworks>
+    <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">$(TargetFrameworks);net452;net461</TargetFrameworks>
+    <TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' AND '$(MSBuildRuntimeType)' != 'Core'">$(TargetFrameworks);uap10.0</TargetFrameworks>
+    <Product />
+    <Company />
+    <Authors />
+    <PackageId />
+  </PropertyGroup>
+
+  <PropertyGroup Condition="'$(TargetFramework)' == 'uap10.0'">
+    <CopyLocalLockFileAssemblies>false</CopyLocalLockFileAssemblies>
+    <NugetTargetMoniker>UAP,Version=v10.0</NugetTargetMoniker>
+    <TargetPlatformIdentifier>UAP</TargetPlatformIdentifier>
+    <TargetPlatformVersion>10.0.17134.0</TargetPlatformVersion>
+    <TargetPlatformMinVersion>10.0.10240.0</TargetPlatformMinVersion>
+    <TargetFrameworkIdentifier>.NETCore</TargetFrameworkIdentifier>
+    <TargetFrameworkVersion>v5.0</TargetFrameworkVersion>
+    <DefineConstants>$(DefineConstants);WINDOWS_UWP</DefineConstants>
+    <DefaultLanguage>en</DefaultLanguage>
+    <LanguageTargets>$(MSBuildExtensionsPath)\Microsoft\WindowsXaml\v$(VisualStudioVersion)\Microsoft.Windows.UI.Xaml.CSharp.targets</LanguageTargets>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="WebSocket4Net" Version="0.15.2" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\MQTTnet\MQTTnet.csproj" />
+  </ItemGroup>
+
+</Project>

Source/MQTTnet.Extensions.WebSocket4Net/MqttFactoryExtensions.cs

@@ -0,0 +1,14 @@
+using System;
+
+namespace MQTTnet.Extensions.WebSocket4Net
+{
+    public static class MqttFactoryExtensions
+    {
+        public static IMqttFactory UseWebSocket4Net(this IMqttFactory mqttFactory)
+        {
+            if (mqttFactory == null) throw new ArgumentNullException(nameof(mqttFactory));
+
+            return mqttFactory.UseClientAdapterFactory(new WebSocket4NetMqttClientAdapterFactory());
+        }
+    }
+}

Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs

@@ -0,0 +1,231 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Net;
+using System.Security.Authentication;
+using System.Security.Cryptography.X509Certificates;
+using System.Threading;
+using System.Threading.Tasks;
+using MQTTnet.Channel;
+using MQTTnet.Client.Options;
+using MQTTnet.Exceptions;
+using MQTTnet.Internal;
+using SuperSocket.ClientEngine;
+using WebSocket4Net;
+
+namespace MQTTnet.Extensions.WebSocket4Net
+{
+    public class WebSocket4NetMqttChannel : IMqttChannel
+    {
+        private readonly BlockingCollection<byte> _receiveBuffer = new BlockingCollection<byte>();
+
+        private readonly IMqttClientOptions _clientOptions;
+        private readonly MqttClientWebSocketOptions _webSocketOptions;
+
+        private WebSocket _webSocket;
+
+        public WebSocket4NetMqttChannel(IMqttClientOptions clientOptions, MqttClientWebSocketOptions webSocketOptions)
+        {
+            _clientOptions = clientOptions ?? throw new ArgumentNullException(nameof(clientOptions));
+            _webSocketOptions = webSocketOptions ?? throw new ArgumentNullException(nameof(webSocketOptions));
+        }
+
+        public string Endpoint => _webSocketOptions.Uri;
+
+        public bool IsSecureConnection { get; private set; }
+
+        public async Task ConnectAsync(CancellationToken cancellationToken)
+        {
+            var uri = _webSocketOptions.Uri;
+            if (!uri.StartsWith("ws://", StringComparison.OrdinalIgnoreCase) && !uri.StartsWith("wss://", StringComparison.OrdinalIgnoreCase))
+            {
+                if (_webSocketOptions.TlsOptions?.UseTls == false)
+                {
+                    uri = "ws://" + uri;
+                }
+                else
+                {
+                    uri = "wss://" + uri;
+                }
+            }
+
+            var sslProtocols = _webSocketOptions?.TlsOptions?.SslProtocol ?? SslProtocols.None;
+            var subProtocol = _webSocketOptions.SubProtocols.FirstOrDefault() ?? string.Empty;
+
+            var cookies = new List<KeyValuePair<string, string>>();
+            if (_webSocketOptions.CookieContainer != null)
+            {
+                throw new NotSupportedException("Cookies are not supported.");
+            }
+
+            List<KeyValuePair<string, string>> customHeaders = null;
+            if (_webSocketOptions.RequestHeaders != null)
+            {
+                customHeaders = _webSocketOptions.RequestHeaders.Select(i => new KeyValuePair<string, string>(i.Key, i.Value)).ToList();
+            }
+
+            EndPoint proxy = null;
+            if (_webSocketOptions.ProxyOptions != null)
+            {
+                throw new NotSupportedException("Proxies are not supported.");
+            }
+
+            // The user agent can be empty always because it is just added to the custom headers as "User-Agent".
+            var userAgent = string.Empty;
+
+            var origin = string.Empty;
+            var webSocketVersion = WebSocketVersion.None;
+            var receiveBufferSize = 0;
+
+            var certificates = new X509CertificateCollection();
+            if (_webSocketOptions?.TlsOptions?.Certificates != null)
+            {
+                foreach (var certificate in _webSocketOptions.TlsOptions.Certificates)
+                {
+                    certificates.Add(new X509Certificate(certificate));
+                }
+            }
+
+            _webSocket = new WebSocket(uri, subProtocol, cookies, customHeaders, userAgent, origin, webSocketVersion, proxy, sslProtocols, receiveBufferSize)
+            {
+                NoDelay = true,
+                Security =
+                {
+                    AllowUnstrustedCertificate = _webSocketOptions?.TlsOptions?.AllowUntrustedCertificates == true,
+                    AllowCertificateChainErrors = _webSocketOptions?.TlsOptions?.IgnoreCertificateChainErrors == true,
+                    Certificates = certificates
+                }
+            };
+
+            await ConnectInternalAsync(cancellationToken).ConfigureAwait(false);
+            
+            IsSecureConnection = uri.StartsWith("wss://", StringComparison.OrdinalIgnoreCase);
+        }
+
+        public Task DisconnectAsync(CancellationToken cancellationToken)
+        {
+            if (_webSocket != null && _webSocket.State == WebSocketState.Open)
+            {
+                _webSocket.Close();
+            }
+            
+            return Task.FromResult(0);
+        }
+
+        public Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        {
+            var readBytes = 0;
+            while (count > 0 && !cancellationToken.IsCancellationRequested)
+            {
+                if (!_receiveBuffer.TryTake(out var @byte))
+                {
+                    if (readBytes == 0)
+                    {
+                        // Block until at least one byte was received.
+                        @byte = _receiveBuffer.Take(cancellationToken);
+                    }
+                    else
+                    {
+                        return Task.FromResult(readBytes);
+                    }
+                }
+
+                buffer[offset] = @byte;
+                offset++;
+                count--;
+                readBytes++;
+            }
+
+            return Task.FromResult(readBytes);
+        }
+
+        public Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        {
+            _webSocket.Send(buffer, offset, count);
+            return Task.FromResult(0);
+        }
+
+        public void Dispose()
+        {
+            if (_webSocket == null)
+            {
+                return;
+            }
+
+            _webSocket.DataReceived -= OnDataReceived;
+            _webSocket.Error -= OnError;
+            _webSocket.Dispose();
+            _webSocket = null;
+        }
+
+        private void OnError(object sender, ErrorEventArgs e)
+        {
+            System.Diagnostics.Debug.Write(e.Exception.ToString());
+        }
+
+        private void OnDataReceived(object sender, DataReceivedEventArgs e)
+        {
+            foreach (var @byte in e.Data)
+            {
+                _receiveBuffer.Add(@byte);
+            }
+        }
+
+        private async Task ConnectInternalAsync(CancellationToken cancellationToken)
+        {
+            _webSocket.Error += OnError;
+            _webSocket.DataReceived += OnDataReceived;
+
+            var taskCompletionSource = new TaskCompletionSource<Exception>();
+
+            void ErrorHandler(object sender, ErrorEventArgs e)
+            {
+                taskCompletionSource.TrySetResult(e.Exception);
+            }
+
+            void SuccessHandler(object sender, EventArgs e)
+            {
+                taskCompletionSource.TrySetResult(null);
+            }
+
+            try
+            {
+                _webSocket.Opened += SuccessHandler;
+                _webSocket.Error += ErrorHandler;
+
+                _webSocket.Open();
+
+                var exception = await MqttTaskTimeout.WaitAsync(c =>
+                {
+                    c.Register(() => taskCompletionSource.TrySetCanceled());
+                    return taskCompletionSource.Task;
+                }, _clientOptions.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
+
+                if (exception != null)
+                {
+                    if (exception is AuthenticationException authenticationException)
+                    {
+                        throw new MqttCommunicationException(authenticationException.InnerException);
+                    }
+
+                    if (exception is OperationCanceledException)
+                    {
+                        throw new MqttCommunicationTimedOutException();
+                    }
+
+                    throw new MqttCommunicationException(exception);
+                }
+            }
+            catch (OperationCanceledException)
+            {
+                throw new MqttCommunicationTimedOutException();
+            }
+            finally
+            {
+                _webSocket.Opened -= SuccessHandler;
+                _webSocket.Error -= ErrorHandler;
+            }
+        }
+    }
+}

Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs

@@ -0,0 +1,36 @@
+using System;
+using MQTTnet.Adapter;
+using MQTTnet.Client.Options;
+using MQTTnet.Diagnostics;
+using MQTTnet.Formatter;
+using MQTTnet.Implementations;
+
+namespace MQTTnet.Extensions.WebSocket4Net
+{
+    public class WebSocket4NetMqttClientAdapterFactory : IMqttClientAdapterFactory
+    {
+        public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger)
+        {
+            if (options == null) throw new ArgumentNullException(nameof(options));
+            if (logger == null) throw new ArgumentNullException(nameof(logger));
+
+            switch (options.ChannelOptions)
+            {
+                case MqttClientTcpOptions _:
+                {
+                    return new MqttChannelAdapter(new MqttTcpChannel(options), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger);
+                }
+
+                case MqttClientWebSocketOptions webSocketOptions:
+                {
+                    return new MqttChannelAdapter(new WebSocket4NetMqttChannel(options, webSocketOptions), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger);
+                }
+
+                default:
+                {
+                    throw new NotSupportedException();
+                }
+            }
+        }
+    }
+}

Source/MQTTnet/Client/IMqttClientFactory.cs

@@ -5,6 +5,8 @@ namespace MQTTnet.Client
 {
     public interface IMqttClientFactory
     {
+        IMqttFactory UseClientAdapterFactory(IMqttClientAdapterFactory clientAdapterFactory);
+
         IMqttClient CreateMqttClient();
 
         IMqttClient CreateMqttClient(IMqttNetLogger logger);

Source/MQTTnet/MqttFactory.cs

@@ -10,6 +10,8 @@ namespace MQTTnet
 {
     public class MqttFactory : IMqttFactory
     {
+        private IMqttClientAdapterFactory _clientAdapterFactory = new MqttClientAdapterFactory();
+
         public MqttFactory() : this(new MqttNetLogger())
         {
         }
@@ -21,6 +23,12 @@ namespace MQTTnet
 
         public IMqttNetLogger DefaultLogger { get; }
 
+        public IMqttFactory UseClientAdapterFactory(IMqttClientAdapterFactory clientAdapterFactory)
+        {
+            _clientAdapterFactory = clientAdapterFactory ?? throw new ArgumentNullException(nameof(clientAdapterFactory));
+            return this;
+        }
+
         public IMqttClient CreateMqttClient()
         {
             return CreateMqttClient(DefaultLogger);
@@ -30,22 +38,22 @@ namespace MQTTnet
         {
             if (logger == null) throw new ArgumentNullException(nameof(logger));
 
-            return new MqttClient(new MqttClientAdapterFactory(), logger);
+            return new MqttClient(_clientAdapterFactory, logger);
         }
 
-        public IMqttClient CreateMqttClient(IMqttClientAdapterFactory adapterFactory)
+        public IMqttClient CreateMqttClient(IMqttClientAdapterFactory clientAdapterFactory)
         {
-            if (adapterFactory == null) throw new ArgumentNullException(nameof(adapterFactory));
+            if (clientAdapterFactory == null) throw new ArgumentNullException(nameof(clientAdapterFactory));
 
-            return new MqttClient(adapterFactory, DefaultLogger);
+            return new MqttClient(clientAdapterFactory, DefaultLogger);
         }
 
-        public IMqttClient CreateMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory adapterFactory)
+        public IMqttClient CreateMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory clientAdapterFactory)
         {
             if (logger == null) throw new ArgumentNullException(nameof(logger));
-            if (adapterFactory == null) throw new ArgumentNullException(nameof(adapterFactory));
+            if (clientAdapterFactory == null) throw new ArgumentNullException(nameof(clientAdapterFactory));
 
-            return new MqttClient(adapterFactory, logger);
+            return new MqttClient(clientAdapterFactory, logger);
         }
 
         public IMqttServer CreateMqttServer()
@@ -60,19 +68,19 @@ namespace MQTTnet
             return CreateMqttServer(new List<IMqttServerAdapter> { new MqttTcpServerAdapter(logger.CreateChildLogger()) }, logger);
         }
 
-        public IMqttServer CreateMqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
+        public IMqttServer CreateMqttServer(IEnumerable<IMqttServerAdapter> serverAdapters, IMqttNetLogger logger)
         {
-            if (adapters == null) throw new ArgumentNullException(nameof(adapters));
+            if (serverAdapters == null) throw new ArgumentNullException(nameof(serverAdapters));
             if (logger == null) throw new ArgumentNullException(nameof(logger));
 
-            return new MqttServer(adapters, logger.CreateChildLogger());
+            return new MqttServer(serverAdapters, logger.CreateChildLogger());
         }
 
-        public IMqttServer CreateMqttServer(IEnumerable<IMqttServerAdapter> adapters)
+        public IMqttServer CreateMqttServer(IEnumerable<IMqttServerAdapter> serverAdapters)
         {
-            if (adapters == null) throw new ArgumentNullException(nameof(adapters));
+            if (serverAdapters == null) throw new ArgumentNullException(nameof(serverAdapters));
             
-            return new MqttServer(adapters, DefaultLogger.CreateChildLogger());
+            return new MqttServer(serverAdapters, DefaultLogger.CreateChildLogger());
         }
     }
 }