From 0907eadb55f59107489e3d59209cd57d6bb87778 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sat, 9 Sep 2017 21:50:55 +0200 Subject: [PATCH 1/4] Make string format optional if no trace listener is attached --- MQTTnet.Core/Diagnostics/MqttTrace.cs | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/MQTTnet.Core/Diagnostics/MqttTrace.cs b/MQTTnet.Core/Diagnostics/MqttTrace.cs index 86f5b45a..c6104dc9 100644 --- a/MQTTnet.Core/Diagnostics/MqttTrace.cs +++ b/MQTTnet.Core/Diagnostics/MqttTrace.cs @@ -38,7 +38,14 @@ namespace MQTTnet.Core.Diagnostics private static void Publish(string source, MqttTraceLevel traceLevel, Exception exception, string message) { - TraceMessagePublished?.Invoke(null, new MqttTraceMessagePublishedEventArgs(Environment.CurrentManagedThreadId, source, traceLevel, message, exception)); + var handler = TraceMessagePublished; + if (handler == null) + { + return; + } + + message = string.Format(message, 1); + handler.Invoke(null, new MqttTraceMessagePublishedEventArgs(Environment.CurrentManagedThreadId, source, traceLevel, message, exception)); } } } From 04036cf63a661fc9603597248fe409716128b2f0 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sat, 9 Sep 2017 22:29:14 +0200 Subject: [PATCH 2/4] Prepare version information for the next release --- Build/MQTTnet.nuspec | 4 ++-- Frameworks/MQTTnet.NetFramework/Properties/AssemblyInfo.cs | 4 ++-- Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj | 4 ++-- .../MQTTnet.UniversalWindows/Properties/AssemblyInfo.cs | 4 ++-- MQTTnet.Core/MQTTnet.Core.csproj | 4 ++-- 5 files changed, 10 insertions(+), 10 deletions(-) diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 8bafd6d4..6da1a6fe 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -2,7 +2,7 @@ MQTTnet - 2.2.1 + 2.3.0 Christian Kratky Christian Kratky https://github.com/chkr1011/MQTTnet/blob/master/LICENSE @@ -10,7 +10,7 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false MQTTnet is a .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker). - * [Server] Fixed a bug in connection management (Thanks to rydergillen-compacSort) + * Copyright Christian Kratky 2016-2017 MQTT MQTTClient MQTTServer MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Queue Hardware Arduino diff --git a/Frameworks/MQTTnet.NetFramework/Properties/AssemblyInfo.cs b/Frameworks/MQTTnet.NetFramework/Properties/AssemblyInfo.cs index e04de7b4..c6f89a7f 100644 --- a/Frameworks/MQTTnet.NetFramework/Properties/AssemblyInfo.cs +++ b/Frameworks/MQTTnet.NetFramework/Properties/AssemblyInfo.cs @@ -11,5 +11,5 @@ using System.Runtime.InteropServices; [assembly: AssemblyCulture("")] [assembly: ComVisible(false)] [assembly: Guid("a480ef90-0eaa-4d9a-b271-47a9c47f6f7d")] -[assembly: AssemblyVersion("2.2.1.0")] -[assembly: AssemblyFileVersion("2.2.1.0")] \ No newline at end of file +[assembly: AssemblyVersion("2.3.0.0")] +[assembly: AssemblyFileVersion("2.3.0.0")] \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj b/Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj index de3a2894..25aaef06 100644 --- a/Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj +++ b/Frameworks/MQTTnet.NetStandard/MQTTnet.Netstandard.csproj @@ -4,8 +4,8 @@ netstandard1.3 MQTTnet MQTTnet - 2.2.1.0 - 2.2.1.0 + 2.3.0.0 + 2.3.0.0 0.0.0.0 diff --git a/Frameworks/MQTTnet.UniversalWindows/Properties/AssemblyInfo.cs b/Frameworks/MQTTnet.UniversalWindows/Properties/AssemblyInfo.cs index 1ee6ca16..cffa1cfd 100644 --- a/Frameworks/MQTTnet.UniversalWindows/Properties/AssemblyInfo.cs +++ b/Frameworks/MQTTnet.UniversalWindows/Properties/AssemblyInfo.cs @@ -10,5 +10,5 @@ using System.Runtime.InteropServices; [assembly: AssemblyTrademark("")] [assembly: AssemblyCulture("")] [assembly: ComVisible(false)] -[assembly: AssemblyVersion("2.2.1.0")] -[assembly: AssemblyFileVersion("2.2.1.0")] \ No newline at end of file +[assembly: AssemblyVersion("2.3.0.0")] +[assembly: AssemblyFileVersion("2.3.0.0")] \ No newline at end of file diff --git a/MQTTnet.Core/MQTTnet.Core.csproj b/MQTTnet.Core/MQTTnet.Core.csproj index e76ec3ac..480b4c7e 100644 --- a/MQTTnet.Core/MQTTnet.Core.csproj +++ b/MQTTnet.Core/MQTTnet.Core.csproj @@ -16,8 +16,8 @@ - 2.2.1.0 - 2.2.1.0 + 2.3.0.0 + 2.3.0.0 From 8aec1583df007287f82cea9ec5d5f98c99f13d4f Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sat, 9 Sep 2017 22:40:37 +0200 Subject: [PATCH 3/4] Performance improvements in area of logging and async/await --- .../Implementations/MqttServerAdapter.cs | 6 +- .../Implementations/MqttTcpChannel.cs | 4 +- .../Implementations/MqttTcpChannel.cs | 4 +- .../MqttChannelCommunicationAdapter.cs | 24 +++--- MQTTnet.Core/Client/MqttClient.cs | 65 ++++++++------- MQTTnet.Core/Diagnostics/MqttTrace.cs | 41 ++++++---- MQTTnet.Core/Serializer/MqttPacketReader.cs | 37 ++++----- .../Serializer/MqttPacketSerializer.cs | 82 +++++++++---------- MQTTnet.Core/Server/MqttClientSession.cs | 6 +- .../Server/MqttClientSessionsManager.cs | 6 +- MQTTnet.Core/Server/MqttServer.cs | 2 +- .../TestMqttCommunicationAdapter.cs | 8 +- 12 files changed, 153 insertions(+), 132 deletions(-) diff --git a/Frameworks/MQTTnet.NetFramework/Implementations/MqttServerAdapter.cs b/Frameworks/MQTTnet.NetFramework/Implementations/MqttServerAdapter.cs index faa5eff4..92e071e5 100644 --- a/Frameworks/MQTTnet.NetFramework/Implementations/MqttServerAdapter.cs +++ b/Frameworks/MQTTnet.NetFramework/Implementations/MqttServerAdapter.cs @@ -85,7 +85,7 @@ namespace MQTTnet.Implementations { try { - var clientSocket = await Task.Factory.FromAsync(_defaultEndpointSocket.BeginAccept, _defaultEndpointSocket.EndAccept, null).ConfigureAwait(false); + var clientSocket = await Task.Factory.FromAsync(_defaultEndpointSocket.BeginAccept, _defaultEndpointSocket.EndAccept, null); var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, null), new MqttPacketSerializer()); ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter)); } @@ -105,10 +105,10 @@ namespace MQTTnet.Implementations { try { - var clientSocket = await Task.Factory.FromAsync(_tlsEndpointSocket.BeginAccept, _tlsEndpointSocket.EndAccept, null).ConfigureAwait(false); + var clientSocket = await Task.Factory.FromAsync(_tlsEndpointSocket.BeginAccept, _tlsEndpointSocket.EndAccept, null); var sslStream = new SslStream(new NetworkStream(clientSocket)); - await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, SslProtocols.Tls12, false).ConfigureAwait(false); + await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, SslProtocols.Tls12, false); var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketSerializer()); ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter)); diff --git a/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs b/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs index f5dd3268..c9c080d2 100644 --- a/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs +++ b/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs @@ -35,12 +35,12 @@ namespace MQTTnet.Implementations _socket = new Socket(SocketType.Stream, ProtocolType.Tcp); } - await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, options.Server, options.GetPort(), null).ConfigureAwait(false); + await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, options.Server, options.GetPort(), null); if (options.TlsOptions.UseTls) { _sslStream = new SslStream(new NetworkStream(_socket, true)); - await _sslStream.AuthenticateAsClientAsync(options.Server, LoadCertificates(options), SslProtocols.Tls12, options.TlsOptions.CheckCertificateRevocation).ConfigureAwait(false); + await _sslStream.AuthenticateAsClientAsync(options.Server, LoadCertificates(options), SslProtocols.Tls12, options.TlsOptions.CheckCertificateRevocation); } } catch (SocketException exception) diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs index 75fedba1..a4247b09 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs @@ -35,12 +35,12 @@ namespace MQTTnet.Implementations _socket = new Socket(SocketType.Stream, ProtocolType.Tcp); } - await _socket.ConnectAsync(options.Server, options.GetPort()).ConfigureAwait(false); + await _socket.ConnectAsync(options.Server, options.GetPort()); if (options.TlsOptions.UseTls) { _sslStream = new SslStream(new NetworkStream(_socket, true)); - await _sslStream.AuthenticateAsClientAsync(options.Server, LoadCertificates(options), SslProtocols.Tls12, options.TlsOptions.CheckCertificateRevocation).ConfigureAwait(false); + await _sslStream.AuthenticateAsClientAsync(options.Server, LoadCertificates(options), SslProtocols.Tls12, options.TlsOptions.CheckCertificateRevocation); } } catch (SocketException exception) diff --git a/MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs b/MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs index fcea65ac..50421678 100644 --- a/MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs +++ b/MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs @@ -21,21 +21,21 @@ namespace MQTTnet.Core.Adapter public IMqttPacketSerializer PacketSerializer { get; } - public async Task ConnectAsync(MqttClientOptions options, TimeSpan timeout) + public Task ConnectAsync(MqttClientOptions options, TimeSpan timeout) { - await ExecuteWithTimeoutAsync(_channel.ConnectAsync(options), timeout).ConfigureAwait(false); + return ExecuteWithTimeoutAsync(_channel.ConnectAsync(options), timeout); } - public async Task DisconnectAsync() + public Task DisconnectAsync() { - await _channel.DisconnectAsync().ConfigureAwait(false); + return _channel.DisconnectAsync(); } - public async Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout) + public Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout) { - MqttTrace.Information(nameof(MqttChannelCommunicationAdapter), $"TX >>> {packet} [Timeout={timeout}]"); + MqttTrace.Information(nameof(MqttChannelCommunicationAdapter), "TX >>> {0} [Timeout={1}]", packet, timeout); - await ExecuteWithTimeoutAsync(PacketSerializer.SerializeAsync(packet, _channel), timeout).ConfigureAwait(false); + return ExecuteWithTimeoutAsync(PacketSerializer.SerializeAsync(packet, _channel), timeout); } public async Task ReceivePacketAsync(TimeSpan timeout) @@ -43,11 +43,11 @@ namespace MQTTnet.Core.Adapter MqttBasePacket packet; if (timeout > TimeSpan.Zero) { - packet = await ExecuteWithTimeoutAsync(PacketSerializer.DeserializeAsync(_channel), timeout).ConfigureAwait(false); + packet = await ExecuteWithTimeoutAsync(PacketSerializer.DeserializeAsync(_channel), timeout); } else { - packet = await PacketSerializer.DeserializeAsync(_channel).ConfigureAwait(false); + packet = await PacketSerializer.DeserializeAsync(_channel); } if (packet == null) @@ -55,14 +55,14 @@ namespace MQTTnet.Core.Adapter throw new MqttProtocolViolationException("Received malformed packet."); } - MqttTrace.Information(nameof(MqttChannelCommunicationAdapter), $"RX <<< {packet}"); + MqttTrace.Information(nameof(MqttChannelCommunicationAdapter), "RX <<< {0}", packet); return packet; } private static async Task ExecuteWithTimeoutAsync(Task task, TimeSpan timeout) { var timeoutTask = Task.Delay(timeout); - if (await Task.WhenAny(timeoutTask, task).ConfigureAwait(false) == timeoutTask) + if (await Task.WhenAny(timeoutTask, task) == timeoutTask) { throw new MqttCommunicationTimedOutException(); } @@ -78,7 +78,7 @@ namespace MQTTnet.Core.Adapter private static async Task ExecuteWithTimeoutAsync(Task task, TimeSpan timeout) { var timeoutTask = Task.Delay(timeout); - if (await Task.WhenAny(timeoutTask, task).ConfigureAwait(false) == timeoutTask) + if (await Task.WhenAny(timeoutTask, task) == timeoutTask) { throw new MqttCommunicationTimedOutException(); } diff --git a/MQTTnet.Core/Client/MqttClient.cs b/MQTTnet.Core/Client/MqttClient.cs index efd794d2..518eb376 100644 --- a/MQTTnet.Core/Client/MqttClient.cs +++ b/MQTTnet.Core/Client/MqttClient.cs @@ -53,7 +53,7 @@ namespace MQTTnet.Core.Client { _disconnectedEventSuspended = false; - await _adapter.ConnectAsync(_options, _options.DefaultCommunicationTimeout).ConfigureAwait(false); + await _adapter.ConnectAsync(_options, _options.DefaultCommunicationTimeout); MqttTrace.Verbose(nameof(MqttClient), "Connection with server established."); @@ -73,10 +73,10 @@ namespace MQTTnet.Core.Client StartReceivePackets(); - var response = await SendAndReceiveAsync(connectPacket).ConfigureAwait(false); + var response = await SendAndReceiveAsync(connectPacket); if (response.ConnectReturnCode != MqttConnectReturnCode.ConnectionAccepted) { - await DisconnectInternalAsync().ConfigureAwait(false); + await DisconnectInternalAsync(); throw new MqttConnectingFailedException(response.ConnectReturnCode); } @@ -92,7 +92,7 @@ namespace MQTTnet.Core.Client } catch (Exception) { - await DisconnectInternalAsync().ConfigureAwait(false); + await DisconnectInternalAsync(); throw; } } @@ -101,11 +101,11 @@ namespace MQTTnet.Core.Client { try { - await SendAsync(new MqttDisconnectPacket()).ConfigureAwait(false); + await SendAsync(new MqttDisconnectPacket()); } finally { - await DisconnectInternalAsync().ConfigureAwait(false); + await DisconnectInternalAsync(); } } @@ -146,7 +146,7 @@ namespace MQTTnet.Core.Client return Unsubscribe(topicFilters.ToList()); } - public async Task Unsubscribe(IList topicFilters) + public Task Unsubscribe(IList topicFilters) { if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); if (!topicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.10.3-2]."); @@ -158,10 +158,10 @@ namespace MQTTnet.Core.Client TopicFilters = topicFilters }; - await SendAndReceiveAsync(unsubscribePacket).ConfigureAwait(false); + return SendAndReceiveAsync(unsubscribePacket); } - public async Task PublishAsync(MqttApplicationMessage applicationMessage) + public Task PublishAsync(MqttApplicationMessage applicationMessage) { if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); ThrowIfNotConnected(); @@ -171,19 +171,28 @@ namespace MQTTnet.Core.Client if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce) { // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier] - await SendAsync(publishPacket).ConfigureAwait(false); + return SendAsync(publishPacket); } - else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) + + if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) { publishPacket.PacketIdentifier = GetNewPacketIdentifier(); - await SendAndReceiveAsync(publishPacket).ConfigureAwait(false); + return SendAndReceiveAsync(publishPacket); } - else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) + + if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) { publishPacket.PacketIdentifier = GetNewPacketIdentifier(); - var pubRecPacket = await SendAndReceiveAsync(publishPacket).ConfigureAwait(false); - await SendAndReceiveAsync(pubRecPacket.CreateResponse()).ConfigureAwait(false); + return PublishExactlyOncePacketAsync(publishPacket); } + + throw new InvalidOperationException(); + } + + private async Task PublishExactlyOncePacketAsync(MqttBasePacket publishPacket) + { + var pubRecPacket = await SendAndReceiveAsync(publishPacket); + await SendAndReceiveAsync(pubRecPacket.CreateResponse()); } private void ThrowIfNotConnected() @@ -195,7 +204,7 @@ namespace MQTTnet.Core.Client { try { - await _adapter.DisconnectAsync().ConfigureAwait(false); + await _adapter.DisconnectAsync(); } catch (Exception exception) { @@ -294,14 +303,14 @@ namespace MQTTnet.Core.Client throw new MqttCommunicationException("Received a not supported QoS level."); } - private async Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket) + private Task ProcessReceivedPubRelPacket(MqttPubRelPacket pubRelPacket) { lock (_unacknowledgedPublishPackets) { _unacknowledgedPublishPackets.Remove(pubRelPacket.PacketIdentifier); } - await SendAsync(pubRelPacket.CreateResponse()).ConfigureAwait(false); + return SendAsync(pubRelPacket.CreateResponse()); } private Task SendAsync(MqttBasePacket packet) @@ -326,8 +335,8 @@ namespace MQTTnet.Core.Client return pi1.PacketIdentifier == pi2.PacketIdentifier; } - await _adapter.SendPacketAsync(requestPacket, _options.DefaultCommunicationTimeout).ConfigureAwait(false); - return (TResponsePacket)await _packetDispatcher.WaitForPacketAsync(ResponsePacketSelector, _options.DefaultCommunicationTimeout).ConfigureAwait(false); + await _adapter.SendPacketAsync(requestPacket, _options.DefaultCommunicationTimeout); + return (TResponsePacket)await _packetDispatcher.WaitForPacketAsync(ResponsePacketSelector, _options.DefaultCommunicationTimeout); } private ushort GetNewPacketIdentifier() @@ -343,19 +352,19 @@ namespace MQTTnet.Core.Client { while (!cancellationToken.IsCancellationRequested) { - await Task.Delay(_options.KeepAlivePeriod, cancellationToken).ConfigureAwait(false); - await SendAndReceiveAsync(new MqttPingReqPacket()).ConfigureAwait(false); + await Task.Delay(_options.KeepAlivePeriod, cancellationToken); + await SendAndReceiveAsync(new MqttPingReqPacket()); } } catch (MqttCommunicationException exception) { MqttTrace.Warning(nameof(MqttClient), exception, "MQTT communication error while receiving packets."); - await DisconnectInternalAsync().ConfigureAwait(false); + await DisconnectInternalAsync(); } catch (Exception exception) { MqttTrace.Warning(nameof(MqttClient), exception, "Error while sending/receiving keep alive packets."); - await DisconnectInternalAsync().ConfigureAwait(false); + await DisconnectInternalAsync(); } finally { @@ -370,8 +379,8 @@ namespace MQTTnet.Core.Client { while (!cancellationToken.IsCancellationRequested) { - var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero).ConfigureAwait(false); - MqttTrace.Information(nameof(MqttClient), $"Received <<< {packet}"); + var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero); + MqttTrace.Information(nameof(MqttClient), "Received <<< {0}", packet); StartProcessReceivedPacket(packet, cancellationToken); } @@ -379,12 +388,12 @@ namespace MQTTnet.Core.Client catch (MqttCommunicationException exception) { MqttTrace.Warning(nameof(MqttClient), exception, "MQTT communication exception while receiving packets."); - await DisconnectInternalAsync().ConfigureAwait(false); + await DisconnectInternalAsync(); } catch (Exception exception) { MqttTrace.Error(nameof(MqttClient), exception, "Unhandled exception while receiving packets."); - await DisconnectInternalAsync().ConfigureAwait(false); + await DisconnectInternalAsync(); } finally { diff --git a/MQTTnet.Core/Diagnostics/MqttTrace.cs b/MQTTnet.Core/Diagnostics/MqttTrace.cs index c6104dc9..1036028f 100644 --- a/MQTTnet.Core/Diagnostics/MqttTrace.cs +++ b/MQTTnet.Core/Diagnostics/MqttTrace.cs @@ -1,4 +1,5 @@ using System; +using System.Linq; namespace MQTTnet.Core.Diagnostics { @@ -6,37 +7,37 @@ namespace MQTTnet.Core.Diagnostics { public static event EventHandler TraceMessagePublished; - public static void Verbose(string source, string message) + public static void Verbose(string source, string message, params object[] parameters) { - Publish(source, MqttTraceLevel.Verbose, null, message); + Publish(source, MqttTraceLevel.Verbose, null, message, parameters); } - public static void Information(string source, string message) + public static void Information(string source, string message, params object[] parameters) { - Publish(source, MqttTraceLevel.Information, null, message); + Publish(source, MqttTraceLevel.Information, null, message, parameters); } - public static void Warning(string source, string message) + public static void Warning(string source, string message, params object[] parameters) { - Publish(source, MqttTraceLevel.Warning, null, message); + Publish(source, MqttTraceLevel.Warning, null, message, parameters); } - public static void Warning(string source, Exception exception, string message) + public static void Warning(string source, Exception exception, string message, params object[] parameters) { - Publish(source, MqttTraceLevel.Warning, exception, message); + Publish(source, MqttTraceLevel.Warning, exception, message, parameters); } - public static void Error(string source, string message) + public static void Error(string source, string message, params object[] parameters) { - Publish(source, MqttTraceLevel.Error, null, message); + Publish(source, MqttTraceLevel.Error, null, message, parameters); } - public static void Error(string source, Exception exception, string message) + public static void Error(string source, Exception exception, string message, params object[] parameters) { - Publish(source, MqttTraceLevel.Error, exception, message); + Publish(source, MqttTraceLevel.Error, exception, message, parameters); } - private static void Publish(string source, MqttTraceLevel traceLevel, Exception exception, string message) + private static void Publish(string source, MqttTraceLevel traceLevel, Exception exception, string message, params object[] parameters) { var handler = TraceMessagePublished; if (handler == null) @@ -44,7 +45,19 @@ namespace MQTTnet.Core.Diagnostics return; } - message = string.Format(message, 1); + if (parameters?.Length > 0) + { + try + { + message = string.Format(message, parameters); + } + catch (Exception formatException) + { + Error(nameof(MqttTrace), formatException, "Error while tracing message: " + message); + return; + } + } + handler.Invoke(null, new MqttTraceMessagePublishedEventArgs(Environment.CurrentManagedThreadId, source, traceLevel, message, exception)); } } diff --git a/MQTTnet.Core/Serializer/MqttPacketReader.cs b/MQTTnet.Core/Serializer/MqttPacketReader.cs index 349c1370..6a31fcad 100644 --- a/MQTTnet.Core/Serializer/MqttPacketReader.cs +++ b/MQTTnet.Core/Serializer/MqttPacketReader.cs @@ -29,7 +29,7 @@ namespace MQTTnet.Core.Serializer public async Task ReadToEndAsync() { await ReadFixedHeaderAsync(); - await ReadRemainingLengthAsync().ConfigureAwait(false); + await ReadRemainingLengthAsync(); if (_remainingLength == 0) { @@ -37,20 +37,20 @@ namespace MQTTnet.Core.Serializer } var buffer = new byte[_remainingLength]; - await ReadFromSourceAsync(buffer).ConfigureAwait(false); + await ReadFromSourceAsync(buffer); _remainingData.Write(buffer, 0, buffer.Length); _remainingData.Position = 0; } - public async Task ReadRemainingDataByteAsync() + public byte ReadRemainingDataByte() { - return (await ReadRemainingDataAsync(1).ConfigureAwait(false))[0]; + return ReadRemainingData(1)[0]; } - public async Task ReadRemainingDataUShortAsync() + public ushort ReadRemainingDataUShort() { - var buffer = await ReadRemainingDataAsync(2).ConfigureAwait(false); + var buffer = ReadRemainingData(2); var temp = buffer[0]; buffer[0] = buffer[1]; @@ -59,28 +59,27 @@ namespace MQTTnet.Core.Serializer return BitConverter.ToUInt16(buffer, 0); } - public async Task ReadRemainingDataStringWithLengthPrefixAsync() + public string ReadRemainingDataStringWithLengthPrefix() { - var buffer = await ReadRemainingDataWithLengthPrefixAsync(); + var buffer = ReadRemainingDataWithLengthPrefix(); return Encoding.UTF8.GetString(buffer, 0, buffer.Length); } - public async Task ReadRemainingDataWithLengthPrefixAsync() + public byte[] ReadRemainingDataWithLengthPrefix() { - var length = await ReadRemainingDataUShortAsync(); - return await ReadRemainingDataAsync(length).ConfigureAwait(false); + var length = ReadRemainingDataUShort(); + return ReadRemainingData(length); } - public Task ReadRemainingDataAsync() + public byte[] ReadRemainingData() { - return ReadRemainingDataAsync(_remainingLength - (int)_remainingData.Position); + return ReadRemainingData(_remainingLength - (int)_remainingData.Position); } - public async Task ReadRemainingDataAsync(int length) + public byte[] ReadRemainingData(int length) { var buffer = new byte[length]; - await _remainingData.ReadAsync(buffer, 0, buffer.Length).ConfigureAwait(false); - + _remainingData.Read(buffer, 0, buffer.Length); return buffer; } @@ -92,7 +91,7 @@ namespace MQTTnet.Core.Serializer byte encodedByte; do { - encodedByte = await ReadStreamByteAsync().ConfigureAwait(false); + encodedByte = await ReadStreamByteAsync(); value += (encodedByte & 127) * multiplier; multiplier *= 128; if (multiplier > 128 * 128 * 128) @@ -119,13 +118,13 @@ namespace MQTTnet.Core.Serializer private async Task ReadStreamByteAsync() { var buffer = new byte[1]; - await ReadFromSourceAsync(buffer).ConfigureAwait(false); + await ReadFromSourceAsync(buffer); return buffer[0]; } private async Task ReadFixedHeaderAsync() { - FixedHeader = await ReadStreamByteAsync().ConfigureAwait(false); + FixedHeader = await ReadStreamByteAsync(); var byteReader = new ByteReader(FixedHeader); byteReader.Read(4); diff --git a/MQTTnet.Core/Serializer/MqttPacketSerializer.cs b/MQTTnet.Core/Serializer/MqttPacketSerializer.cs index 7d6bb815..38017b63 100644 --- a/MQTTnet.Core/Serializer/MqttPacketSerializer.cs +++ b/MQTTnet.Core/Serializer/MqttPacketSerializer.cs @@ -106,12 +106,12 @@ namespace MQTTnet.Core.Serializer { case MqttControlPacketType.Connect: { - return await DeserializeConnectAsync(mqttPacketReader).ConfigureAwait(false); + return DeserializeConnect(mqttPacketReader); } case MqttControlPacketType.ConnAck: { - return await DeserializeConnAck(mqttPacketReader).ConfigureAwait(false); + return DeserializeConnAck(mqttPacketReader); } case MqttControlPacketType.Disconnect: @@ -121,14 +121,14 @@ namespace MQTTnet.Core.Serializer case MqttControlPacketType.Publish: { - return await DeserializePublishAsync(mqttPacketReader).ConfigureAwait(false); + return DeserializePublish(mqttPacketReader); } case MqttControlPacketType.PubAck: { return new MqttPubAckPacket { - PacketIdentifier = await mqttPacketReader.ReadRemainingDataUShortAsync().ConfigureAwait(false) + PacketIdentifier = mqttPacketReader.ReadRemainingDataUShort() }; } @@ -136,7 +136,7 @@ namespace MQTTnet.Core.Serializer { return new MqttPubRecPacket { - PacketIdentifier = await mqttPacketReader.ReadRemainingDataUShortAsync().ConfigureAwait(false) + PacketIdentifier = mqttPacketReader.ReadRemainingDataUShort() }; } @@ -144,7 +144,7 @@ namespace MQTTnet.Core.Serializer { return new MqttPubRelPacket { - PacketIdentifier = await mqttPacketReader.ReadRemainingDataUShortAsync().ConfigureAwait(false) + PacketIdentifier = mqttPacketReader.ReadRemainingDataUShort() }; } @@ -152,7 +152,7 @@ namespace MQTTnet.Core.Serializer { return new MqttPubCompPacket { - PacketIdentifier = await mqttPacketReader.ReadRemainingDataUShortAsync().ConfigureAwait(false) + PacketIdentifier = mqttPacketReader.ReadRemainingDataUShort() }; } @@ -168,24 +168,24 @@ namespace MQTTnet.Core.Serializer case MqttControlPacketType.Subscribe: { - return await DeserializeSubscribeAsync(mqttPacketReader).ConfigureAwait(false); + return DeserializeSubscribe(mqttPacketReader); } case MqttControlPacketType.SubAck: { - return await DeserializeSubAck(mqttPacketReader).ConfigureAwait(false); + return DeserializeSubAck(mqttPacketReader); } case MqttControlPacketType.Unsubscibe: { - return await DeserializeUnsubscribeAsync(mqttPacketReader).ConfigureAwait(false); + return DeserializeUnsubscribe(mqttPacketReader); } case MqttControlPacketType.UnsubAck: { return new MqttUnsubAckPacket { - PacketIdentifier = await mqttPacketReader.ReadRemainingDataUShortAsync().ConfigureAwait(false) + PacketIdentifier = mqttPacketReader.ReadRemainingDataUShort() }; } @@ -197,51 +197,51 @@ namespace MQTTnet.Core.Serializer } } - private static async Task DeserializeUnsubscribeAsync(MqttPacketReader reader) + private static MqttBasePacket DeserializeUnsubscribe(MqttPacketReader reader) { var packet = new MqttUnsubscribePacket { - PacketIdentifier = await reader.ReadRemainingDataUShortAsync().ConfigureAwait(false), + PacketIdentifier = reader.ReadRemainingDataUShort(), }; while (!reader.EndOfRemainingData) { - packet.TopicFilters.Add(await reader.ReadRemainingDataStringWithLengthPrefixAsync().ConfigureAwait(false)); + packet.TopicFilters.Add(reader.ReadRemainingDataStringWithLengthPrefix()); } return packet; } - private static async Task DeserializeSubscribeAsync(MqttPacketReader reader) + private static MqttBasePacket DeserializeSubscribe(MqttPacketReader reader) { var packet = new MqttSubscribePacket { - PacketIdentifier = await reader.ReadRemainingDataUShortAsync().ConfigureAwait(false), + PacketIdentifier = reader.ReadRemainingDataUShort() }; while (!reader.EndOfRemainingData) { packet.TopicFilters.Add(new TopicFilter( - await reader.ReadRemainingDataStringWithLengthPrefixAsync(), - (MqttQualityOfServiceLevel)await reader.ReadRemainingDataByteAsync().ConfigureAwait(false))); + reader.ReadRemainingDataStringWithLengthPrefix(), + (MqttQualityOfServiceLevel)reader.ReadRemainingDataByte())); } return packet; } - private static async Task DeserializePublishAsync(MqttPacketReader reader) + private static MqttBasePacket DeserializePublish(MqttPacketReader reader) { var fixedHeader = new ByteReader(reader.FixedHeader); var retain = fixedHeader.Read(); var qualityOfServiceLevel = (MqttQualityOfServiceLevel)fixedHeader.Read(2); var dup = fixedHeader.Read(); - var topic = await reader.ReadRemainingDataStringWithLengthPrefixAsync().ConfigureAwait(false); + var topic = reader.ReadRemainingDataStringWithLengthPrefix(); ushort packetIdentifier = 0; if (qualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) { - packetIdentifier = await reader.ReadRemainingDataUShortAsync().ConfigureAwait(false); + packetIdentifier = reader.ReadRemainingDataUShort(); } var packet = new MqttPublishPacket @@ -250,22 +250,22 @@ namespace MQTTnet.Core.Serializer QualityOfServiceLevel = qualityOfServiceLevel, Dup = dup, Topic = topic, - Payload = await reader.ReadRemainingDataAsync().ConfigureAwait(false), + Payload = reader.ReadRemainingData(), PacketIdentifier = packetIdentifier }; return packet; } - private static async Task DeserializeConnectAsync(MqttPacketReader reader) + private static MqttBasePacket DeserializeConnect(MqttPacketReader reader) { - await reader.ReadRemainingDataAsync(2).ConfigureAwait(false); // Skip 2 bytes + reader.ReadRemainingData(2); // Skip 2 bytes MqttProtocolVersion protocolVersion; - var protocolName = await reader.ReadRemainingDataAsync(4).ConfigureAwait(false); + var protocolName = reader.ReadRemainingData(4); if (protocolName.SequenceEqual(ProtocolVersionV310Name)) { - await reader.ReadRemainingDataAsync(2).ConfigureAwait(false); + reader.ReadRemainingData(2); protocolVersion = MqttProtocolVersion.V310; } else if (protocolName.SequenceEqual(ProtocolVersionV311Name)) @@ -277,8 +277,8 @@ namespace MQTTnet.Core.Serializer throw new MqttProtocolViolationException("Protocol name is not supported."); } - var protocolLevel = await reader.ReadRemainingDataByteAsync().ConfigureAwait(false); - var connectFlags = await reader.ReadRemainingDataByteAsync().ConfigureAwait(false); + var protocolLevel = reader.ReadRemainingDataByte(); + var connectFlags = reader.ReadRemainingDataByte(); var connectFlagsReader = new ByteReader(connectFlags); connectFlagsReader.Read(); // Reserved. @@ -295,51 +295,51 @@ namespace MQTTnet.Core.Serializer var passwordFlag = connectFlagsReader.Read(); var usernameFlag = connectFlagsReader.Read(); - packet.KeepAlivePeriod = await reader.ReadRemainingDataUShortAsync().ConfigureAwait(false); - packet.ClientId = await reader.ReadRemainingDataStringWithLengthPrefixAsync().ConfigureAwait(false); + packet.KeepAlivePeriod = reader.ReadRemainingDataUShort(); + packet.ClientId = reader.ReadRemainingDataStringWithLengthPrefix(); if (willFlag) { packet.WillMessage = new MqttApplicationMessage( - await reader.ReadRemainingDataStringWithLengthPrefixAsync().ConfigureAwait(false), - await reader.ReadRemainingDataWithLengthPrefixAsync().ConfigureAwait(false), + reader.ReadRemainingDataStringWithLengthPrefix(), + reader.ReadRemainingDataWithLengthPrefix(), (MqttQualityOfServiceLevel)willQoS, willRetain); } if (usernameFlag) { - packet.Username = await reader.ReadRemainingDataStringWithLengthPrefixAsync().ConfigureAwait(false); + packet.Username = reader.ReadRemainingDataStringWithLengthPrefix(); } if (passwordFlag) { - packet.Password = await reader.ReadRemainingDataStringWithLengthPrefixAsync().ConfigureAwait(false); + packet.Password = reader.ReadRemainingDataStringWithLengthPrefix(); } ValidateConnectPacket(packet); return packet; } - private static async Task DeserializeSubAck(MqttPacketReader reader) + private static MqttBasePacket DeserializeSubAck(MqttPacketReader reader) { var packet = new MqttSubAckPacket { - PacketIdentifier = await reader.ReadRemainingDataUShortAsync().ConfigureAwait(false) + PacketIdentifier = reader.ReadRemainingDataUShort() }; while (!reader.EndOfRemainingData) { - packet.SubscribeReturnCodes.Add((MqttSubscribeReturnCode)await reader.ReadRemainingDataByteAsync().ConfigureAwait(false)); + packet.SubscribeReturnCodes.Add((MqttSubscribeReturnCode)reader.ReadRemainingDataByte()); } return packet; } - private static async Task DeserializeConnAck(MqttPacketReader reader) + private static MqttBasePacket DeserializeConnAck(MqttPacketReader reader) { - var variableHeader1 = await reader.ReadRemainingDataByteAsync().ConfigureAwait(false); - var variableHeader2 = await reader.ReadRemainingDataByteAsync().ConfigureAwait(false); + var variableHeader1 = reader.ReadRemainingDataByte(); + var variableHeader2 = reader.ReadRemainingDataByte(); var packet = new MqttConnAckPacket { @@ -457,7 +457,7 @@ namespace MQTTnet.Core.Serializer output.Write(packet.PacketIdentifier); output.InjectFixedHeader(MqttControlPacketType.PubRel, 0x02); - await output.WriteToAsync(destination).ConfigureAwait(false); + await output.WriteToAsync(destination); } } diff --git a/MQTTnet.Core/Server/MqttClientSession.cs b/MQTTnet.Core/Server/MqttClientSession.cs index bd99fe31..0b69425b 100644 --- a/MQTTnet.Core/Server/MqttClientSession.cs +++ b/MQTTnet.Core/Server/MqttClientSession.cs @@ -63,7 +63,7 @@ namespace MQTTnet.Core.Server } catch (Exception exception) { - MqttTrace.Error(nameof(MqttClientSession), exception, $"Client '{_identifier}': Unhandled exception while processing client packets."); + MqttTrace.Error(nameof(MqttClientSession), exception, "Client '{0}': Unhandled exception while processing client packets.", _identifier); } finally { @@ -90,7 +90,7 @@ namespace MQTTnet.Core.Server } _messageQueue.Enqueue(publishPacket); - MqttTrace.Verbose(nameof(MqttClientSession), $"Client '{_identifier}: Enqueued pending publish packet."); + MqttTrace.Verbose(nameof(MqttClientSession), "Client '{0}': Enqueued pending publish packet.", _identifier); } public void Dispose() @@ -143,7 +143,7 @@ namespace MQTTnet.Core.Server return Task.FromResult((object)null); } - MqttTrace.Warning(nameof(MqttClientSession), $"Client '{_identifier}': Received not supported packet ({packet}). Closing connection."); + MqttTrace.Warning(nameof(MqttClientSession), "Client '{0}': Received not supported packet ({1}). Closing connection.", _identifier, packet); _cancellationTokenSource.Cancel(); return Task.FromResult((object)null); diff --git a/MQTTnet.Core/Server/MqttClientSessionsManager.cs b/MQTTnet.Core/Server/MqttClientSessionsManager.cs index 6165143d..79823e78 100644 --- a/MQTTnet.Core/Server/MqttClientSessionsManager.cs +++ b/MQTTnet.Core/Server/MqttClientSessionsManager.cs @@ -111,11 +111,11 @@ namespace MQTTnet.Core.Server _clientSessions.Remove(connectPacket.ClientId); clientSession.Dispose(); clientSession = null; - MqttTrace.Verbose(nameof(MqttClientSessionsManager), $"Disposed existing session of client '{connectPacket.ClientId}'."); + MqttTrace.Verbose(nameof(MqttClientSessionsManager), "Disposed existing session of client '{0}'.", connectPacket.ClientId); } else { - MqttTrace.Verbose(nameof(MqttClientSessionsManager), $"Reusing existing session of client '{connectPacket.ClientId}'."); + MqttTrace.Verbose(nameof(MqttClientSessionsManager), "Reusing existing session of client '{0}'.", connectPacket.ClientId); } } @@ -127,7 +127,7 @@ namespace MQTTnet.Core.Server clientSession = new MqttClientSession(connectPacket.ClientId, _options, DispatchPublishPacket); _clientSessions[connectPacket.ClientId] = clientSession; - MqttTrace.Verbose(nameof(MqttClientSessionsManager), $"Created a new session for client '{connectPacket.ClientId}'."); + MqttTrace.Verbose(nameof(MqttClientSessionsManager), "Created a new session for client '{0}'.", connectPacket.ClientId); } return new GetOrCreateClientSessionResult { IsExistingSession = isExistingSession, Session = clientSession }; diff --git a/MQTTnet.Core/Server/MqttServer.cs b/MQTTnet.Core/Server/MqttServer.cs index 231cfcf9..f3025e6b 100644 --- a/MQTTnet.Core/Server/MqttServer.cs +++ b/MQTTnet.Core/Server/MqttServer.cs @@ -84,7 +84,7 @@ namespace MQTTnet.Core.Server private void OnClientConnected(object sender, MqttClientConnectedEventArgs eventArgs) { - MqttTrace.Information(nameof(MqttServer), $"Client '{eventArgs.Identifier}': Connected."); + MqttTrace.Information(nameof(MqttServer), "Client '{0}': Connected.", eventArgs.Identifier); ClientConnected?.Invoke(this, eventArgs); Task.Run(() => _clientSessionsManager.RunClientSessionAsync(eventArgs), _cancellationTokenSource.Token); diff --git a/Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs b/Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs index 01ec4840..c8c6da6e 100644 --- a/Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs +++ b/Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs @@ -18,12 +18,12 @@ namespace MQTTnet.Core.Tests public async Task ConnectAsync(MqttClientOptions options, TimeSpan timeout) { - await Task.FromResult(0).ConfigureAwait(false); + await Task.FromResult(0); } public async Task DisconnectAsync() { - await Task.FromResult(0).ConfigureAwait(false); + await Task.FromResult(0); } public async Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout) @@ -31,14 +31,14 @@ namespace MQTTnet.Core.Tests ThrowIfPartnerIsNull(); Partner.SendPacketInternal(packet); - await Task.FromResult(0).ConfigureAwait(false); + await Task.FromResult(0); } public async Task ReceivePacketAsync(TimeSpan timeout) { ThrowIfPartnerIsNull(); - return await Task.Run(() => _incomingPackets.Take()).ConfigureAwait(false); + return await Task.Run(() => _incomingPackets.Take()); } private void SendPacketInternal(MqttBasePacket packet) From 1c30d149108d99362403eb2f26826e48d52969cc Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sat, 9 Sep 2017 23:13:53 +0200 Subject: [PATCH 4/4] Refactoring of WebSocket support --- Build/MQTTnet.nuspec | 3 +- .../Implementations/MqttWebSocketChannel.cs} | 17 +-- .../MQTTnet.NetFramework.csproj | 1 + .../MQTTnet.NetFramework/MqttClientFactory.cs | 19 ++- .../Implementations/MqttWebSocketChannel.cs | 110 ++++++++++++++++++ .../MQTTnet.NetStandard/MqttClientFactory.cs | 12 +- .../Implementations/MqttWebSocketChannel.cs | 110 ++++++++++++++++++ .../MQTTnet.UniversalWindows.csproj | 1 + .../MqttClientFactory.cs | 19 ++- MQTTnet.Core/Client/ConnectionType.cs | 14 --- MQTTnet.Core/Client/MqttClientOptions.cs | 2 +- MQTTnet.Core/Client/MqttConnectionType.cs | 10 ++ MQTTnet.TestApp.NetCore/Program.cs | 2 +- 13 files changed, 282 insertions(+), 38 deletions(-) rename Frameworks/{MQTTnet.NetStandard/Implementations/MqttWebSocketsChannel.cs => MQTTnet.NetFramework/Implementations/MqttWebSocketChannel.cs} (87%) create mode 100644 Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs create mode 100644 Frameworks/MQTTnet.UniversalWindows/Implementations/MqttWebSocketChannel.cs delete mode 100644 MQTTnet.Core/Client/ConnectionType.cs create mode 100644 MQTTnet.Core/Client/MqttConnectionType.cs diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 6da1a6fe..78c44f9e 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -10,7 +10,8 @@ https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png false MQTTnet is a .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker). - * + * [Client] Added support for web socket communication channel (thanks to nowakpiotr) +* [Core] Performance optimizations (thanks to JanEggers) Copyright Christian Kratky 2016-2017 MQTT MQTTClient MQTTServer MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Queue Hardware Arduino diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketsChannel.cs b/Frameworks/MQTTnet.NetFramework/Implementations/MqttWebSocketChannel.cs similarity index 87% rename from Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketsChannel.cs rename to Frameworks/MQTTnet.NetFramework/Implementations/MqttWebSocketChannel.cs index 8d518122..29fcae97 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketsChannel.cs +++ b/Frameworks/MQTTnet.NetFramework/Implementations/MqttWebSocketChannel.cs @@ -8,20 +8,15 @@ using System.Threading.Tasks; namespace MQTTnet.Implementations { - public sealed class MqttWebSocketsChannel : IMqttCommunicationChannel, IDisposable + public sealed class MqttWebSocketChannel : IMqttCommunicationChannel, IDisposable { - private ClientWebSocket _webSocket; + private ClientWebSocket _webSocket = new ClientWebSocket(); private const int BufferSize = 4096; private const int BufferAmplifier = 20; private readonly byte[] WebSocketBuffer = new byte[BufferSize * BufferAmplifier]; private int WebSocketBufferSize; private int WebSocketBufferOffset; - public MqttWebSocketsChannel() - { - _webSocket = new ClientWebSocket(); - } - public async Task ConnectAsync(MqttClientOptions options) { _webSocket = null; @@ -45,10 +40,7 @@ namespace MQTTnet.Implementations public void Dispose() { - if (_webSocket != null) - { - _webSocket.Dispose(); - } + _webSocket?.Dispose(); } public Task ReadAsync(byte[] buffer) @@ -70,8 +62,7 @@ namespace MQTTnet.Implementations WebSocketReceiveResult response; do { - response = - await _webSocket.ReceiveAsync(new ArraySegment(temporaryBuffer), CancellationToken.None); + response = await _webSocket.ReceiveAsync(new ArraySegment(temporaryBuffer), CancellationToken.None); temporaryBuffer.CopyTo(WebSocketBuffer, offset); offset += response.Count; diff --git a/Frameworks/MQTTnet.NetFramework/MQTTnet.NetFramework.csproj b/Frameworks/MQTTnet.NetFramework/MQTTnet.NetFramework.csproj index 7f4ab44b..7578afe3 100644 --- a/Frameworks/MQTTnet.NetFramework/MQTTnet.NetFramework.csproj +++ b/Frameworks/MQTTnet.NetFramework/MQTTnet.NetFramework.csproj @@ -100,6 +100,7 @@ + diff --git a/Frameworks/MQTTnet.NetFramework/MqttClientFactory.cs b/Frameworks/MQTTnet.NetFramework/MqttClientFactory.cs index 7dd7500c..8075f92d 100644 --- a/Frameworks/MQTTnet.NetFramework/MqttClientFactory.cs +++ b/Frameworks/MQTTnet.NetFramework/MqttClientFactory.cs @@ -1,5 +1,6 @@ using System; using MQTTnet.Core.Adapter; +using MQTTnet.Core.Channel; using MQTTnet.Core.Client; using MQTTnet.Core.Serializer; using MQTTnet.Implementations; @@ -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 static IMqttCommunicationChannel GetMqttCommunicationChannel(MqttClientOptions options) + { + switch (options.ConnectionType) + { + case MqttConnectionType.Tcp: + case MqttConnectionType.Tls: + return new MqttTcpChannel(); + case MqttConnectionType.Ws: + case MqttConnectionType.Wss: + return new MqttWebSocketChannel(); + + default: + throw new NotSupportedException(); + } } } } \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs new file mode 100644 index 00000000..29fcae97 --- /dev/null +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs @@ -0,0 +1,110 @@ +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 MqttWebSocketChannel : IMqttCommunicationChannel, IDisposable + { + private ClientWebSocket _webSocket = new ClientWebSocket(); + private const int BufferSize = 4096; + private const int BufferAmplifier = 20; + private readonly byte[] WebSocketBuffer = new byte[BufferSize * BufferAmplifier]; + private int WebSocketBufferSize; + private int WebSocketBufferOffset; + + 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); + } + } + + public async Task DisconnectAsync() + { + await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None); + } + + public void Dispose() + { + _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(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)); + } + + try + { + return _webSocket.SendAsync(new ArraySegment(buffer), WebSocketMessageType.Binary, true, + CancellationToken.None); + } + catch (WebSocketException exception) + { + throw new MqttCommunicationException(exception); + } + } + } +} \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetStandard/MqttClientFactory.cs b/Frameworks/MQTTnet.NetStandard/MqttClientFactory.cs index 1cd6d6fd..58570755 100644 --- a/Frameworks/MQTTnet.NetStandard/MqttClientFactory.cs +++ b/Frameworks/MQTTnet.NetStandard/MqttClientFactory.cs @@ -22,15 +22,15 @@ namespace MQTTnet { switch (options.ConnectionType) { - case ConnectionTypes.TCP: - case ConnectionTypes.TLS: + case MqttConnectionType.Tcp: + case MqttConnectionType.Tls: return new MqttTcpChannel(); - case ConnectionTypes.WS: - case ConnectionTypes.WSS: - return new MqttWebSocketsChannel(); + case MqttConnectionType.Ws: + case MqttConnectionType.Wss: + return new MqttWebSocketChannel(); default: - return null; + throw new NotSupportedException(); } } } diff --git a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttWebSocketChannel.cs b/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttWebSocketChannel.cs new file mode 100644 index 00000000..29fcae97 --- /dev/null +++ b/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttWebSocketChannel.cs @@ -0,0 +1,110 @@ +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 MqttWebSocketChannel : IMqttCommunicationChannel, IDisposable + { + private ClientWebSocket _webSocket = new ClientWebSocket(); + private const int BufferSize = 4096; + private const int BufferAmplifier = 20; + private readonly byte[] WebSocketBuffer = new byte[BufferSize * BufferAmplifier]; + private int WebSocketBufferSize; + private int WebSocketBufferOffset; + + 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); + } + } + + public async Task DisconnectAsync() + { + await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None); + } + + public void Dispose() + { + _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(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)); + } + + try + { + return _webSocket.SendAsync(new ArraySegment(buffer), WebSocketMessageType.Binary, true, + CancellationToken.None); + } + catch (WebSocketException exception) + { + throw new MqttCommunicationException(exception); + } + } + } +} \ No newline at end of file diff --git a/Frameworks/MQTTnet.UniversalWindows/MQTTnet.UniversalWindows.csproj b/Frameworks/MQTTnet.UniversalWindows/MQTTnet.UniversalWindows.csproj index 791ee28f..3dd8a2dc 100644 --- a/Frameworks/MQTTnet.UniversalWindows/MQTTnet.UniversalWindows.csproj +++ b/Frameworks/MQTTnet.UniversalWindows/MQTTnet.UniversalWindows.csproj @@ -110,6 +110,7 @@ bin\x64\Release\MQTTnet.XML + diff --git a/Frameworks/MQTTnet.UniversalWindows/MqttClientFactory.cs b/Frameworks/MQTTnet.UniversalWindows/MqttClientFactory.cs index 7dd7500c..8075f92d 100644 --- a/Frameworks/MQTTnet.UniversalWindows/MqttClientFactory.cs +++ b/Frameworks/MQTTnet.UniversalWindows/MqttClientFactory.cs @@ -1,5 +1,6 @@ using System; using MQTTnet.Core.Adapter; +using MQTTnet.Core.Channel; using MQTTnet.Core.Client; using MQTTnet.Core.Serializer; using MQTTnet.Implementations; @@ -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 static IMqttCommunicationChannel GetMqttCommunicationChannel(MqttClientOptions options) + { + switch (options.ConnectionType) + { + case MqttConnectionType.Tcp: + case MqttConnectionType.Tls: + return new MqttTcpChannel(); + case MqttConnectionType.Ws: + case MqttConnectionType.Wss: + return new MqttWebSocketChannel(); + + default: + throw new NotSupportedException(); + } } } } \ No newline at end of file diff --git a/MQTTnet.Core/Client/ConnectionType.cs b/MQTTnet.Core/Client/ConnectionType.cs deleted file mode 100644 index 1c6abad6..00000000 --- a/MQTTnet.Core/Client/ConnectionType.cs +++ /dev/null @@ -1,14 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Text; - -namespace MQTTnet.Core.Client -{ - public enum ConnectionTypes - { - TCP, - TLS, - WS, - WSS - } -} diff --git a/MQTTnet.Core/Client/MqttClientOptions.cs b/MQTTnet.Core/Client/MqttClientOptions.cs index 8b652995..12e7ae98 100644 --- a/MQTTnet.Core/Client/MqttClientOptions.cs +++ b/MQTTnet.Core/Client/MqttClientOptions.cs @@ -25,6 +25,6 @@ namespace MQTTnet.Core.Client public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311; - public ConnectionTypes ConnectionType { get; set; } = ConnectionTypes.TCP; + public MqttConnectionType ConnectionType { get; set; } = MqttConnectionType.Tcp; } } diff --git a/MQTTnet.Core/Client/MqttConnectionType.cs b/MQTTnet.Core/Client/MqttConnectionType.cs new file mode 100644 index 00000000..636054dc --- /dev/null +++ b/MQTTnet.Core/Client/MqttConnectionType.cs @@ -0,0 +1,10 @@ +namespace MQTTnet.Core.Client +{ + public enum MqttConnectionType + { + Tcp, + Tls, + Ws, + Wss + } +} diff --git a/MQTTnet.TestApp.NetCore/Program.cs b/MQTTnet.TestApp.NetCore/Program.cs index 46444714..e5470d94 100644 --- a/MQTTnet.TestApp.NetCore/Program.cs +++ b/MQTTnet.TestApp.NetCore/Program.cs @@ -51,7 +51,7 @@ namespace MQTTnet.TestApp.NetCore Server = "localhost", ClientId = "XYZ", CleanSession = true, - ConnectionType = ConnectionTypes.WS + ConnectionType = MqttConnectionType.Ws }; var client = new MqttClientFactory().CreateMqttClient(options);