diff --git a/Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs b/Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs index 35af34fe..936e2111 100644 --- a/Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs +++ b/Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs @@ -1,4 +1,6 @@ -namespace MQTTnet.Client.Options +using System.Net.Sockets; + +namespace MQTTnet.Client.Options { public class MqttClientTcpOptions : IMqttClientChannelOptions { @@ -6,7 +8,13 @@ public int? Port { get; set; } - public int BufferSize { get; set; } = 4096; + public int BufferSize { get; set; } = 1024; + + public bool DualMode { get; set; } = true; + + public bool NoDelay { get; set; } = true; + + public AddressFamily AddressFamily { get; set; } = AddressFamily.Unspecified; public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions(); diff --git a/Source/MQTTnet/Formatter/MqttPacketWriter.cs b/Source/MQTTnet/Formatter/MqttPacketWriter.cs index da483bb6..6cbdf9f6 100644 --- a/Source/MQTTnet/Formatter/MqttPacketWriter.cs +++ b/Source/MQTTnet/Formatter/MqttPacketWriter.cs @@ -13,10 +13,14 @@ namespace MQTTnet.Formatter /// public class MqttPacketWriter { + private static readonly ArraySegment ZeroVariableLengthIntegerArray = new ArraySegment(new byte[1], 0, 1); + private static readonly ArraySegment ZeroTwoByteIntegerArray = new ArraySegment(new byte[2], 0, 2); + + public static int InitialBufferSize = 128; public static int MaxBufferSize = 4096; // TODO: Consider using the ArrayPool here together with FreeBuffer. - private byte[] _buffer = new byte[128]; + private byte[] _buffer = new byte[InitialBufferSize]; private int _offset; @@ -33,7 +37,7 @@ namespace MQTTnet.Formatter { if (value == 0) { - return new ArraySegment(new byte[1], 0, 1); + return ZeroVariableLengthIntegerArray; } if (value <= 127) @@ -68,7 +72,14 @@ namespace MQTTnet.Formatter public void WriteWithLengthPrefix(string value) { - WriteWithLengthPrefix(Encoding.UTF8.GetBytes(value ?? string.Empty)); + if (string.IsNullOrEmpty(value)) + { + Write(ZeroTwoByteIntegerArray); + } + else + { + WriteWithLengthPrefix(Encoding.UTF8.GetBytes(value)); + } } public void WriteWithLengthPrefix(byte[] value) @@ -77,8 +88,15 @@ namespace MQTTnet.Formatter EnsureAdditionalCapacity(value.Length + 2); - Write((ushort)value.Length); - Write(value, 0, value.Length); + if (value.Length == 0) + { + Write(ZeroTwoByteIntegerArray); + } + else + { + Write((ushort)value.Length); + Write(value, 0, value.Length); + } } public void Write(byte @byte) @@ -103,19 +121,17 @@ namespace MQTTnet.Formatter { if (buffer == null) throw new ArgumentNullException(nameof(buffer)); + if (count == 0) + { + return; + } + EnsureAdditionalCapacity(count); Array.Copy(buffer, offset, _buffer, _offset, count); IncreasePosition(count); } - - public void Write(ArraySegment buffer) - { - if (buffer == null) throw new ArgumentNullException(nameof(buffer)); - - Write(buffer.Array, buffer.Offset, buffer.Count); - } - + public void Write(MqttPacketWriter propertyWriter) { if (propertyWriter == null) throw new ArgumentNullException(nameof(propertyWriter)); @@ -160,6 +176,11 @@ namespace MQTTnet.Formatter Array.Resize(ref _buffer, MaxBufferSize); } + private void Write(ArraySegment buffer) + { + Write(buffer.Array, buffer.Offset, buffer.Count); + } + [MethodImpl(MethodImplOptions.AggressiveInlining)] private void EnsureAdditionalCapacity(int additionalCapacity) { diff --git a/Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs index 092b7030..509c3e85 100644 --- a/Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs +++ b/Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs @@ -62,7 +62,7 @@ namespace MQTTnet.Implementations if (_socket == null) { _socket = new StreamSocket(); - _socket.Control.NoDelay = true; + _socket.Control.NoDelay = _options.NoDelay; _socket.Control.KeepAlive = true; } diff --git a/Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs b/Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs index 9fbf27b8..5ceba0e1 100644 --- a/Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs +++ b/Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs @@ -36,7 +36,7 @@ namespace MQTTnet.Implementations _listener = new StreamSocketListener(); // This also affects the client sockets. - _listener.Control.NoDelay = true; + _listener.Control.NoDelay = options.DefaultEndpointOptions.NoDelay; _listener.Control.KeepAlive = true; _listener.Control.QualityOfService = SocketQualityOfService.LowLatency; _listener.ConnectionReceived += AcceptDefaultEndpointConnectionsAsync; diff --git a/Source/MQTTnet/Server/MqttServerTcpEndpointBaseOptions.cs b/Source/MQTTnet/Server/MqttServerTcpEndpointBaseOptions.cs index 45df2780..01b8c704 100644 --- a/Source/MQTTnet/Server/MqttServerTcpEndpointBaseOptions.cs +++ b/Source/MQTTnet/Server/MqttServerTcpEndpointBaseOptions.cs @@ -10,6 +10,8 @@ namespace MQTTnet.Server public int ConnectionBacklog { get; set; } = 10; + public bool NoDelay { get; set; } = true; + #if WINDOWS_UWP public int BufferSize { get; set; } = 4096; #endif