diff --git a/Build/MQTTnet.AspNetCore.nuspec b/Build/MQTTnet.AspNetCore.nuspec index c9936f35..43f056c6 100644 --- a/Build/MQTTnet.AspNetCore.nuspec +++ b/Build/MQTTnet.AspNetCore.nuspec @@ -10,12 +10,13 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false This is a support library to integrate MQTTnet into AspNetCore. - * Updated to MQTTnet 2.8.0. + * Updated to MQTTnet 2.8.1. +* For more release notes please check the MQTTnet release notes. Copyright Christian Kratky 2016-2018 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin - + diff --git a/Build/MQTTnet.Extensions.ManagedClient.nuspec b/Build/MQTTnet.Extensions.ManagedClient.nuspec index ba5333f2..44114a5d 100644 --- a/Build/MQTTnet.Extensions.ManagedClient.nuspec +++ b/Build/MQTTnet.Extensions.ManagedClient.nuspec @@ -10,12 +10,13 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false This is an extension library which provides a managed MQTT client with additional features using MQTTnet. - * Updated to MQTTnet 2.8.0. + * Updated to MQTTnet 2.8.1. +* For more release notes please check the MQTTnet release notes. Copyright Christian Kratky 2016-2018 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin - + diff --git a/Build/MQTTnet.Extensions.Rpc.nuspec b/Build/MQTTnet.Extensions.Rpc.nuspec index 0df8e7e6..1c8bee74 100644 --- a/Build/MQTTnet.Extensions.Rpc.nuspec +++ b/Build/MQTTnet.Extensions.Rpc.nuspec @@ -10,12 +10,13 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false This is an extension library which allows executing synchronous device calls including a response using MQTTnet. - * Updated to MQTTnet 2.8.0. + * Updated to MQTTnet 2.8.1. +* For more release notes please check the MQTTnet release notes. Copyright Christian Kratky 2016-2018 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin - + diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 5cdf3d52..34e563bc 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -10,33 +10,12 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker). - ** MQTTnet is now available at Open Collective for donations (https://opencollective.com/mqttnet). ** -* [Core] Performance optimizations. -* [Core] Due to performance reasons the timestamp of log messages is now in UTC format. -* [Core] Added several packet validations. -* [Core] Log messages now contain the complete source path including parent components. -* [Core] The adapter now has an _Endpoint_ definition as string containing remote IP and port. -* [Client] Received messages are now processed completely in the worker thread without creating new Tasks. -* [Client] Fixed wrong calculation for sending keep alive packets (thanks to @cstichlberger) -* [Client] A clean disconnect (via DisconnectAsync) will no longer throw an exception. -* [Client] Added new overloads for quick message publishing. -* [ManagedClient] The managed client is moved to a separate nuget package. -* [ManagedClient] Added an own message format with extended properties like ID (BREAKING CHANGE). -* [ManagedClient] Fixed a loading issue of stored application messages (thanks to @JTrotta). -* [ManagedClient] Added a new event which is fired when a synchronization of the subscriptions has failed. -* [ManagedClient] Added a new event which is fired when a connection attempt has failed. -* [ManagedClient] Exposed a new property which provides the count of not published messages (pending messages count). -* [Server] Added support for other WebSocket sub protocol formats like mqttv-3.1.1 (thanks to @israellot). -* [Server] The takeover of an existing client sessions is now treated as a _clean_ disconnect of the previous client. -* [Server] The pending messages queue per client is now limited to 250 messages. Overflow strategy and count can be changed via options (thanks to @VladimirAkopyan) -* [Server] Keep alive checking is now suspended while large packages are being received (and thus the client is connected). Keep alive checking continues after a large packet is received completely. -* [Server] Rewritten the _ConnectedClients_ API and added new features for disconnecting and Endpoint information (IP etc.). -* [Server] Added settings for disabling persistent sessions and defining a max pending messages queue size per session. -* [Server] Persistent sessions are disabled by default (BREAKING CHANGE!). -* [Server] Added a new interceptor which is invoked before a new message is added to the client queue. -* [Server] Added support for Linux servers by dividing IPv4 and IPv6 support and adding new options (BREAKING CHANGE!). -* [Server] Gracefully closed connections are no longer reported as warnings. -* [Server] Added new overloads for initializing the ASP.NET Core integration. + * [Core] Performance optimizations. +* [Core] Fixed a bug which prevents receiving large packets (UWP only) +* [Client] The ManagedClient options now allow configuring the interval for connection checks. +* [Server] Added the Endpoint of the Adapter (remote IP and port) to the connection validation callback. +* [Server] The ipv4 and ipv6 endpoint can be disabled now by setting the bound IP address to _None_. +* [Server] Fix a bug in the keep alive monitor which caused high CPU load (thanks to @GarageGadget). Copyright Christian Kratky 2016-2018 MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin diff --git a/Source/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs b/Source/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs index 22b9e937..2125c7df 100644 --- a/Source/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs +++ b/Source/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs @@ -3,6 +3,7 @@ using Microsoft.AspNetCore.Builder; using Microsoft.Extensions.DependencyInjection; using System.Linq; using MQTTnet.Server; +using System.Collections.Generic; namespace MQTTnet.AspNetCore { @@ -23,10 +24,7 @@ namespace MQTTnet.AspNetCore if (context.Request.Headers.TryGetValue("Sec-WebSocket-Protocol", out var requestedSubProtocolValues)) { - // Order the protocols to also match "mqtt", "mqttv-3.1", "mqttv-3.11" etc. - subProtocol = requestedSubProtocolValues - .OrderByDescending(p => p.Length) - .FirstOrDefault(p => p.ToLower().StartsWith("mqtt")); + subProtocol = SelectSubProtocol(requestedSubProtocolValues); } var adapter = app.ApplicationServices.GetRequiredService(); @@ -40,6 +38,14 @@ namespace MQTTnet.AspNetCore return app; } + public static string SelectSubProtocol(IList requestedSubProtocolValues) + { + // Order the protocols to also match "mqtt", "mqttv-3.1", "mqttv-3.11" etc. + return requestedSubProtocolValues + .OrderByDescending(p => p.Length) + .FirstOrDefault(p => p.ToLower().StartsWith("mqtt")); + } + public static IApplicationBuilder UseMqttServer(this IApplicationBuilder app, Action configure) { var server = app.ApplicationServices.GetRequiredService(); diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs b/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs index 49a5c098..23903894 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs @@ -1,4 +1,5 @@ using Microsoft.AspNetCore.Connections; +using Microsoft.AspNetCore.Connections.Features; using MQTTnet.Adapter; using MQTTnet.Serializer; using MQTTnet.Server; @@ -13,6 +14,14 @@ namespace MQTTnet.AspNetCore public override async Task OnConnectedAsync(ConnectionContext connection) { + // required for websocket transport to work + var transferFormatFeature = connection.Features.Get(); + if (transferFormatFeature != null) + { + transferFormatFeature.ActiveFormat = TransferFormat.Binary; + } + + var serializer = new MqttPacketSerializer(); using (var adapter = new MqttConnectionContext(serializer, connection)) { diff --git a/Source/MQTTnet.AspnetCore/ReaderExtensions.cs b/Source/MQTTnet.AspnetCore/ReaderExtensions.cs index b893e43e..6c770b9e 100644 --- a/Source/MQTTnet.AspnetCore/ReaderExtensions.cs +++ b/Source/MQTTnet.AspnetCore/ReaderExtensions.cs @@ -32,7 +32,8 @@ namespace MQTTnet.AspNetCore } var bodySlice = copy.Slice(0, bodyLength); - packet = serializer.Deserialize(new ReceivedMqttPacket(fixedheader, new MqttPacketBodyReader(bodySlice.GetArray(), 0))); + var buffer = bodySlice.GetArray(); + packet = serializer.Deserialize(new ReceivedMqttPacket(fixedheader, new MqttPacketBodyReader(buffer, 0, buffer.Length))); consumed = bodySlice.End; observed = bodySlice.End; return true; diff --git a/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs b/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs index 8b65bbeb..b5d0440a 100644 --- a/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs +++ b/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs @@ -44,7 +44,7 @@ namespace MQTTnet.AspNetCore return services; } - public static IServiceCollection AddHostedMqttServer(this IServiceCollection services) + private static IServiceCollection AddHostedMqttServer(this IServiceCollection services) { var logger = new MqttNetLogger(); var childLogger = logger.CreateChildLogger(); diff --git a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs index 77736c44..8d48babd 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs @@ -9,6 +9,8 @@ namespace MQTTnet.Extensions.ManagedClient TimeSpan AutoReconnectDelay { get; } + TimeSpan ConnectionCheckInterval { get; } + IManagedMqttClientStorage Storage { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index bcd98ae7..2eabce33 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -206,7 +206,7 @@ namespace MQTTnet.Extensions.ManagedClient if (connectionState == ReconnectionResult.StillConnected) { - await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false); + await Task.Delay(_options.ConnectionCheckInterval, cancellationToken).ConfigureAwait(false); } } catch (OperationCanceledException) diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs index e07e2d68..cecf2274 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs @@ -9,6 +9,8 @@ namespace MQTTnet.Extensions.ManagedClient public TimeSpan AutoReconnectDelay { get; set; } = TimeSpan.FromSeconds(5); + public TimeSpan ConnectionCheckInterval { get; set; } = TimeSpan.FromSeconds(1); + public IManagedMqttClientStorage Storage { get; set; } } } diff --git a/Source/MQTTnet.Extensions.Rpc/SampleCCode.c b/Source/MQTTnet.Extensions.Rpc/SampleCCode.c index c20aa4af..abcd6551 100644 --- a/Source/MQTTnet.Extensions.Rpc/SampleCCode.c +++ b/Source/MQTTnet.Extensions.Rpc/SampleCCode.c @@ -3,7 +3,7 @@ _mqttClient.subscribe("MQTTnet.RPC/+/ping"); _mqttClient.subscribe("MQTTnet.RPC/+/do_something"); // It is not allowed to change the structure of the topic. Otherwise RPC will not work. So method names can be separated using -// an _ or . but no +, # or . If it is required to distinguish between devices own rules can be defined like the following. +// an _ or . but no +, # or /. If it is required to distinguish between devices own rules can be defined like the following. _mqttClient.subscribe("MQTTnet.RPC/+/deviceA.ping"); _mqttClient.subscribe("MQTTnet.RPC/+/deviceB.ping"); _mqttClient.subscribe("MQTTnet.RPC/+/deviceC.getTemperature"); diff --git a/Source/MQTTnet.Extensions.Wrappers.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs b/Source/MQTTnet.Extensions.Wrappers.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs new file mode 100644 index 00000000..ed29be81 --- /dev/null +++ b/Source/MQTTnet.Extensions.Wrappers.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs @@ -0,0 +1,134 @@ +using MQTTnet.Client; +using System; +using System.Collections.Concurrent; +using System.Linq; +using System.Security.Authentication; +using System.Threading; +using System.Threading.Tasks; +using MQTTnet.Adapter; +using MQTTnet.Channel; +using MQTTnet.Diagnostics; +using MQTTnet.Serializer; +using WebSocket4Net; + +namespace MQTTnet.TestApp.NetCore +{ + 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)); + + if (!(options.ChannelOptions is MqttClientWebSocketOptions)) + { + throw new NotSupportedException("Only WebSocket connections are supported."); + } + + return new MqttChannelAdapter(new WebSocket4NetMqttChannel(options), new MqttPacketSerializer(), logger); + } + + private class WebSocket4NetMqttChannel : IMqttChannel + { + private readonly BlockingCollection _receiveBuffer = new BlockingCollection(); + + private readonly IMqttClientOptions _clientOptions; + private WebSocket4Net.WebSocket _webSocket; + + public WebSocket4NetMqttChannel(IMqttClientOptions clientOptions) + { + _clientOptions = clientOptions ?? throw new ArgumentNullException(nameof(clientOptions)); + } + + public string Endpoint { get; } = ""; + + public Task ConnectAsync(CancellationToken cancellationToken) + { + var channelOptions = (MqttClientWebSocketOptions)_clientOptions.ChannelOptions; + + var uri = "ws://" + channelOptions.Uri; + var sslProtocols = SslProtocols.None; + + if (channelOptions.TlsOptions.UseTls) + { + uri = "wss://" + channelOptions.Uri; + sslProtocols = SslProtocols.Tls12; + } + + var subProtocol = channelOptions.SubProtocols.FirstOrDefault() ?? string.Empty; + + _webSocket = new WebSocket4Net.WebSocket(uri, subProtocol, sslProtocols: sslProtocols); + _webSocket.DataReceived += OnDataReceived; + _webSocket.Open(); + SpinWait.SpinUntil(() => _webSocket.State == WebSocketState.Open, _clientOptions.CommunicationTimeout); + + return Task.FromResult(0); + } + + public Task DisconnectAsync() + { + if (_webSocket != null) + { + _webSocket.DataReceived -= OnDataReceived; + _webSocket.Close(); + SpinWait.SpinUntil(() => _webSocket.State == WebSocketState.Closed, _clientOptions.CommunicationTimeout); + } + + _webSocket = null; + + return Task.FromResult(0); + } + + public Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + { + var readBytes = 0; + while (count > 0 && !cancellationToken.IsCancellationRequested) + { + byte @byte; + if (readBytes == 0) + { + // Block until at lease one byte was received. + @byte = _receiveBuffer.Take(cancellationToken); + } + else + { + if (!_receiveBuffer.TryTake(out @byte)) + { + 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) + { + _webSocket.DataReceived -= OnDataReceived; + _webSocket.Dispose(); + } + } + + private void OnDataReceived(object sender, WebSocket4Net.DataReceivedEventArgs e) + { + foreach (var @byte in e.Data) + { + _receiveBuffer.Add(@byte); + } + } + } + } +} diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index 0059f201..4e617330 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -97,13 +97,11 @@ namespace MQTTnet.Adapter await _writerSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false); try { - _logger.Verbose("TX >>> {0}", packet); - var packetData = PacketSerializer.Serialize(packet); - await _channel.WriteAsync(packetData.Array, packetData.Offset, packetData.Count, cancellationToken).ConfigureAwait(false); - PacketSerializer.FreeBuffer(); + + _logger.Verbose("TX >>> {0}", packet); } catch (Exception exception) { @@ -130,11 +128,11 @@ namespace MQTTnet.Adapter if (timeout > TimeSpan.Zero) { - receivedMqttPacket = await Internal.TaskExtensions.TimeoutAfterAsync(ct => ReceiveAsync(_channel, ct), timeout, cancellationToken).ConfigureAwait(false); + receivedMqttPacket = await Internal.TaskExtensions.TimeoutAfterAsync(ReceiveAsync, timeout, cancellationToken).ConfigureAwait(false); } else { - receivedMqttPacket = await ReceiveAsync(_channel, cancellationToken).ConfigureAwait(false); + receivedMqttPacket = await ReceiveAsync(cancellationToken).ConfigureAwait(false); } if (receivedMqttPacket == null || cancellationToken.IsCancellationRequested) @@ -149,7 +147,7 @@ namespace MQTTnet.Adapter } _logger.Verbose("RX <<< {0}", packet); - + return packet; } catch (Exception exception) @@ -165,9 +163,9 @@ namespace MQTTnet.Adapter return null; } - private async Task ReceiveAsync(IMqttChannel channel, CancellationToken cancellationToken) + private async Task ReceiveAsync(CancellationToken cancellationToken) { - var fixedHeader = await MqttPacketReader.ReadFixedHeaderAsync(channel, _fixedHeaderBuffer, _singleByteBuffer, cancellationToken).ConfigureAwait(false); + var fixedHeader = await MqttPacketReader.ReadFixedHeaderAsync(_channel, _fixedHeaderBuffer, _singleByteBuffer, cancellationToken).ConfigureAwait(false); try { @@ -190,18 +188,21 @@ namespace MQTTnet.Adapter chunkSize = bytesLeft; } +#if WINDOWS_UWP + var readBytes = await _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false); +#else // async/await is not used to avoid the overhead of context switches. We assume that the reamining data // has been sent from the sender directly after the initial bytes. - var readBytes = channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).GetAwaiter().GetResult(); - if (readBytes <= 0) - { - ExceptionHelper.ThrowGracefulSocketClose(); - } + var readBytes = _channel.ReadAsync(body, bodyOffset, chunkSize, cancellationToken).ConfigureAwait(false).GetAwaiter().GetResult(); +#endif + + cancellationToken.ThrowIfCancellationRequested(); + ExceptionHelper.ThrowIfGracefulSocketClose(readBytes); bodyOffset += readBytes; } while (bodyOffset < body.Length); - return new ReceivedMqttPacket(fixedHeader.Flags, new MqttPacketBodyReader(body, 0)); + return new ReceivedMqttPacket(fixedHeader.Flags, new MqttPacketBodyReader(body, 0, body.Length)); } finally { diff --git a/Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs index 2ccc7d97..6da9da24 100644 --- a/Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs +++ b/Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs @@ -154,7 +154,11 @@ namespace MQTTnet.Implementations private void CreateStreams() { - _readStream = _socket.InputStream.AsStreamForRead(_bufferSize); + // Attention! Do not set the buffer for the read method. This will + // limit the internal buffer and the read operation will hang forever + // if more data than the buffer size was received. + _readStream = _socket.InputStream.AsStreamForRead(); + _writeStream = _socket.OutputStream.AsStreamForWrite(_bufferSize); } diff --git a/Source/MQTTnet/Implementations/MqttTcpServerAdapter.cs b/Source/MQTTnet/Implementations/MqttTcpServerAdapter.cs index eeefb344..cfbf7251 100644 --- a/Source/MQTTnet/Implementations/MqttTcpServerAdapter.cs +++ b/Source/MQTTnet/Implementations/MqttTcpServerAdapter.cs @@ -1,6 +1,7 @@ #if !WINDOWS_UWP using System; using System.Collections.Generic; +using System.Net; using System.Net.Sockets; using System.Security.Cryptography.X509Certificates; using System.Threading; @@ -79,27 +80,33 @@ namespace MQTTnet.Implementations private void RegisterListeners(MqttServerTcpEndpointBaseOptions options, X509Certificate2 tlsCertificate) { - var listenerV4 = new MqttTcpServerListener( - AddressFamily.InterNetwork, - options, - tlsCertificate, - _cancellationTokenSource.Token, - _logger); + if (!options.BoundInterNetworkAddress.Equals(IPAddress.None)) + { + var listenerV4 = new MqttTcpServerListener( + AddressFamily.InterNetwork, + options, + tlsCertificate, + _cancellationTokenSource.Token, + _logger); - listenerV4.ClientAccepted += OnClientAccepted; - listenerV4.Start(); - _listeners.Add(listenerV4); + listenerV4.ClientAccepted += OnClientAccepted; + listenerV4.Start(); + _listeners.Add(listenerV4); + } - var listenerV6 = new MqttTcpServerListener( - AddressFamily.InterNetworkV6, - options, - tlsCertificate, - _cancellationTokenSource.Token, - _logger); + if (!options.BoundInterNetworkV6Address.Equals(IPAddress.None)) + { + var listenerV6 = new MqttTcpServerListener( + AddressFamily.InterNetworkV6, + options, + tlsCertificate, + _cancellationTokenSource.Token, + _logger); - listenerV6.ClientAccepted += OnClientAccepted; - listenerV6.Start(); - _listeners.Add(listenerV6); + listenerV6.ClientAccepted += OnClientAccepted; + listenerV6.Start(); + _listeners.Add(listenerV6); + } } private void OnClientAccepted(object sender, MqttServerAdapterClientAcceptedEventArgs e) diff --git a/Source/MQTTnet/Internal/ExceptionHelper.cs b/Source/MQTTnet/Internal/ExceptionHelper.cs index f3cadc91..5bc8e435 100644 --- a/Source/MQTTnet/Internal/ExceptionHelper.cs +++ b/Source/MQTTnet/Internal/ExceptionHelper.cs @@ -8,5 +8,13 @@ namespace MQTTnet.Internal { throw new MqttCommunicationClosedGracefullyException(); } + + public static void ThrowIfGracefulSocketClose(int readBytesCount) + { + if (readBytesCount <= 0) + { + throw new MqttCommunicationClosedGracefullyException(); + } + } } } diff --git a/Source/MQTTnet/Internal/TestMqttChannel.cs b/Source/MQTTnet/Internal/TestMqttChannel.cs index e48b8ede..08920e3c 100644 --- a/Source/MQTTnet/Internal/TestMqttChannel.cs +++ b/Source/MQTTnet/Internal/TestMqttChannel.cs @@ -14,11 +14,7 @@ namespace MQTTnet.Internal _stream = stream; } - public void Dispose() - { - } - - public string Endpoint { get; } + public string Endpoint { get; } = ""; public Task ConnectAsync(CancellationToken cancellationToken) { @@ -39,5 +35,9 @@ namespace MQTTnet.Internal { return _stream.WriteAsync(buffer, offset, count, cancellationToken); } + + public void Dispose() + { + } } } diff --git a/Source/MQTTnet/MQTTnet.csproj b/Source/MQTTnet/MQTTnet.csproj index 26bf0029..984b5630 100644 --- a/Source/MQTTnet/MQTTnet.csproj +++ b/Source/MQTTnet/MQTTnet.csproj @@ -20,7 +20,7 @@ UAP,Version=v10.0 UAP 10.0.17134.0 - 10.0.10240.0 + 10.0.10586.212 .NETCore v5.0 $(DefineConstants);WINDOWS_UWP @@ -53,7 +53,7 @@ - + diff --git a/Source/MQTTnet/MqttFactory.cs b/Source/MQTTnet/MqttFactory.cs index e830ad20..3f1e1acb 100644 --- a/Source/MQTTnet/MqttFactory.cs +++ b/Source/MQTTnet/MqttFactory.cs @@ -22,12 +22,19 @@ namespace MQTTnet return new MqttClient(new MqttClientAdapterFactory(), logger); } - public IMqttClient CreateMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory mqttClientAdapterFactory) + public IMqttClient CreateMqttClient(IMqttClientAdapterFactory adapterFactory) + { + if (adapterFactory == null) throw new ArgumentNullException(nameof(adapterFactory)); + + return new MqttClient(adapterFactory, new MqttNetLogger()); + } + + public IMqttClient CreateMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory adapterFactory) { if (logger == null) throw new ArgumentNullException(nameof(logger)); - if (mqttClientAdapterFactory == null) throw new ArgumentNullException(nameof(mqttClientAdapterFactory)); + if (adapterFactory == null) throw new ArgumentNullException(nameof(adapterFactory)); - return new MqttClient(mqttClientAdapterFactory, logger); + return new MqttClient(adapterFactory, logger); } public IMqttServer CreateMqttServer() diff --git a/Source/MQTTnet/Serializer/MqttPacketBodyReader.cs b/Source/MQTTnet/Serializer/MqttPacketBodyReader.cs index 2cf37dd4..f0e1b562 100644 --- a/Source/MQTTnet/Serializer/MqttPacketBodyReader.cs +++ b/Source/MQTTnet/Serializer/MqttPacketBodyReader.cs @@ -6,17 +6,20 @@ namespace MQTTnet.Serializer public class MqttPacketBodyReader { private readonly byte[] _buffer; - private int _offset; + private readonly int _length; - public MqttPacketBodyReader(byte[] buffer, int offset) + private int _offset; + + public MqttPacketBodyReader(byte[] buffer, int offset, int length) { _buffer = buffer; _offset = offset; + _length = length; } - public int Length => _buffer.Length - _offset; + public int Length => _length - _offset; - public bool EndOfStream => _offset == _buffer.Length; + public bool EndOfStream => _offset == _length; public byte ReadByte() { @@ -26,7 +29,7 @@ namespace MQTTnet.Serializer public ArraySegment ReadRemainingData() { - return new ArraySegment(_buffer, _offset, _buffer.Length - _offset); + return new ArraySegment(_buffer, _offset, _length - _offset); } public ushort ReadUInt16() @@ -53,9 +56,9 @@ namespace MQTTnet.Serializer private void ValidateReceiveBuffer(ushort length) { - if (_buffer.Length < _offset + length) + if (_length < _offset + length) { - throw new ArgumentOutOfRangeException(nameof(_buffer), $"expected at least {_offset + length} bytes but there are only {_buffer.Length} bytes"); + throw new ArgumentOutOfRangeException(nameof(_buffer), $"expected at least {_offset + length} bytes but there are only {_length} bytes"); } } diff --git a/Source/MQTTnet/Serializer/MqttPacketReader.cs b/Source/MQTTnet/Serializer/MqttPacketReader.cs index 32a92855..f115b507 100644 --- a/Source/MQTTnet/Serializer/MqttPacketReader.cs +++ b/Source/MQTTnet/Serializer/MqttPacketReader.cs @@ -1,5 +1,4 @@ -using System; -using System.Threading; +using System.Threading; using System.Threading.Tasks; using MQTTnet.Channel; using MQTTnet.Exceptions; @@ -21,11 +20,9 @@ namespace MQTTnet.Serializer while (totalBytesRead < buffer.Length) { var bytesRead = await channel.ReadAsync(buffer, totalBytesRead, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false); - if (bytesRead <= 0) - { - cancellationToken.ThrowIfCancellationRequested(); - ExceptionHelper.ThrowGracefulSocketClose(); - } + + cancellationToken.ThrowIfCancellationRequested(); + ExceptionHelper.ThrowIfGracefulSocketClose(bytesRead); totalBytesRead += bytesRead; } @@ -35,34 +32,42 @@ namespace MQTTnet.Serializer { return new MqttFixedHeader(buffer[0], 0); } - + +#if WINDOWS_UWP + // UWP will have a dead lock when calling this not async. + var bodyLength = await ReadBodyLengthAsync(channel, buffer[1], singleByteBuffer, cancellationToken).ConfigureAwait(false); +#else + // Here the async/await pattern is not used becuase the overhead of context switches + // is too big for reading 1 byte in a row. We expect that the remaining data was sent + // directly after the initial bytes. If the client disconnects just in this moment we + // will get an exception anyway. var bodyLength = ReadBodyLength(channel, buffer[1], singleByteBuffer, cancellationToken); +#endif + return new MqttFixedHeader(buffer[0], bodyLength); } +#if !WINDOWS_UWP private static int ReadBodyLength(IMqttChannel channel, byte initialEncodedByte, byte[] singleByteBuffer, CancellationToken cancellationToken) { - // Alorithm taken from https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os-complete.html. + var offset = 0; var multiplier = 128; var value = initialEncodedByte & 127; int encodedByte = initialEncodedByte; while ((encodedByte & 128) != 0) { - cancellationToken.ThrowIfCancellationRequested(); - - // Here the async/await pattern is not used becuase the overhead of context switches - // is too big for reading 1 byte in a row. We expect that the remaining data was sent - // directly after the initial bytes. If the client disconnects just in this moment we - // will get an exception anyway. - encodedByte = ReadByte(channel, singleByteBuffer, cancellationToken); - - value += (byte)(encodedByte & 127) * multiplier; - if (multiplier > 128 * 128 * 128) + offset++; + if (offset > 3) { throw new MqttProtocolViolationException("Remaining length is invalid."); } + cancellationToken.ThrowIfCancellationRequested(); + + encodedByte = ReadByte(channel, singleByteBuffer, cancellationToken); + + value += (byte)(encodedByte & 127) * multiplier; multiplier *= 128; } @@ -71,14 +76,60 @@ namespace MQTTnet.Serializer private static byte ReadByte(IMqttChannel channel, byte[] singleByteBuffer, CancellationToken cancellationToken) { - var readCount = channel.ReadAsync(singleByteBuffer, 0, 1, cancellationToken).GetAwaiter().GetResult(); + var readCount = channel.ReadAsync(singleByteBuffer, 0, 1, cancellationToken).ConfigureAwait(false).GetAwaiter().GetResult(); + + cancellationToken.ThrowIfCancellationRequested(); + if (readCount <= 0) { - cancellationToken.ThrowIfCancellationRequested(); ExceptionHelper.ThrowGracefulSocketClose(); } return singleByteBuffer[0]; } + +#else + + private static async Task ReadBodyLengthAsync(IMqttChannel channel, byte initialEncodedByte, byte[] singleByteBuffer, CancellationToken cancellationToken) + { + var offset = 0; + var multiplier = 128; + var value = initialEncodedByte & 127; + int encodedByte = initialEncodedByte; + + while ((encodedByte & 128) != 0) + { + offset++; + if (offset > 3) + { + throw new MqttProtocolViolationException("Remaining length is invalid."); + } + + cancellationToken.ThrowIfCancellationRequested(); + + encodedByte = await ReadByteAsync(channel, singleByteBuffer, cancellationToken).ConfigureAwait(false); + + value += (byte)(encodedByte & 127) * multiplier; + multiplier *= 128; + } + + return value; + } + + private static async Task ReadByteAsync(IMqttChannel channel, byte[] singleByteBuffer, CancellationToken cancellationToken) + { + var readCount = await channel.ReadAsync(singleByteBuffer, 0, 1, cancellationToken).ConfigureAwait(false); + + cancellationToken.ThrowIfCancellationRequested(); + + if (readCount <= 0) + { + ExceptionHelper.ThrowGracefulSocketClose(); + } + + return singleByteBuffer[0]; + } + +#endif } } diff --git a/Source/MQTTnet/Serializer/MqttPacketWriter.cs b/Source/MQTTnet/Serializer/MqttPacketWriter.cs index c7f1bc2c..67fd4672 100644 --- a/Source/MQTTnet/Serializer/MqttPacketWriter.cs +++ b/Source/MQTTnet/Serializer/MqttPacketWriter.cs @@ -29,7 +29,6 @@ namespace MQTTnet.Serializer public static ArraySegment EncodeRemainingLength(int length) { - // write the encoded remaining length right aligned on the 4 byte buffer if (length <= 0) { return new ArraySegment(new byte[1], 0, 1); @@ -38,7 +37,6 @@ namespace MQTTnet.Serializer var buffer = new byte[4]; var bufferOffset = 0; - // Algorithm taken from http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html. var x = length; do { diff --git a/Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs b/Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs index b3628612..8afb7acd 100644 --- a/Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs +++ b/Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs @@ -70,7 +70,9 @@ namespace MQTTnet.Server while (!cancellationToken.IsCancellationRequested) { // Values described here: [MQTT-3.1.2-24]. - if (!_isPaused && _lastPacketReceivedTracker.Elapsed.TotalSeconds > keepAlivePeriod * 1.5D) + // If the client sends 5 sec. the server will allow up to 7.5 seconds. + // If the client sends 1 sec. the server will allow up to 1.5 seconds. + if (!_isPaused && _lastPacketReceivedTracker.Elapsed.TotalSeconds >= keepAlivePeriod * 1.5D) { _logger.Warning(null, "Client '{0}': Did not receive any packet or keep alive signal.", _clientSession.ClientId); _clientSession.Stop(MqttClientDisconnectType.NotClean); @@ -78,7 +80,11 @@ namespace MQTTnet.Server return; } - await Task.Delay(keepAlivePeriod, cancellationToken).ConfigureAwait(false); + // The server checks the keep alive timeout every 50 % of the overall keep alive timeout + // because the server allows 1.5 times the keep alive value. This means that a value of 5 allows + // up to 7.5 seconds. With an interval of 2.5 (5 / 2) the 7.5 is also affected. Waiting the whole + // keep alive time will hit at 10 instead of 7.5 (but only one time instead of two times). + await Task.Delay(TimeSpan.FromSeconds(keepAlivePeriod * 0.5D), cancellationToken).ConfigureAwait(false); } } catch (OperationCanceledException) diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index a41be4f1..34080a00 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -212,7 +212,7 @@ namespace MQTTnet.Server // Switch to the required protocol version before sending any response. clientAdapter.PacketSerializer.ProtocolVersion = connectPacket.ProtocolVersion; - var connectReturnCode = ValidateConnection(connectPacket); + var connectReturnCode = ValidateConnection(connectPacket, clientAdapter); if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted) { await clientAdapter.SendPacketAsync( @@ -268,7 +268,7 @@ namespace MQTTnet.Server } } - private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket) + private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter clientAdapter) { if (_options.ConnectionValidator == null) { @@ -279,7 +279,8 @@ namespace MQTTnet.Server connectPacket.ClientId, connectPacket.Username, connectPacket.Password, - connectPacket.WillMessage); + connectPacket.WillMessage, + clientAdapter.Endpoint); _options.ConnectionValidator(context); return context.ReturnCode; diff --git a/Source/MQTTnet/Server/MqttConnectionValidatorContext.cs b/Source/MQTTnet/Server/MqttConnectionValidatorContext.cs index a24d7955..5857862b 100644 --- a/Source/MQTTnet/Server/MqttConnectionValidatorContext.cs +++ b/Source/MQTTnet/Server/MqttConnectionValidatorContext.cs @@ -4,12 +4,13 @@ namespace MQTTnet.Server { public class MqttConnectionValidatorContext { - public MqttConnectionValidatorContext(string clientId, string username, string password, MqttApplicationMessage willMessage) + public MqttConnectionValidatorContext(string clientId, string username, string password, MqttApplicationMessage willMessage, string endpoint) { ClientId = clientId; Username = username; Password = password; WillMessage = willMessage; + Endpoint = endpoint; } public string ClientId { get; } @@ -20,6 +21,8 @@ namespace MQTTnet.Server public MqttApplicationMessage WillMessage { get; } + public string Endpoint { get; } + public MqttConnectReturnCode ReturnCode { get; set; } = MqttConnectReturnCode.ConnectionAccepted; } } diff --git a/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs b/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs index 8beb7b49..9b855805 100644 --- a/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs @@ -58,7 +58,7 @@ namespace MQTTnet.Benchmarks var receivedPacket = new ReceivedMqttPacket( header.Flags, - new MqttPacketBodyReader(_serializedPacket.Array, _serializedPacket.Count - header.RemainingLength)); + new MqttPacketBodyReader(_serializedPacket.Array, _serializedPacket.Count - header.RemainingLength, _serializedPacket.Array.Length)); _serializer.Deserialize(receivedPacket); } diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs index d8d23e29..c3a75f8e 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs @@ -1,5 +1,6 @@ using System; using System.IO; +using System.Linq; using System.Text; using System.Threading; using Microsoft.VisualStudio.TestTools.UnitTesting; @@ -149,6 +150,53 @@ namespace MQTTnet.Core.Tests DeserializeAndCompare(p, "IAIABQ==", MqttProtocolVersion.V310); } + [TestMethod] + public void Serialize_LargePacket() + { + var serializer = new MqttPacketSerializer { ProtocolVersion = MqttProtocolVersion.V311 }; + + const int payloadLength = 80000; + + var payload = new byte[payloadLength]; + + var value = 0; + for (var i = 0; i < payloadLength; i++) + { + if (value > 255) + { + value = 0; + } + + payload[i] = (byte)value; + } + + var publishPacket = new MqttPublishPacket + { + Topic = "abcdefghijklmnopqrstuvwxyz0123456789", + Payload = payload + }; + + var buffer = serializer.Serialize(publishPacket); + var testChannel = new TestMqttChannel(new MemoryStream(buffer.Array, buffer.Offset, buffer.Count)); + + var header = MqttPacketReader.ReadFixedHeaderAsync( + testChannel, + new byte[2], + new byte[1], + CancellationToken.None).GetAwaiter().GetResult(); + + var eof = buffer.Offset + buffer.Count; + + var receivedPacket = new ReceivedMqttPacket( + header.Flags, + new MqttPacketBodyReader(buffer.Array, eof - header.RemainingLength, buffer.Count + buffer.Offset)); + + var packet = (MqttPublishPacket)serializer.Deserialize(receivedPacket); + + Assert.AreEqual(publishPacket.Topic, packet.Topic); + Assert.IsTrue(publishPacket.Payload.SequenceEqual(packet.Payload)); + } + [TestMethod] public void SerializeV311_MqttDisconnectPacket() { @@ -463,7 +511,7 @@ namespace MQTTnet.Core.Tests { var serializer = new MqttPacketSerializer { ProtocolVersion = protocolVersion }; var data = serializer.Serialize(packet); - + Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(data))); } @@ -479,10 +527,10 @@ namespace MQTTnet.Core.Tests var fixedHeader = new byte[2]; var singleByteBuffer = new byte[1]; var header = MqttPacketReader.ReadFixedHeaderAsync(channel, fixedHeader, singleByteBuffer, CancellationToken.None).GetAwaiter().GetResult(); - + using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, header.RemainingLength)) { - var deserializedPacket = serializer.Deserialize(new ReceivedMqttPacket(header.Flags, new MqttPacketBodyReader(bodyStream.ToArray(), 0))); + var deserializedPacket = serializer.Deserialize(new ReceivedMqttPacket(header.Flags, new MqttPacketBodyReader(bodyStream.ToArray(), 0, (int)bodyStream.Length))); var buffer2 = serializer.Serialize(deserializedPacket); Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(buffer2))); @@ -507,7 +555,7 @@ namespace MQTTnet.Core.Tests using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, header.RemainingLength)) { - return (T)serializer.Deserialize(new ReceivedMqttPacket(header.Flags, new MqttPacketBodyReader(bodyStream.ToArray(), 0))); + return (T)serializer.Deserialize(new ReceivedMqttPacket(header.Flags, new MqttPacketBodyReader(bodyStream.ToArray(), 0, (int)bodyStream.Length))); } } } diff --git a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs index ea258f56..95bb67da 100644 --- a/Tests/MQTTnet.Core.Tests/MqttServerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttServerTests.cs @@ -375,6 +375,9 @@ namespace MQTTnet.Core.Tests var c1 = await serverAdapter.ConnectTestClient("c1"); await c1.PublishAndWaitForAsync(s, new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build()); + + await Task.Delay(250); + await c1.DisconnectAsync(); } finally @@ -396,7 +399,7 @@ namespace MQTTnet.Core.Tests c2.ApplicationMessageReceived += (_, __) => receivedMessagesCount++; await c2.SubscribeAsync(new TopicFilterBuilder().WithTopic("retained").Build()); - await Task.Delay(500); + await Task.Delay(250); } finally { diff --git a/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs b/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs index f52040a0..db352f3c 100644 --- a/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs +++ b/Tests/MQTTnet.TestApp.AspNetCore2/Startup.cs @@ -22,13 +22,18 @@ namespace MQTTnet.TestApp.AspNetCore2 .Build(); services .AddHostedMqttServer(mqttServerOptions) - .AddMqttConnectionHandler(); + .AddMqttConnectionHandler() + .AddConnections(); } // In class _Startup_ of the ASP.NET Core 2.0 project. public void Configure(IApplicationBuilder app, IHostingEnvironment env) { - app.UseMqttEndpoint(); + app.UseConnections(c => c.MapConnectionHandler("/mqtt", options => { + options.WebSockets.SubProtocolSelector = MQTTnet.AspNetCore.ApplicationBuilderExtensions.SelectSubProtocol; + })); + + //app.UseMqttEndpoint(); app.UseMqttServer(server => { server.Started += async (sender, args) => diff --git a/Tests/MQTTnet.TestApp.NetCore/PublicBrokerTest.cs b/Tests/MQTTnet.TestApp.NetCore/PublicBrokerTest.cs index 1265a6a2..e46ed168 100644 --- a/Tests/MQTTnet.TestApp.NetCore/PublicBrokerTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/PublicBrokerTest.cs @@ -3,6 +3,7 @@ using System; using System.IO; using System.Threading; using System.Threading.Tasks; +using MQTTnet.Diagnostics; using MQTTnet.Protocol; using Newtonsoft.Json; @@ -12,7 +13,7 @@ namespace MQTTnet.TestApp.NetCore { public static async Task RunAsync() { - //MqttNetGlobalLogger.LogMessagePublished += (s, e) => Console.WriteLine(e.TraceMessage); + //MqttNetConsoleLogger.ForwardToConsole(); // iot.eclipse.org await ExecuteTestAsync("iot.eclipse.org TCP", @@ -35,7 +36,7 @@ namespace MQTTnet.TestApp.NetCore new MqttClientOptionsBuilder().WithWebSocketServer("test.mosquitto.org:8080/mqtt").Build()); await ExecuteTestAsync("test.mosquitto.org WS TLS", - new MqttClientOptionsBuilder().WithWebSocketServer("test.mosquitto.org:8081/mqtt").Build()); + new MqttClientOptionsBuilder().WithWebSocketServer("test.mosquitto.org:8081/mqtt").WithTls().Build()); // broker.hivemq.com await ExecuteTestAsync("broker.hivemq.com TCP", @@ -78,7 +79,7 @@ namespace MQTTnet.TestApp.NetCore var topic = Guid.NewGuid().ToString(); MqttApplicationMessage receivedMessage = null; - client.ApplicationMessageReceived += (s, e) => receivedMessage = e.ApplicationMessage; + client.ApplicationMessageReceived += (s, e) => receivedMessage = e.ApplicationMessage; await client.ConnectAsync(options); await client.SubscribeAsync(topic, MqttQualityOfServiceLevel.AtLeastOnce); diff --git a/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj b/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj index d7090834..a6f4566b 100644 --- a/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj +++ b/Tests/MQTTnet.TestApp.UniversalWindows/MQTTnet.TestApp.UniversalWindows.csproj @@ -142,7 +142,7 @@ - 6.1.4 + 6.1.5 3.0.0 diff --git a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml index 41617df8..c99ebd84 100644 --- a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml +++ b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml @@ -5,7 +5,6 @@ xmlns:d="http://schemas.microsoft.com/expression/blend/2008" xmlns:mc="http://schemas.openxmlformats.org/markup-compatibility/2006" xmlns:server="using:MQTTnet.Server" - xmlns:interop="using:Windows.UI.Xaml.Interop" d:DesignHeight="800" d:DesignWidth="800" mc:Ignorable="d"> @@ -19,26 +18,55 @@ - Server: - - Port: - - User: - - Password: - - ClientId: - - Clean session: - - Keep alive interval: - + + + + + + + + + + + + + + + + + - - TCP - WS - Use TLS - + Server: + + + Port: + + + Username: + + + Password: + + + Client ID: + + + Keep Alive interval: + + + Protocol: + + TCP + WS + Use TLS + + + + Clean session + Use managed client @@ -49,25 +77,46 @@ - Topic: - + + + + + + + + + + + + + + - Payload: - - - Text - Base64 - + Topic: + - Retain: - + Payload: + + + Payload format: + + Plain text + Base64 string + + + QoS level: + + 0 (At most once) + 1 (At least once) + 2 (Exactly once) + + + + Retain - QoS: - - 0 (At most once) - 1 (At least once) - 2 (Exactly once) - diff --git a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs index 387364a5..b557fe02 100644 --- a/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs +++ b/Tests/MQTTnet.TestApp.UniversalWindows/MainPage.xaml.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Concurrent; using System.Collections.ObjectModel; +using System.IO; using System.Text; using System.Threading.Tasks; using Windows.Security.Cryptography.Certificates; @@ -25,12 +26,15 @@ namespace MQTTnet.TestApp.UniversalWindows private readonly ObservableCollection _sessions = new ObservableCollection(); private IMqttClient _mqttClient; + private IManagedMqttClient _managedMqttClient; private IMqttServer _mqttServer; public MainPage() { InitializeComponent(); + ClientId.Text = Guid.NewGuid().ToString("D"); + MqttNetGlobalLogger.LogMessagePublished += OnTraceMessagePublished; } @@ -81,7 +85,10 @@ namespace MQTTnet.TestApp.UniversalWindows AllowUntrustedCertificates = true }; - var options = new MqttClientOptions { ClientId = ClientId.Text }; + var options = new MqttClientOptions + { + ClientId = ClientId.Text + }; if (UseTcp.IsChecked == true) { @@ -127,12 +134,28 @@ namespace MQTTnet.TestApp.UniversalWindows } var factory = new MqttFactory(); - _mqttClient = factory.CreateMqttClient(); - _mqttClient.ApplicationMessageReceived += OnApplicationMessageReceived; - _mqttClient.Connected += OnConnected; - _mqttClient.Disconnected += OnDisconnected; - await _mqttClient.ConnectAsync(options); + if (UseManagedClient.IsChecked == true) + { + _managedMqttClient = factory.CreateManagedMqttClient(); + _managedMqttClient.ApplicationMessageReceived += OnApplicationMessageReceived; + _managedMqttClient.Connected += OnConnected; + _managedMqttClient.Disconnected += OnDisconnected; + + await _managedMqttClient.StartAsync(new ManagedMqttClientOptions + { + ClientOptions = options + }); + } + else + { + _mqttClient = factory.CreateMqttClient(); + _mqttClient.ApplicationMessageReceived += OnApplicationMessageReceived; + _mqttClient.Connected += OnConnected; + _mqttClient.Disconnected += OnDisconnected; + + await _mqttClient.ConnectAsync(options); + } } catch (Exception exception) { @@ -171,11 +194,6 @@ namespace MQTTnet.TestApp.UniversalWindows private async void Publish(object sender, RoutedEventArgs e) { - if (_mqttClient == null) - { - return; - } - try { var qos = MqttQualityOfServiceLevel.AtMostOnce; @@ -190,7 +208,7 @@ namespace MQTTnet.TestApp.UniversalWindows } var payload = new byte[0]; - if (Text.IsChecked == true) + if (PlainText.IsChecked == true) { payload = Encoding.UTF8.GetBytes(Payload.Text); } @@ -207,7 +225,15 @@ namespace MQTTnet.TestApp.UniversalWindows .WithRetainFlag(Retain.IsChecked == true) .Build(); - await _mqttClient.PublishAsync(message); + if (_mqttClient != null) + { + await _mqttClient.PublishAsync(message); + } + + if (_managedMqttClient != null) + { + await _managedMqttClient.PublishAsync(message); + } } catch (Exception exception) { @@ -219,7 +245,19 @@ namespace MQTTnet.TestApp.UniversalWindows { try { - await _mqttClient.DisconnectAsync(); + if (_mqttClient != null) + { + await _mqttClient.DisconnectAsync(); + _mqttClient.Dispose(); + _mqttClient = null; + } + + if (_managedMqttClient != null) + { + await _managedMqttClient.StopAsync(); + _managedMqttClient.Dispose(); + _managedMqttClient = null; + } } catch (Exception exception) { @@ -239,11 +277,6 @@ namespace MQTTnet.TestApp.UniversalWindows private async void Subscribe(object sender, RoutedEventArgs e) { - if (_mqttClient == null) - { - return; - } - try { var qos = MqttQualityOfServiceLevel.AtMostOnce; @@ -257,7 +290,15 @@ namespace MQTTnet.TestApp.UniversalWindows qos = MqttQualityOfServiceLevel.ExactlyOnce; } - await _mqttClient.SubscribeAsync(new TopicFilter(SubscribeTopic.Text, qos)); + if (_mqttClient != null) + { + await _mqttClient.SubscribeAsync(new TopicFilter(SubscribeTopic.Text, qos)); + } + + if (_managedMqttClient != null) + { + await _managedMqttClient.SubscribeAsync(new TopicFilter(SubscribeTopic.Text, qos)); + } } catch (Exception exception) { @@ -267,14 +308,17 @@ namespace MQTTnet.TestApp.UniversalWindows private async void Unsubscribe(object sender, RoutedEventArgs e) { - if (_mqttClient == null) - { - return; - } - try { - await _mqttClient.UnsubscribeAsync(SubscribeTopic.Text); + if (_mqttClient != null) + { + await _mqttClient.UnsubscribeAsync(SubscribeTopic.Text); + } + + if (_managedMqttClient != null) + { + await _managedMqttClient.UnsubscribeAsync(SubscribeTopic.Text); + } } catch (Exception exception) { @@ -398,6 +442,8 @@ namespace MQTTnet.TestApp.UniversalWindows ListViewSessions.DataContext = _sessions; } + #region Wiki Code + private async Task WikiCode() { { @@ -632,5 +678,7 @@ namespace MQTTnet.TestApp.UniversalWindows await mqttClient.StartAsync(options); } } + + #endregion } }