@@ -11,6 +11,7 @@ | |||||
<requireLicenseAcceptance>false</requireLicenseAcceptance> | <requireLicenseAcceptance>false</requireLicenseAcceptance> | ||||
<description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker).</description> | <description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker).</description> | ||||
<releaseNotes> * [Client] Added new overloads for the message builder. | <releaseNotes> * [Client] Added new overloads for the message builder. | ||||
* [Core] Performance optimizations (thanks to @ israellot) | |||||
</releaseNotes> | </releaseNotes> | ||||
<copyright>Copyright Christian Kratky 2016-2018</copyright> | <copyright>Copyright Christian Kratky 2016-2018</copyright> | ||||
<tags>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</tags> | <tags>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</tags> | ||||
@@ -1,9 +1,6 @@ | |||||
using System; | using System; | ||||
using System.Collections.Generic; | |||||
using System.IO; | using System.IO; | ||||
using System.Linq; | |||||
using System.Net.Sockets; | using System.Net.Sockets; | ||||
using System.Runtime.ExceptionServices; | |||||
using System.Runtime.InteropServices; | using System.Runtime.InteropServices; | ||||
using System.Threading; | using System.Threading; | ||||
using System.Threading.Tasks; | using System.Threading.Tasks; | ||||
@@ -53,14 +50,13 @@ namespace MQTTnet.Adapter | |||||
public async Task SendPacketsAsync(TimeSpan timeout, CancellationToken cancellationToken, MqttBasePacket[] packets) | public async Task SendPacketsAsync(TimeSpan timeout, CancellationToken cancellationToken, MqttBasePacket[] packets) | ||||
{ | { | ||||
for(var i=0;i<packets.Length;i++) | |||||
foreach (var packet in packets) | |||||
{ | { | ||||
await SendPacketsAsync(timeout, cancellationToken, packets[i]).ConfigureAwait(false); | |||||
await SendPacketsAsync(timeout, cancellationToken, packet).ConfigureAwait(false); | |||||
} | } | ||||
} | } | ||||
public Task SendPacketsAsync(TimeSpan timeout, CancellationToken cancellationToken, MqttBasePacket packet) | |||||
private Task SendPacketsAsync(TimeSpan timeout, CancellationToken cancellationToken, MqttBasePacket packet) | |||||
{ | { | ||||
ThrowIfDisposed(); | ThrowIfDisposed(); | ||||
@@ -75,21 +71,23 @@ namespace MQTTnet.Adapter | |||||
{ | { | ||||
return; | return; | ||||
} | } | ||||
_logger.Verbose<MqttChannelAdapter>("TX >>> {0} [Timeout={1}]", packet, timeout); | _logger.Verbose<MqttChannelAdapter>("TX >>> {0} [Timeout={1}]", packet, timeout); | ||||
var packetData = PacketSerializer.Serialize(packet); | var packetData = PacketSerializer.Serialize(packet); | ||||
if (cancellationToken.IsCancellationRequested) | if (cancellationToken.IsCancellationRequested) | ||||
{ | { | ||||
return; | return; | ||||
} | } | ||||
await _channel.SendStream.WriteAsync( | await _channel.SendStream.WriteAsync( | ||||
packetData.Array, | packetData.Array, | ||||
packetData.Offset, | packetData.Offset, | ||||
(int)packetData.Count, | |||||
packetData.Count, | |||||
cancellationToken).ConfigureAwait(false); | cancellationToken).ConfigureAwait(false); | ||||
await _channel.SendStream.FlushAsync(cancellationToken).ConfigureAwait(false); | |||||
}); | }); | ||||
} | } | ||||
@@ -113,12 +111,11 @@ namespace MQTTnet.Adapter | |||||
{ | { | ||||
receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, linkedCts.Token).ConfigureAwait(false); | receivedMqttPacket = await ReceiveAsync(_channel.ReceiveStream, linkedCts.Token).ConfigureAwait(false); | ||||
} | } | ||||
catch(OperationCanceledException ex) | |||||
catch (OperationCanceledException ex) | |||||
{ | { | ||||
//check if timed out | |||||
if(linkedCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested) | |||||
var timedOut = linkedCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested; | |||||
if (timedOut) | |||||
{ | { | ||||
//only timeout token was cancelled | |||||
throw new MqttCommunicationTimedOutException(ex); | throw new MqttCommunicationTimedOutException(ex); | ||||
} | } | ||||
else | else | ||||
@@ -14,7 +14,7 @@ namespace MQTTnet.Implementations | |||||
{ | { | ||||
public sealed class MqttTcpChannel : IMqttChannel | public sealed class MqttTcpChannel : IMqttChannel | ||||
{ | { | ||||
#if NET452 || NET461 | |||||
#if NET452 || NET461 || NETSTANDARD2_0 | |||||
// ReSharper disable once MemberCanBePrivate.Global | // ReSharper disable once MemberCanBePrivate.Global | ||||
// ReSharper disable once AutoPropertyCanBeMadeGetOnly.Global | // ReSharper disable once AutoPropertyCanBeMadeGetOnly.Global | ||||
public static int BufferSize { get; set; } = 4096 * 20; // Can be changed for fine tuning by library user. | public static int BufferSize { get; set; } = 4096 * 20; // Can be changed for fine tuning by library user. | ||||
@@ -60,7 +60,6 @@ namespace MQTTnet.Implementations | |||||
if (_socket == null) | if (_socket == null) | ||||
{ | { | ||||
_socket = new Socket(SocketType.Stream, ProtocolType.Tcp); | _socket = new Socket(SocketType.Stream, ProtocolType.Tcp); | ||||
} | } | ||||
#if NET452 || NET461 | #if NET452 || NET461 | ||||
@@ -212,8 +211,8 @@ namespace MQTTnet.Implementations | |||||
{ | { | ||||
stream = new NetworkStream(_socket, true); | stream = new NetworkStream(_socket, true); | ||||
} | } | ||||
#if NET452 || NET461 | |||||
#if NET452 || NET461 || NETSTANDARD2_0 | |||||
SendStream = new BufferedStream(stream, _bufferSize); | SendStream = new BufferedStream(stream, _bufferSize); | ||||
ReceiveStream = new BufferedStream(stream, _bufferSize); | ReceiveStream = new BufferedStream(stream, _bufferSize); | ||||
#else | #else | ||||
@@ -38,8 +38,7 @@ namespace MQTTnet.Implementations | |||||
if (options.DefaultEndpointOptions.IsEnabled) | if (options.DefaultEndpointOptions.IsEnabled) | ||||
{ | { | ||||
_defaultEndpointSocket = new Socket(SocketType.Stream, ProtocolType.Tcp); | |||||
_defaultEndpointSocket.NoDelay = true; | |||||
_defaultEndpointSocket = new Socket(SocketType.Stream, ProtocolType.Tcp) { NoDelay = true }; | |||||
_defaultEndpointSocket.Bind(new IPEndPoint(options.DefaultEndpointOptions.BoundIPAddress, options.GetDefaultEndpointPort())); | _defaultEndpointSocket.Bind(new IPEndPoint(options.DefaultEndpointOptions.BoundIPAddress, options.GetDefaultEndpointPort())); | ||||
_defaultEndpointSocket.Listen(options.ConnectionBacklog); | _defaultEndpointSocket.Listen(options.ConnectionBacklog); | ||||
@@ -80,7 +79,7 @@ namespace MQTTnet.Implementations | |||||
_defaultEndpointSocket = null; | _defaultEndpointSocket = null; | ||||
_tlsCertificate = null; | _tlsCertificate = null; | ||||
_tlsEndpointSocket?.Dispose(); | _tlsEndpointSocket?.Dispose(); | ||||
_tlsEndpointSocket = null; | _tlsEndpointSocket = null; | ||||
@@ -104,7 +103,8 @@ namespace MQTTnet.Implementations | |||||
#else | #else | ||||
var clientSocket = await _defaultEndpointSocket.AcceptAsync().ConfigureAwait(false); | var clientSocket = await _defaultEndpointSocket.AcceptAsync().ConfigureAwait(false); | ||||
#endif | #endif | ||||
clientSocket.NoDelay=true; | |||||
clientSocket.NoDelay = true; | |||||
var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, null), new MqttPacketSerializer(), _logger); | var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, null), new MqttPacketSerializer(), _logger); | ||||
ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter)); | ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter)); | ||||
} | } | ||||
@@ -139,7 +139,7 @@ namespace MQTTnet.Implementations | |||||
var sslStream = new SslStream(new NetworkStream(clientSocket)); | 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).ConfigureAwait(false); | ||||
var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketSerializer(), _logger); | var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketSerializer(), _logger); | ||||
ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter)); | ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter)); | ||||
} | } | ||||
@@ -2,7 +2,6 @@ | |||||
using MQTTnet.Packets; | using MQTTnet.Packets; | ||||
using MQTTnet.Protocol; | using MQTTnet.Protocol; | ||||
using System; | using System; | ||||
using System.Collections.Generic; | |||||
using System.IO; | using System.IO; | ||||
using System.Linq; | using System.Linq; | ||||
using System.Text; | using System.Text; | ||||
@@ -23,29 +22,28 @@ namespace MQTTnet.Serializer | |||||
using (var stream = new MemoryStream(128)) | using (var stream = new MemoryStream(128)) | ||||
using (var writer = new MqttPacketWriter(stream)) | using (var writer = new MqttPacketWriter(stream)) | ||||
{ | { | ||||
//leave enough head space for max header (fixed + 4 variable remaining lenght) | |||||
// Leave enough head space for max header size (fixed + 4 variable remaining length) | |||||
stream.Position = 5; | stream.Position = 5; | ||||
var fixedHeader = SerializePacket(packet, writer); | var fixedHeader = SerializePacket(packet, writer); | ||||
var remainingLength = MqttPacketWriter.GetRemainingLength((int)stream.Length-5); | |||||
var remainingLength = MqttPacketWriter.GetRemainingLength((int)stream.Length - 5); | |||||
var headerSize = remainingLength.Length + 1; | var headerSize = remainingLength.Length + 1; | ||||
var headerOffset = 5 - headerSize; | var headerOffset = 5 - headerSize; | ||||
//position curson on correct offset on beginining of array | |||||
// Position cursor on correct offset on beginining of array (has leading 0x0) | |||||
stream.Position = headerOffset; | stream.Position = headerOffset; | ||||
//write header | |||||
writer.Write(fixedHeader); | writer.Write(fixedHeader); | ||||
writer.Write(remainingLength,0,remainingLength.Length); | |||||
writer.Write(remainingLength, 0, remainingLength.Length); | |||||
#if NET461 || NET452 || NETSTANDARD2_0 | #if NET461 || NET452 || NETSTANDARD2_0 | ||||
var buffer = stream.GetBuffer(); | var buffer = stream.GetBuffer(); | ||||
#else | #else | ||||
var buffer = stream.ToArray(); | var buffer = stream.ToArray(); | ||||
#endif | #endif | ||||
return new ArraySegment<byte>(buffer, headerOffset, (int)stream.Length- headerOffset); | |||||
return new ArraySegment<byte>(buffer, headerOffset, (int)stream.Length - headerOffset); | |||||
} | } | ||||
} | } | ||||
@@ -300,7 +298,7 @@ namespace MQTTnet.Serializer | |||||
return packet; | return packet; | ||||
} | } | ||||
private static void ValidateConnectPacket(MqttConnectPacket packet) | private static void ValidateConnectPacket(MqttConnectPacket packet) | ||||
{ | { | ||||
if (packet == null) throw new ArgumentNullException(nameof(packet)); | if (packet == null) throw new ArgumentNullException(nameof(packet)); | ||||
@@ -60,11 +60,11 @@ namespace MQTTnet.Serializer | |||||
{ | { | ||||
if (length <= 0) | if (length <= 0) | ||||
{ | { | ||||
return new [] { (byte)0 }; | |||||
return new[] { (byte)0 }; | |||||
} | } | ||||
var bytes = new byte[4]; | var bytes = new byte[4]; | ||||
int arraySize = 0; | |||||
var offset = 0; | |||||
// Alorithm taken from http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html. | // Alorithm taken from http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html. | ||||
var x = length; | var x = length; | ||||
@@ -77,12 +77,12 @@ namespace MQTTnet.Serializer | |||||
encodedByte = encodedByte | 128; | encodedByte = encodedByte | 128; | ||||
} | } | ||||
bytes[arraySize] = (byte)encodedByte; | |||||
bytes[offset] = (byte)encodedByte; | |||||
arraySize++; | |||||
offset++; | |||||
} while (x > 0); | } while (x > 0); | ||||
return bytes.Take(arraySize).ToArray(); | |||||
return bytes.Take(offset).ToArray(); | |||||
} | } | ||||
} | } | ||||
} | } |
@@ -44,7 +44,7 @@ namespace MQTTnet.TestApp.NetCore | |||||
try | try | ||||
{ | { | ||||
await client.ConnectAsync(options); | |||||
await client.ConnectAsync(options).ConfigureAwait(false); | |||||
} | } | ||||
catch (Exception exception) | catch (Exception exception) | ||||
{ | { | ||||
@@ -165,7 +165,7 @@ namespace MQTTnet.TestApp.NetCore | |||||
Console.WriteLine("Press any key to exit."); | Console.WriteLine("Press any key to exit."); | ||||
Console.ReadLine(); | Console.ReadLine(); | ||||
await mqttServer.StopAsync(); | |||||
await mqttServer.StopAsync().ConfigureAwait(false); | |||||
} | } | ||||
catch (Exception e) | catch (Exception e) | ||||
{ | { | ||||