Add more config values for TCP connections.

This commit is contained in:
Christian Kratky
2019-04-03 23:46:13 +02:00
parent cdeb073bb0
commit b5a1d3a69f
5 changed files with 48 additions and 17 deletions
@@ -1,4 +1,6 @@
namespace MQTTnet.Client.Options using System.Net.Sockets;
namespace MQTTnet.Client.Options
{ {
public class MqttClientTcpOptions : IMqttClientChannelOptions public class MqttClientTcpOptions : IMqttClientChannelOptions
{ {
@@ -6,7 +8,13 @@
public int? Port { get; set; } 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(); public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
+34 -13
View File
@@ -13,10 +13,14 @@ namespace MQTTnet.Formatter
/// </summary> /// </summary>
public class MqttPacketWriter public class MqttPacketWriter
{ {
private static readonly ArraySegment<byte> ZeroVariableLengthIntegerArray = new ArraySegment<byte>(new byte[1], 0, 1);
private static readonly ArraySegment<byte> ZeroTwoByteIntegerArray = new ArraySegment<byte>(new byte[2], 0, 2);
public static int InitialBufferSize = 128;
public static int MaxBufferSize = 4096; public static int MaxBufferSize = 4096;
// TODO: Consider using the ArrayPool here together with FreeBuffer. // TODO: Consider using the ArrayPool here together with FreeBuffer.
private byte[] _buffer = new byte[128]; private byte[] _buffer = new byte[InitialBufferSize];
private int _offset; private int _offset;
@@ -33,7 +37,7 @@ namespace MQTTnet.Formatter
{ {
if (value == 0) if (value == 0)
{ {
return new ArraySegment<byte>(new byte[1], 0, 1); return ZeroVariableLengthIntegerArray;
} }
if (value <= 127) if (value <= 127)
@@ -68,7 +72,14 @@ namespace MQTTnet.Formatter
public void WriteWithLengthPrefix(string value) 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) public void WriteWithLengthPrefix(byte[] value)
@@ -77,8 +88,15 @@ namespace MQTTnet.Formatter
EnsureAdditionalCapacity(value.Length + 2); EnsureAdditionalCapacity(value.Length + 2);
Write((ushort)value.Length); if (value.Length == 0)
Write(value, 0, value.Length); {
Write(ZeroTwoByteIntegerArray);
}
else
{
Write((ushort)value.Length);
Write(value, 0, value.Length);
}
} }
public void Write(byte @byte) public void Write(byte @byte)
@@ -103,19 +121,17 @@ namespace MQTTnet.Formatter
{ {
if (buffer == null) throw new ArgumentNullException(nameof(buffer)); if (buffer == null) throw new ArgumentNullException(nameof(buffer));
if (count == 0)
{
return;
}
EnsureAdditionalCapacity(count); EnsureAdditionalCapacity(count);
Array.Copy(buffer, offset, _buffer, _offset, count); Array.Copy(buffer, offset, _buffer, _offset, count);
IncreasePosition(count); IncreasePosition(count);
} }
public void Write(ArraySegment<byte> buffer)
{
if (buffer == null) throw new ArgumentNullException(nameof(buffer));
Write(buffer.Array, buffer.Offset, buffer.Count);
}
public void Write(MqttPacketWriter propertyWriter) public void Write(MqttPacketWriter propertyWriter)
{ {
if (propertyWriter == null) throw new ArgumentNullException(nameof(propertyWriter)); if (propertyWriter == null) throw new ArgumentNullException(nameof(propertyWriter));
@@ -160,6 +176,11 @@ namespace MQTTnet.Formatter
Array.Resize(ref _buffer, MaxBufferSize); Array.Resize(ref _buffer, MaxBufferSize);
} }
private void Write(ArraySegment<byte> buffer)
{
Write(buffer.Array, buffer.Offset, buffer.Count);
}
[MethodImpl(MethodImplOptions.AggressiveInlining)] [MethodImpl(MethodImplOptions.AggressiveInlining)]
private void EnsureAdditionalCapacity(int additionalCapacity) private void EnsureAdditionalCapacity(int additionalCapacity)
{ {
@@ -62,7 +62,7 @@ namespace MQTTnet.Implementations
if (_socket == null) if (_socket == null)
{ {
_socket = new StreamSocket(); _socket = new StreamSocket();
_socket.Control.NoDelay = true; _socket.Control.NoDelay = _options.NoDelay;
_socket.Control.KeepAlive = true; _socket.Control.KeepAlive = true;
} }
@@ -36,7 +36,7 @@ namespace MQTTnet.Implementations
_listener = new StreamSocketListener(); _listener = new StreamSocketListener();
// This also affects the client sockets. // This also affects the client sockets.
_listener.Control.NoDelay = true; _listener.Control.NoDelay = options.DefaultEndpointOptions.NoDelay;
_listener.Control.KeepAlive = true; _listener.Control.KeepAlive = true;
_listener.Control.QualityOfService = SocketQualityOfService.LowLatency; _listener.Control.QualityOfService = SocketQualityOfService.LowLatency;
_listener.ConnectionReceived += AcceptDefaultEndpointConnectionsAsync; _listener.ConnectionReceived += AcceptDefaultEndpointConnectionsAsync;
@@ -10,6 +10,8 @@ namespace MQTTnet.Server
public int ConnectionBacklog { get; set; } = 10; public int ConnectionBacklog { get; set; } = 10;
public bool NoDelay { get; set; } = true;
#if WINDOWS_UWP #if WINDOWS_UWP
public int BufferSize { get; set; } = 4096; public int BufferSize { get; set; } = 4096;
#endif #endif