Added support for websockets

2026-10-19 11:03:07 +08:00 · 2017-09-01 09:39:59 +02:00
parent 96b92aabae
commit 3a1e32f5aa
10 changed files with 361 additions and 3 deletions
@@ -0,0 +1,119 @@
+using MQTTnet.Core.Channel;
+using MQTTnet.Core.Client;
+using MQTTnet.Core.Exceptions;
+using System;
+using System.Net.WebSockets;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Implementations
+{
+    public sealed class MqttWebSocketsChannel : IMqttCommunicationChannel, IDisposable
+    {
+        private ClientWebSocket _webSocket = null;
+        private const int BufferSize = 4096;
+        private const int BufferAmplifier = 20;
+        private byte[] WebSocketBuffer = new byte[BufferSize * BufferAmplifier];
+        private int WebSocketBufferSize = 0;
+        private int WebSocketBufferOffset = 0;
+
+        public MqttWebSocketsChannel()
+        {
+            _webSocket = new ClientWebSocket();
+        }
+
+        public async Task ConnectAsync(MqttClientOptions options)
+        {
+            _webSocket = null;
+
+            try
+            {
+                _webSocket = new ClientWebSocket();
+
+                await _webSocket.ConnectAsync(new Uri(options.Server), CancellationToken.None);
+            }
+            catch (WebSocketException exception)
+            {
+                throw new MqttCommunicationException(exception);
+            }
+            finally
+            {
+            }
+        }
+
+        public async Task DisconnectAsync()
+        {
+            await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None);
+        }
+
+        public void Dispose()
+        {
+            if (_webSocket != null)
+                _webSocket.Dispose();
+        }
+
+        public Task ReadAsync(byte[] buffer)
+        {
+            return Task.WhenAll(ReadToBufferAsync(buffer));
+        }
+
+        private async Task ReadToBufferAsync(byte[] buffer)
+        {
+            var temporaryBuffer = new byte[BufferSize];
+            var offset = 0;
+
+            while (_webSocket.State == WebSocketState.Open)
+            {
+                if (WebSocketBufferSize == 0)
+                {
+                    WebSocketBufferOffset = 0;
+
+                    WebSocketReceiveResult response;
+                    do
+                    {
+                        response =
+                            await _webSocket.ReceiveAsync(new ArraySegment<byte>(temporaryBuffer), CancellationToken.None);
+
+                        temporaryBuffer.CopyTo(WebSocketBuffer, offset);
+                        offset += response.Count;
+                        temporaryBuffer = new byte[BufferSize];
+                    } while (!response.EndOfMessage);
+
+                    WebSocketBufferSize = response.Count;
+                    if (response.MessageType == WebSocketMessageType.Close)
+                    {
+                        await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None);
+                    }
+
+                    Buffer.BlockCopy(WebSocketBuffer, 0, buffer, 0, buffer.Length);
+                    WebSocketBufferSize -= buffer.Length;
+                    WebSocketBufferOffset += buffer.Length;
+                }
+                else
+                {
+                    Buffer.BlockCopy(WebSocketBuffer, WebSocketBufferOffset, buffer, 0, buffer.Length);
+                    WebSocketBufferSize -= buffer.Length;
+                    WebSocketBufferOffset += buffer.Length;
+                }
+
+                return;
+            }
+        }
+
+        public Task WriteAsync(byte[] buffer)
+        {
+            if (buffer == null) throw new ArgumentNullException(nameof(buffer));
+
+            var writeBuffer = System.Text.Encoding.ASCII.GetString(buffer);
+            try
+            {
+                return _webSocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Binary, true,
+                 CancellationToken.None);
+            }
+            catch (WebSocketException exception)
+            {
+                throw new MqttCommunicationException(exception);
+            }
+        }
+    }
+}
@@ -24,6 +24,8 @@

  <ItemGroup>
    <PackageReference Include="System.Net.Security" Version="4.3.1" />
+    <PackageReference Include="System.Net.WebSockets" Version="4.3.0" />
+    <PackageReference Include="System.Net.WebSockets.Client" Version="4.3.1" />
    <PackageReference Include="System.Threading.Thread" Version="4.3.0" />
  </ItemGroup>

@@ -3,6 +3,7 @@ using MQTTnet.Core.Adapter;
 using MQTTnet.Core.Client;
 using MQTTnet.Core.Serializer;
 using MQTTnet.Implementations;
+using MQTTnet.Core.Channel;

 namespace MQTTnet
 {
@@ -12,7 +13,23 @@ namespace MQTTnet
        {
            if (options == null) throw new ArgumentNullException(nameof(options));

-            return new MqttClient(options, new MqttChannelCommunicationAdapter(new MqttTcpChannel(), new MqttPacketSerializer()));
+            return new MqttClient(options, new MqttChannelCommunicationAdapter(GetMqttCommunicationChannel(options), new MqttPacketSerializer()));
+        }
+
+        private IMqttCommunicationChannel GetMqttCommunicationChannel(MqttClientOptions options)
+        {
+            switch (options.ConnectionType)
+            {
+                case ConnectionTypes.TCP:
+                case ConnectionTypes.TLS:
+                    return new MqttTcpChannel();
+                case ConnectionTypes.WS:
+                case ConnectionTypes.WSS:
+                    return new MqttWebSocketsChannel();
+
+                default:
+                    return null;
+            }
        }
    }
 }