diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index b6052a62..4cf58e59 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1 +1,2 @@ -* [Client] Keep alive mechanism now uses the configured timeout value from the options (thanks to @Stannieman, #1495). \ No newline at end of file +* [Client] Keep alive mechanism now uses the configured timeout value from the options (thanks to @Stannieman, #1495). +* [Core] MQTT Packets being sent over web socket transport are now setting the web socket frame boundaries correctly (#1499). \ No newline at end of file diff --git a/Source/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs b/Source/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs index c54d2ee2..5c3a1d1f 100644 --- a/Source/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs +++ b/Source/MQTTnet.Benchmarks/MqttTcpChannelBenchmark.cs @@ -2,6 +2,7 @@ // The .NET Foundation licenses this file to you under the MIT license. // See the LICENSE file in the project root for more information. +using System; using BenchmarkDotNet.Attributes; using MQTTnet.Channel; using MQTTnet.Implementations; @@ -70,8 +71,8 @@ namespace MQTTnet.Benchmarks while (read < expected) { - var readresult = await _clientChannel.ReadAsync(new byte[size], 0, size, CancellationToken.None).ConfigureAwait(false); - read += readresult; + var readResult = await _clientChannel.ReadAsync(new byte[size], 0, size, CancellationToken.None).ConfigureAwait(false); + read += readResult; } } @@ -79,9 +80,11 @@ namespace MQTTnet.Benchmarks { await Task.Yield(); + var buffer = new ArraySegment(new byte[size]); + for (var i = 0; i < iterations; i++) { - await _serverChannel.WriteAsync(new byte[size], 0, size, CancellationToken.None).ConfigureAwait(false); + await _serverChannel.WriteAsync(buffer, true, CancellationToken.None).ConfigureAwait(false); } } } diff --git a/Source/MQTTnet.Benchmarks/SerializerBenchmark.cs b/Source/MQTTnet.Benchmarks/SerializerBenchmark.cs index ba52d4c1..eca6771a 100644 --- a/Source/MQTTnet.Benchmarks/SerializerBenchmark.cs +++ b/Source/MQTTnet.Benchmarks/SerializerBenchmark.cs @@ -104,7 +104,7 @@ namespace MQTTnet.Benchmarks return Task.FromResult(count); } - public Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + public Task WriteAsync(ArraySegment buffer, bool isEndOfPacket, CancellationToken cancellationToken) { throw new NotSupportedException(); } diff --git a/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs b/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs index 47e78eec..bbbdc81f 100644 --- a/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs +++ b/Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs @@ -154,9 +154,9 @@ namespace MQTTnet.Extensions.WebSocket4Net return Task.FromResult(readBytes); } - public Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + public Task WriteAsync(ArraySegment buffer, bool isEndOfPacket, CancellationToken cancellationToken) { - _webSocket.Send(buffer, offset, count); + _webSocket.Send(buffer.Array, buffer.Offset, buffer.Count); return Task.FromResult(0); } diff --git a/Source/MQTTnet.Tests/Mockups/MemoryMqttChannel.cs b/Source/MQTTnet.Tests/Mockups/MemoryMqttChannel.cs index 0dae51f3..7bdf97de 100644 --- a/Source/MQTTnet.Tests/Mockups/MemoryMqttChannel.cs +++ b/Source/MQTTnet.Tests/Mockups/MemoryMqttChannel.cs @@ -2,6 +2,7 @@ // The .NET Foundation licenses this file to you under the MIT license. // See the LICENSE file in the project root for more information. +using System; using System.IO; using System.Security.Cryptography.X509Certificates; using System.Threading; @@ -45,9 +46,9 @@ namespace MQTTnet.Tests.Mockups return _stream.ReadAsync(buffer, offset, count, cancellationToken); } - public Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + public Task WriteAsync(ArraySegment buffer, bool isEndOfPacket, CancellationToken cancellationToken) { - return _stream.WriteAsync(buffer, offset, count, cancellationToken); + return _stream.WriteAsync(buffer.Array, buffer.Offset, buffer.Count, cancellationToken); } public void Dispose() diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index 6b9d8409..5a0bcd6c 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -188,11 +188,14 @@ namespace MQTTnet.Adapter _logger.Verbose("TX ({0} bytes) >>> {1}", packetBuffer.Length, packet); - await _channel.WriteAsync(packetBuffer.Packet.Array, packetBuffer.Packet.Offset, packetBuffer.Packet.Count, cancellationToken).ConfigureAwait(false); - if (packetBuffer.Payload.Count > 0) { - await _channel.WriteAsync(packetBuffer.Payload.Array, packetBuffer.Payload.Offset, packetBuffer.Payload.Count, cancellationToken).ConfigureAwait(false); + await _channel.WriteAsync(packetBuffer.Packet, false, cancellationToken).ConfigureAwait(false); + await _channel.WriteAsync(packetBuffer.Payload, true, cancellationToken).ConfigureAwait(false); + } + else + { + await _channel.WriteAsync(packetBuffer.Packet, true, cancellationToken).ConfigureAwait(false); } Interlocked.Add(ref _bytesReceived, packetBuffer.Length); diff --git a/Source/MQTTnet/Channel/IMqttChannel.cs b/Source/MQTTnet/Channel/IMqttChannel.cs index 5360ce2e..9705c330 100644 --- a/Source/MQTTnet/Channel/IMqttChannel.cs +++ b/Source/MQTTnet/Channel/IMqttChannel.cs @@ -12,13 +12,17 @@ namespace MQTTnet.Channel public interface IMqttChannel : IDisposable { string Endpoint { get; } + bool IsSecureConnection { get; } + X509Certificate2 ClientCertificate { get; } Task ConnectAsync(CancellationToken cancellationToken); + Task DisconnectAsync(CancellationToken cancellationToken); Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken); - Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken); + + Task WriteAsync(ArraySegment buffer, bool isEndOfPacket, CancellationToken cancellationToken); } } diff --git a/Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs index c43948fe..dbca7465 100644 --- a/Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs +++ b/Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs @@ -108,12 +108,12 @@ namespace MQTTnet.Implementations return _readStream.ReadAsync(buffer, offset, count, cancellationToken); } - public Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + public Task WriteAsync(ArraySegment buffer, bool isEndOfPacket, CancellationToken cancellationToken) { // In the write method only the internal buffer will be filled. So here is no // async/await required. The real network transmit is done when calling the // Flush method. - _writeStream.Write(buffer, offset, count); + _writeStream.Write(buffer.Array, buffer.Offset, buffer.Count); return _writeStream.FlushAsync(cancellationToken); } diff --git a/Source/MQTTnet/Implementations/MqttTcpChannel.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.cs index f1391bc1..ad2c58bf 100644 --- a/Source/MQTTnet/Implementations/MqttTcpChannel.cs +++ b/Source/MQTTnet/Implementations/MqttTcpChannel.cs @@ -225,7 +225,7 @@ namespace MQTTnet.Implementations } } - public async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + public async Task WriteAsync(ArraySegment buffer, bool isEndOfPacket, CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); @@ -239,12 +239,12 @@ namespace MQTTnet.Implementations } #if NETCOREAPP3_0_OR_GREATER || NETSTANDARD2_1_OR_GREATER - await stream.WriteAsync(buffer.AsMemory(offset, count), cancellationToken).ConfigureAwait(false); + await stream.WriteAsync(buffer.AsMemory(), cancellationToken).ConfigureAwait(false); #else // Workaround for: https://github.com/dotnet/corefx/issues/24430 using (cancellationToken.Register(_disposeAction)) { - await stream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false); + await stream.WriteAsync(buffer.Array, buffer.Offset, buffer.Count, cancellationToken).ConfigureAwait(false); } #endif } diff --git a/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs b/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs index 3a54843d..e39d545e 100644 --- a/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs +++ b/Source/MQTTnet/Implementations/MqttWebSocketChannel.cs @@ -2,15 +2,15 @@ // The .NET Foundation licenses this file to you under the MIT license. // See the LICENSE file in the project root for more information. -using MQTTnet.Channel; -using MQTTnet.Internal; using System; using System.Net; using System.Net.WebSockets; using System.Security.Cryptography.X509Certificates; using System.Threading; using System.Threading.Tasks; +using MQTTnet.Channel; using MQTTnet.Client; +using MQTTnet.Internal; namespace MQTTnet.Implementations { @@ -35,12 +35,12 @@ namespace MQTTnet.Implementations ClientCertificate = clientCertificate; } + public X509Certificate2 ClientCertificate { get; } + public string Endpoint { get; } public bool IsSecureConnection { get; private set; } - public X509Certificate2 ClientCertificate { get; } - public async Task ConnectAsync(CancellationToken cancellationToken) { var uri = _options.Uri; @@ -89,14 +89,26 @@ namespace MQTTnet.Implementations Cleanup(); } + public void Dispose() + { + Cleanup(); + } + public async Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) { var response = await _webSocket.ReceiveAsync(new ArraySegment(buffer, offset, count), cancellationToken).ConfigureAwait(false); return response.Count; } - public async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + public async Task WriteAsync(ArraySegment buffer, bool isEndOfPacket, CancellationToken cancellationToken) { + cancellationToken.ThrowIfCancellationRequested(); + +#if NET5_0_OR_GREATER + // MQTT Control Packets MUST be sent in WebSocket binary data frames. If any other type of data frame is received the recipient MUST close the Network Connection [MQTT-6.0.0-1]. + // A single WebSocket data frame can contain multiple or partial MQTT Control Packets. The receiver MUST NOT assume that MQTT Control Packets are aligned on WebSocket frame boundaries [MQTT-6.0.0-2]. + await _webSocket.SendAsync(buffer, WebSocketMessageType.Binary, isEndOfPacket, cancellationToken).ConfigureAwait(false); +#else // The lock is required because the client will throw an exception if _SendAsync_ is // called from multiple threads at the same time. But this issue only happens with several // framework versions. @@ -107,86 +119,9 @@ namespace MQTTnet.Implementations using (await _sendLock.WaitAsync(cancellationToken).ConfigureAwait(false)) { - await _webSocket.SendAsync(new ArraySegment(buffer, offset, count), WebSocketMessageType.Binary, true, cancellationToken).ConfigureAwait(false); + await _webSocket.SendAsync(buffer, WebSocketMessageType.Binary, isEndOfPacket, cancellationToken).ConfigureAwait(false); } - } - - public void Dispose() - { - Cleanup(); - } - - void SetupClientWebSocket(ClientWebSocket clientWebSocket) - { - if (_options.ProxyOptions != null) - { - clientWebSocket.Options.Proxy = CreateProxy(); - } - - if (_options.RequestHeaders != null) - { - foreach (var requestHeader in _options.RequestHeaders) - { - clientWebSocket.Options.SetRequestHeader(requestHeader.Key, requestHeader.Value); - } - } - - if (_options.SubProtocols != null) - { - foreach (var subProtocol in _options.SubProtocols) - { - clientWebSocket.Options.AddSubProtocol(subProtocol); - } - } - - if (_options.CookieContainer != null) - { - clientWebSocket.Options.Cookies = _options.CookieContainer; - } - - if (_options.TlsOptions?.UseTls == true && _options.TlsOptions?.Certificates != null) - { - clientWebSocket.Options.ClientCertificates = new X509CertificateCollection(); - foreach (var certificate in _options.TlsOptions.Certificates) - { -#if WINDOWS_UWP - clientWebSocket.Options.ClientCertificates.Add(new X509Certificate(certificate)); -#else - clientWebSocket.Options.ClientCertificates.Add(certificate); #endif - - } - } - - var certificateValidationHandler = _options.TlsOptions?.CertificateValidationHandler; - if (certificateValidationHandler != null) - { -#if NETSTANDARD1_3 - throw new NotSupportedException("Remote certificate validation callback is not supported when using 'netstandard1.3'."); -#elif NETSTANDARD2_0 - throw new NotSupportedException("Remote certificate validation callback is not supported when using 'netstandard2.0'."); -#elif WINDOWS_UWP - throw new NotSupportedException("Remote certificate validation callback is not supported when using 'uap10.0'."); -#elif NET452 - throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net452'."); -#elif NET461 - throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net461'."); -#else - clientWebSocket.Options.RemoteCertificateValidationCallback = (sender, certificate, chain, sslPolicyErrors) => - { - // TODO: Find a way to add client options to same callback. Problem is that they have a different type. - var context = new MqttClientCertificateValidationEventArgs - { - Certificate = certificate, - Chain = chain, - SslPolicyErrors = sslPolicyErrors, - ClientOptions = _options - }; - - return certificateValidationHandler(context); - }; -#endif - } } void Cleanup() @@ -230,5 +165,77 @@ namespace MQTTnet.Implementations return new WebProxy(proxyUri, _options.ProxyOptions.BypassOnLocal, _options.ProxyOptions.BypassList); #endif } + + void SetupClientWebSocket(ClientWebSocket clientWebSocket) + { + if (_options.ProxyOptions != null) + { + clientWebSocket.Options.Proxy = CreateProxy(); + } + + if (_options.RequestHeaders != null) + { + foreach (var requestHeader in _options.RequestHeaders) + { + clientWebSocket.Options.SetRequestHeader(requestHeader.Key, requestHeader.Value); + } + } + + if (_options.SubProtocols != null) + { + foreach (var subProtocol in _options.SubProtocols) + { + clientWebSocket.Options.AddSubProtocol(subProtocol); + } + } + + if (_options.CookieContainer != null) + { + clientWebSocket.Options.Cookies = _options.CookieContainer; + } + + if (_options.TlsOptions?.UseTls == true && _options.TlsOptions?.Certificates != null) + { + clientWebSocket.Options.ClientCertificates = new X509CertificateCollection(); + foreach (var certificate in _options.TlsOptions.Certificates) + { +#if WINDOWS_UWP + clientWebSocket.Options.ClientCertificates.Add(new X509Certificate(certificate)); +#else + clientWebSocket.Options.ClientCertificates.Add(certificate); +#endif + } + } + + var certificateValidationHandler = _options.TlsOptions?.CertificateValidationHandler; + if (certificateValidationHandler != null) + { +#if NETSTANDARD1_3 + throw new NotSupportedException("Remote certificate validation callback is not supported when using 'netstandard1.3'."); +#elif NETSTANDARD2_0 + throw new NotSupportedException("Remote certificate validation callback is not supported when using 'netstandard2.0'."); +#elif WINDOWS_UWP + throw new NotSupportedException("Remote certificate validation callback is not supported when using 'uap10.0'."); +#elif NET452 + throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net452'."); +#elif NET461 + throw new NotSupportedException("Remote certificate validation callback is not supported when using 'net461'."); +#else + clientWebSocket.Options.RemoteCertificateValidationCallback = (sender, certificate, chain, sslPolicyErrors) => + { + // TODO: Find a way to add client options to same callback. Problem is that they have a different type. + var context = new MqttClientCertificateValidationEventArgs + { + Certificate = certificate, + Chain = chain, + SslPolicyErrors = sslPolicyErrors, + ClientOptions = _options + }; + + return certificateValidationHandler(context); + }; +#endif + } + } } } \ No newline at end of file