Compare commits

..
45 Commits
Author SHA1 Message Date
Christian Kratky efef23e3f5 Merge branch 'develop' 2019-05-06 20:24:12 +02:00
Christian Kratky 53ab2ac140 Add Wiki Code. 2019-05-06 20:22:41 +02:00
Christian Kratky 2ecbd83d06 Add UnitTests. 2019-05-06 20:22:20 +02:00
Christian Kratky a278b6189b Refactor empty payload detection for retained messages. 2019-05-01 19:41:34 +02:00
Christian Kratky cdbdfdec6d Add UnitTests. 2019-05-01 19:39:10 +02:00
Christian Kratky 808c2da5f6 Add PacketIdentifier for QoS 2. 2019-04-28 23:03:54 +02:00
Christian Kratky 6ed3b0cedd Add ClientId for application messages published by the server. 2019-04-24 21:37:29 +02:00
Christian Kratky 9da3377aa7 Fix credential generation. 2019-04-24 20:19:55 +02:00
Christian Kratky 0cac2aa1bc Fix broken log message. 2019-04-24 20:19:46 +02:00
Christian Kratky 2d59b5c1ac Refactoring. 2019-04-23 14:49:19 +02:00
Christian Kratky 9dfa0b32e4 Add extension methods for server. 2019-04-23 14:49:12 +02:00
Christian Kratky 20a7579156 Fix issues in ApplicationMessageInterceptor. 2019-04-23 14:48:51 +02:00
Christian Kratky c7409c2ff4 Fix missing endpoint in client status. 2019-04-23 14:48:38 +02:00
Christian Kratky d4c6fb44c9 Expose PacketIdentifier in PublishResult. 2019-04-23 14:47:42 +02:00
Christian Kratky e6cf23aebd Add API controllers for MQTTnet.Server. 2019-04-23 14:46:13 +02:00
Christian Kratky 0fec860b25 Add UnitTests. 2019-04-23 14:46:02 +02:00
Christian Kratky 2c927c5995 Cleanup. 2019-04-22 10:59:39 +02:00
Christian Kratky 6399172582 Improve connection disposal. 2019-04-22 10:50:47 +02:00
Christian Kratky 5c903e6fbf Improve connection handling. 2019-04-20 21:12:30 +02:00
Christian Kratky f60bbf7dd7 MQTTnet.Server startup cleanup. 2019-04-20 20:59:28 +02:00
Christian Kratky 519bb05aac Fix broken code. 2019-04-18 00:09:19 +02:00
Christian Kratky a8434d1b58 Expose results of publish operations for MQTTv5. 2019-04-18 00:01:22 +02:00
Christian Kratky 420b35e37d Merge remote-tracking branch 'origin/develop' into develop 2019-04-17 21:44:12 +02:00
Christian Kratky a25c8cd4c7 Fix some minor TODOs. 2019-04-17 21:44:09 +02:00
Christian 8ac77558af Merge pull request #593 from JanEggers/fixTodo
fixed todo in MqttConnectionContext
2019-04-17 19:14:00 +02:00
JanEggers 953578eba2 fixed todo in MqttConnectionContext 2019-04-17 09:47:17 +02:00
Christian 527713ccf8 Merge pull request #592 from JanEggers/spanV2
performance improvements span based serializers
2019-04-16 19:28:38 +02:00
JanEggers 477e2d23ab added benchmark option to compare netfx and core runtime 2019-04-15 15:52:51 +02:00
JanEggers 03c61ad82c fixed tests and added testcase for user property serialization 2019-04-15 15:37:15 +02:00
JanEggers 9475ccd0d8 revert accidential change 2019-04-15 15:31:31 +02:00
JanEggers 73b051d159 reuse reader 2019-04-15 15:00:09 +02:00
JanEggers a9944def68 added span based writer 2019-04-15 14:43:49 +02:00
JanEggers aee10ba6fa added netcore 2.1 build 2019-04-15 14:24:27 +02:00
JanEggers cfc1b27950 fixed build and added implementations for new read functions 2019-04-15 14:14:32 +02:00
JanEggers 149d01e173 added tests 2019-04-15 14:09:22 +02:00
JanEggers 888e1797d7 fixed cherry pick 2019-04-15 14:04:21 +02:00
JanEggers f8ade2daf4 slight improvements 2019-04-15 14:04:06 +02:00
JanEggers a85792c392 fixed another issue 2019-04-15 14:00:22 +02:00
JanEggers fbca8a6dc1 span based reader 2019-04-15 14:00:00 +02:00
Christian Kratky 732ea84caa Update build script. 2019-04-13 00:09:10 +02:00
Christian Kratky 65896269da Improve protocol version detection. 2019-04-11 20:13:30 +02:00
Christian Kratky 631f660d5c Add MQTTnet.Server to build script. 2019-04-11 20:13:08 +02:00
Christian Kratky 7f08371532 Update tests. 2019-04-08 22:29:52 +02:00
Christian Kratky 1afd59325f Add validations for client options builder. 2019-04-08 22:29:46 +02:00
Christian Kratky 45c033f401 Improve payload handling and allocation. 2019-04-08 22:29:29 +02:00
91 changed files with 1726 additions and 617 deletions
+1
View File
@@ -17,6 +17,7 @@
<dependency id="MQTTnet" version="$nugetVersion" />
<dependency id="Microsoft.AspNetCore.Connections.Abstractions" version="2.1.3" />
<dependency id="Microsoft.AspNetCore.Http.Connections" version="1.0.3" />
<dependency id="Microsoft.AspNetCore.WebSockets" version="2.1.1" />
<dependency id="Microsoft.Extensions.Hosting.Abstractions" version="2.1.1" />
</dependencies>
+1
View File
@@ -25,6 +25,7 @@
* [Client] Fixed issues in QoS 2 handling which leads to message loss.
* [Client] Replaced all events with proper async compatible handlers (BREAKING CHANGE!).
* [Client] Added more configuration values to TCP endpoint options.
* [Client] Added used PacketIdentifier to publish result.
* [ManagedClient] Replaced all events with proper async compatible handlers (BREAKING CHANGE!).
* [ManagedClient] The log ID is now propagated to the internal client (thanks to @vbBerni).
* [ManagedClient] Added validation of topics before publishing.
+11
View File
@@ -36,6 +36,17 @@ Write-Host
&$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netstandard2.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
&$msbuild ..\Source\MQTTnet.Extensions.ManagedClient\MQTTnet.Extensions.ManagedClient.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="uap10.0" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
# Build MQTTnet.Server
Remove-Item ..\Source\MQTTnet.Server\bin\Release\netcoreapp2.2 -Recurse -Force -ErrorAction SilentlyContinue
&$msbuild ..\Source\MQTTnet.Server\MQTTnet.Server.csproj /t:Build /p:Configuration="Release" /p:TargetFramework="netcoreapp2.2" /p:FileVersion=$assemblyVersion /p:AssemblyVersion=$assemblyVersion /verbosity:m /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=".\..\..\Build\codeSigningKey.pfx"
$source = "..\Source\MQTTnet.Server\bin\Release\netcoreapp2.2"
$destination = "..\Source\MQTTnet.Server\bin\MQTTnet.Server-Portable-v$nugetVersion.zip"
If(Test-path $destination) {Remove-item $destination}
Add-Type -assembly "system.io.compression.filesystem"
[io.compression.zipfile]::CreateFromDirectory($source, $destination)
# Create NuGet packages.
Remove-Item .\NuGet -Force -Recurse -ErrorAction SilentlyContinue
Copy-Item MQTTnet.AspNetCore.nuspec -Destination MQTTnet.AspNetCore.nuspec.old -Force
+1 -1
View File
@@ -26,7 +26,7 @@ MQTTnet is a high performance .NET library for MQTT based communication. It prov
* Uniform API across all supported versions of the MQTT protocol
* Interfaces included for mocking and testing
* Access to internal trace messages
* Unit tested (~150 tests)
* Unit tested (~200 tests)
* No external dependencies
\* Tested on local machine (Intel i7 8700K) with MQTTnet client and server running in the same process using the TCP channel. The app for verification is part of this repository and stored in _/Tests/MQTTnet.TestApp.NetCore_.
@@ -28,7 +28,7 @@ namespace MQTTnet.AspNetCore
}
var adapter = app.ApplicationServices.GetRequiredService<MqttWebSocketServerAdapter>();
using (var webSocket = await context.WebSockets.AcceptWebSocketAsync(subProtocol))
using (var webSocket = await context.WebSockets.AcceptWebSocketAsync(subProtocol).ConfigureAwait(false))
{
await adapter.RunWebSocketConnectionAsync(webSocket, context);
}
@@ -21,7 +21,10 @@ namespace MQTTnet.AspNetCore.Client
{
var endpoint = new DnsEndPoint(tcpOptions.Server, tcpOptions.GetPort());
var tcpConnection = new TcpConnection(endpoint);
return new MqttConnectionContext(new MqttPacketFormatterAdapter(options.ProtocolVersion), tcpConnection);
var writer = new SpanBasedMqttPacketWriter();
var formatter = new MqttPacketFormatterAdapter(options.ProtocolVersion, writer);
return new MqttConnectionContext(formatter, tcpConnection);
}
default:
{
@@ -5,6 +5,10 @@ using System.Diagnostics;
using System.IO.Pipelines;
using System.Net.Sockets;
#if NETCOREAPP2_1
using System.Runtime.InteropServices;
#endif
namespace MQTTnet.AspNetCore.Client.Tcp
{
public class SocketSender
@@ -1,12 +1,13 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>netstandard2.0</TargetFramework>
<TargetFrameworks>netstandard2.0;netcoreapp2.1</TargetFrameworks>
<Product />
<Company />
<Authors />
<PackageId />
<LangVersion>7.2</LangVersion>
<AllowUnsafeBlocks>true</AllowUnsafeBlocks>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|AnyCPU'">
@@ -15,6 +16,7 @@
<ItemGroup>
<PackageReference Include="Microsoft.AspNetCore.Connections.Abstractions" Version="2.1.3" />
<PackageReference Include="Microsoft.AspNetCore.Http.Connections" Version="1.0.3" />
<PackageReference Include="Microsoft.AspNetCore.WebSockets" Version="2.1.1" />
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="2.1.1" />
</ItemGroup>
@@ -1,13 +1,14 @@
using Microsoft.AspNetCore.Connections;
using Microsoft.AspNetCore.Http.Connections.Features;
using MQTTnet.Adapter;
using MQTTnet.AspNetCore.Client.Tcp;
using MQTTnet.Exceptions;
using MQTTnet.Formatter;
using MQTTnet.Packets;
using System;
using System.IO.Pipelines;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Formatter;
namespace MQTTnet.AspNetCore
{
@@ -17,36 +18,64 @@ namespace MQTTnet.AspNetCore
{
PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter));
Connection = connection ?? throw new ArgumentNullException(nameof(connection));
if (Connection.Transport != null)
{
_input = Connection.Transport.Input;
_output = Connection.Transport.Output;
}
_reader = new SpanBasedMqttPacketBodyReader();
}
public string Endpoint => Connection.ConnectionId;
public bool IsSecureConnection => false; // TODO: Fix detection (WS vs. WSS).
private PipeReader _input;
private PipeWriter _output;
private readonly SpanBasedMqttPacketBodyReader _reader;
public string Endpoint
{
get
{
var connection = Http?.HttpContext?.Connection;
if (connection == null)
{
return Connection.ConnectionId;
}
return $"{connection.RemoteIpAddress}:{connection.RemotePort}";
}
}
public bool IsSecureConnection => Http?.HttpContext?.Request?.IsHttps ?? false;
private IHttpContextFeature Http => Connection.Features.Get<IHttpContextFeature>();
public ConnectionContext Connection { get; }
public MqttPacketFormatterAdapter PacketFormatterAdapter { get; }
public long BytesSent { get; } // TODO: Fix calculation.
public long BytesReceived { get; } // TODO: Fix calculation.
public long BytesSent { get; set; }
public long BytesReceived { get; set; }
public Action ReadingPacketStartedCallback { get; set; }
public Action ReadingPacketCompletedCallback { get; set; }
private readonly SemaphoreSlim _writerSemaphore = new SemaphoreSlim(1, 1);
public Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
public async Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
{
if (Connection is TcpConnection tcp && !tcp.IsConnected)
{
return tcp.StartAsync();
await tcp.StartAsync().ConfigureAwait(false);
}
return Task.CompletedTask;
_input = Connection.Transport.Input;
_output = Connection.Transport.Output;
}
public Task DisconnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
{
Connection.Transport.Input.Complete();
Connection.Transport.Output.Complete();
_input?.Complete();
_output?.Complete();
return Task.CompletedTask;
}
@@ -79,8 +108,9 @@ namespace MQTTnet.AspNetCore
{
if (!buffer.IsEmpty)
{
if (PacketFormatterAdapter.TryDecode(buffer, out var packet, out consumed, out observed))
if (PacketFormatterAdapter.TryDecode(_reader, buffer, out var packet, out consumed, out observed, out var received))
{
BytesReceived += received;
return packet;
}
else
@@ -114,13 +144,20 @@ namespace MQTTnet.AspNetCore
public async Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout, CancellationToken cancellationToken)
{
var buffer = PacketFormatterAdapter.Encode(packet).AsMemory();
var output = Connection.Transport.Output;
var formatter = PacketFormatterAdapter;
await _writerSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
await output.WriteAsync(buffer, cancellationToken).ConfigureAwait(false);
var buffer = formatter.Encode(packet);
var msg = buffer.AsMemory();
var output = _output;
msg.CopyTo(output.GetMemory(msg.Length));
BytesSent += msg.Length;
PacketFormatterAdapter.FreeBuffer();
output.Advance(msg.Length);
await output.FlushAsync().ConfigureAwait(false);
}
finally
{
@@ -10,7 +10,7 @@ namespace MQTTnet.AspNetCore
{
public class MqttConnectionHandler : ConnectionHandler, IMqttServerAdapter
{
public Action<MqttServerAdapterClientAcceptedEventArgs> ClientAcceptedHandler { get; set; }
public Func<IMqttChannelAdapter, Task> ClientHandler { get; set; }
public override async Task OnConnectedAsync(ConnectionContext connection)
{
@@ -21,12 +21,15 @@ namespace MQTTnet.AspNetCore
transferFormatFeature.ActiveFormat = TransferFormat.Binary;
}
using (var adapter = new MqttConnectionContext(new MqttPacketFormatterAdapter(), connection))
var writer = new SpanBasedMqttPacketWriter();
var formatter = new MqttPacketFormatterAdapter(writer);
using (var adapter = new MqttConnectionContext(formatter, connection))
{
var args = new MqttServerAdapterClientAcceptedEventArgs(adapter);
ClientAcceptedHandler?.Invoke(args);
await args.SessionTask.ConfigureAwait(false);
var clientHandler = ClientHandler;
if (clientHandler != null)
{
await clientHandler(adapter).ConfigureAwait(false);
}
}
}
@@ -21,7 +21,7 @@ namespace MQTTnet.AspNetCore
_logger = logger.CreateChildLogger(nameof(MqttTcpServerAdapter));
}
public Action<MqttServerAdapterClientAcceptedEventArgs> ClientAcceptedHandler { get; set; }
public Func<IMqttChannelAdapter, Task> ClientHandler { get; set; }
public Task StartAsync(IMqttServerOptions options)
{
@@ -43,15 +43,16 @@ namespace MQTTnet.AspNetCore
var isSecureConnection = clientCertificate != null;
clientCertificate?.Dispose();
var channel = new MqttWebSocketChannel(webSocket, endpoint, isSecureConnection);
var channelAdapter = new MqttChannelAdapter(channel, new MqttPacketFormatterAdapter(), _logger.CreateChildLogger(nameof(MqttWebSocketServerAdapter)));
var eventArgs = new MqttServerAdapterClientAcceptedEventArgs(channelAdapter);
ClientAcceptedHandler?.Invoke(eventArgs);
if (eventArgs.SessionTask != null)
var clientHandler = ClientHandler;
if (clientHandler != null)
{
await eventArgs.SessionTask.ConfigureAwait(false);
var writer = new SpanBasedMqttPacketWriter();
var formatter = new MqttPacketFormatterAdapter(writer);
var channel = new MqttWebSocketChannel(webSocket, endpoint, isSecureConnection);
using (var channelAdapter = new MqttChannelAdapter(channel, formatter, _logger.CreateChildLogger(nameof(MqttWebSocketServerAdapter))))
{
await clientHandler(channelAdapter).ConfigureAwait(false);
}
}
}
+32 -13
View File
@@ -9,13 +9,20 @@ namespace MQTTnet.AspNetCore
{
public static class ReaderExtensions
{
public static bool TryDecode(this MqttPacketFormatterAdapter formatter, in ReadOnlySequence<byte> input, out MqttBasePacket packet, out SequencePosition consumed, out SequencePosition observed)
public static bool TryDecode(this MqttPacketFormatterAdapter formatter,
SpanBasedMqttPacketBodyReader reader,
in ReadOnlySequence<byte> input,
out MqttBasePacket packet,
out SequencePosition consumed,
out SequencePosition observed,
out int bytesRead)
{
if (formatter == null) throw new ArgumentNullException(nameof(formatter));
packet = null;
consumed = input.Start;
observed = input.End;
bytesRead = 0;
var copy = input;
if (copy.Length < 2)
@@ -24,7 +31,7 @@ namespace MQTTnet.AspNetCore
}
var fixedheader = copy.First.Span[0];
if (!TryReadBodyLength(ref copy, out var bodyLength))
if (!TryReadBodyLength(ref copy, out int headerLength, out var bodyLength))
{
return false;
}
@@ -35,34 +42,45 @@ namespace MQTTnet.AspNetCore
}
var bodySlice = copy.Slice(0, bodyLength);
var buffer = bodySlice.GetArray();
packet = formatter.Decode(new ReceivedMqttPacket(fixedheader, new MqttPacketBodyReader(buffer, 0, buffer.Length), buffer.Length + 2));
var buffer = bodySlice.GetMemory();
reader.SetBuffer(buffer);
var receivedMqttPacket = new ReceivedMqttPacket(fixedheader, reader, buffer.Length + 2);
if (formatter.ProtocolVersion == MqttProtocolVersion.Unknown)
{
formatter.DetectProtocolVersion(receivedMqttPacket);
}
packet = formatter.Decode(receivedMqttPacket);
consumed = bodySlice.End;
observed = bodySlice.End;
bytesRead = headerLength + bodyLength;
return true;
}
private static byte[] GetArray(this in ReadOnlySequence<byte> input)
private static ReadOnlyMemory<byte> GetMemory(this in ReadOnlySequence<byte> input)
{
if (input.IsSingleSegment)
{
return input.First.Span.ToArray();
return input.First;
}
// Should be rare
return input.ToArray();
}
private static bool TryReadBodyLength(ref ReadOnlySequence<byte> input, out int result)
private static bool TryReadBodyLength(ref ReadOnlySequence<byte> input, out int headerLength, out int bodyLength)
{
// Alorithm taken from https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os-complete.html.
var multiplier = 1;
var value = 0;
byte encodedByte;
var index = 1;
result = 0;
var temp = input.Slice(0, Math.Min(5, input.Length)).GetArray();
headerLength = 0;
bodyLength = 0;
var temp = input.Slice(0, Math.Min(5, input.Length)).GetMemory();
do
{
@@ -70,13 +88,13 @@ namespace MQTTnet.AspNetCore
{
return false;
}
encodedByte = temp[index];
encodedByte = temp.Span[index];
index++;
value += (byte)(encodedByte & 127) * multiplier;
if (multiplier > 128 * 128 * 128)
{
throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", temp.AsSpan(1, index).ToArray())}).");
throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", temp.Slice(1, index).ToArray())}).");
}
multiplier *= 128;
@@ -84,7 +102,8 @@ namespace MQTTnet.AspNetCore
input = input.Slice(index);
result = value;
headerLength = index;
bodyLength = value;
return true;
}
}
@@ -0,0 +1,124 @@
using MQTTnet.Exceptions;
using MQTTnet.Formatter;
using System;
using System.Buffers.Binary;
using System.Text;
namespace MQTTnet.AspNetCore
{
public class SpanBasedMqttPacketBodyReader : IMqttPacketBodyReader
{
private ReadOnlyMemory<byte> _buffer;
private int _offset;
public void SetBuffer(ReadOnlyMemory<byte> buffer)
{
_buffer = buffer;
_offset = 0;
}
public ulong Length => (ulong)_buffer.Length;
public bool EndOfStream => _buffer.Length.Equals(_offset);
public ulong Offset => (ulong)_offset;
public byte ReadByte()
{
return _buffer.Span[_offset++];
}
public byte[] ReadRemainingData()
{
return _buffer.Slice(_offset).ToArray();
}
public byte[] ReadWithLengthPrefix()
{
return ReadSegmentWithLengthPrefix().ToArray();
}
private ReadOnlySpan<byte> ReadSegmentWithLengthPrefix()
{
var span = _buffer.Span;
var length = BinaryPrimitives.ReadUInt16BigEndian(span.Slice(_offset));
var result = span.Slice(_offset+2, length);
_offset += 2 + length;
return result;
}
public unsafe string ReadStringWithLengthPrefix()
{
var buffer = ReadSegmentWithLengthPrefix();
if (buffer.Length == 0)
{
return string.Empty;
}
fixed (byte* bytes = &buffer.GetPinnableReference())
{
var result = Encoding.UTF8.GetString(bytes, buffer.Length);
return result;
}
}
public ushort ReadTwoByteInteger()
{
var result = BinaryPrimitives.ReadUInt16BigEndian(_buffer.Span.Slice(_offset));
_offset += 2;
return result;
}
public uint ReadFourByteInteger()
{
var result = BinaryPrimitives.ReadUInt32BigEndian(_buffer.Span.Slice(_offset));
_offset += 4;
return result;
}
public uint ReadVariableLengthInteger()
{
var multiplier = 1;
var value = 0U;
byte encodedByte;
do
{
encodedByte = ReadByte();
value += (uint)((encodedByte & 127) * multiplier);
if (multiplier > 2097152)
{
throw new MqttProtocolViolationException("Variable length integer is invalid.");
}
multiplier *= 128;
} while ((encodedByte & 128) != 0);
return value;
}
public bool ReadBoolean()
{
var buffer = ReadByte();
if (buffer == 0)
{
return false;
}
if (buffer == 1)
{
return true;
}
throw new MqttProtocolViolationException("Boolean values can be 0 or 1 only.");
}
public void Seek(ulong position)
{
_offset = (int)position;
}
}
}
@@ -0,0 +1,146 @@
using MQTTnet.Formatter;
using System;
using System.Buffers;
using System.Buffers.Binary;
using System.Text;
namespace MQTTnet.AspNetCore
{
public class SpanBasedMqttPacketWriter : IMqttPacketWriter
{
private readonly ArrayPool<byte> _pool;
public SpanBasedMqttPacketWriter()
{
_pool = ArrayPool<byte>.Create();
}
private byte[] _buffer;
private int _position;
public int Length { get; set; }
public void FreeBuffer()
{
_pool.Return(_buffer);
}
public byte[] GetBuffer()
{
return _buffer;
}
public void Reset(int v)
{
_buffer = _pool.Rent(1500);
Length = v;
_position = v;
}
public void Seek(int v)
{
_position = v;
}
public void Write(byte value)
{
GrowIfNeeded(1);
_buffer[_position] = value;
Commit(1);
}
public void Write(ushort value)
{
GrowIfNeeded(2);
BinaryPrimitives.WriteUInt16BigEndian(_buffer.AsSpan(_position), value);
Commit(2);
}
public void Write(IMqttPacketWriter propertyWriter)
{
if (propertyWriter == null) throw new ArgumentNullException(nameof(propertyWriter));
GrowIfNeeded(propertyWriter.Length);
Write(propertyWriter.GetBuffer(), 0, propertyWriter.Length);
Commit(propertyWriter.Length);
}
public void Write(byte[] payload, int start, int length)
{
GrowIfNeeded(length);
payload.AsSpan(start, length).CopyTo(_buffer.AsSpan(_position));
Commit(length);
}
public void WriteVariableLengthInteger(uint value)
{
GrowIfNeeded(4);
var x = value;
do
{
var encodedByte = x % 128;
x = x / 128;
if (x > 0)
{
encodedByte = encodedByte | 128;
}
_buffer[_position] = (byte)encodedByte;
Commit(1);
} while (x > 0);
}
public void WriteWithLengthPrefix(string value)
{
var bytesLength = Encoding.UTF8.GetByteCount(value ?? string.Empty);
GrowIfNeeded(bytesLength + 2);
Write((ushort)bytesLength);
Encoding.UTF8.GetBytes(value ?? string.Empty, 0, value?.Length ?? 0, _buffer, _position);
Commit(bytesLength);
}
public void WriteWithLengthPrefix(byte[] payload)
{
GrowIfNeeded(payload.Length + 2);
Write((ushort)payload.Length);
payload.CopyTo(_buffer, _position);
Commit(payload.Length);
}
private void Commit(int count)
{
if (_position == Length)
{
Length += count;
}
_position += count;
}
private void GrowIfNeeded(int requiredAdditional)
{
var requiredTotal = _position + requiredAdditional;
if (_buffer.Length >= requiredTotal)
{
return;
}
var newBufferLength = _buffer.Length;
while (newBufferLength < requiredTotal)
{
newBufferLength *= 2;
}
var newBuffer = _pool.Rent(newBufferLength);
Array.Copy(_buffer, newBuffer, _buffer.Length);
_pool.Return(_buffer);
_buffer = newBuffer;
}
}
}
@@ -45,6 +45,9 @@
/// </summary>
public RetainedApplicationMessagesModel RetainedApplicationMessages { get; set; } = new RetainedApplicationMessagesModel();
/// <summary>
/// Enables or disables the MQTTnet internal logging.
/// </summary>
public bool EnableDebugLogging { get; set; } = false;
}
}
@@ -60,13 +60,13 @@ namespace MQTTnet.Server.Configuration
{
if (IPv4 == "*")
{
address = IPAddress.Parse("0.0.0.0");
address = IPAddress.Any;
return true;
}
if (IPv4 == "localhost")
{
address = IPAddress.Parse("127.0.0.1");
address = IPAddress.Loopback;
return true;
}
@@ -81,10 +81,8 @@ namespace MQTTnet.Server.Configuration
address = ip;
return true;
}
else
{
throw new System.Exception($"Could not parse IPv4 address: {IPv4}");
}
throw new System.Exception($"Could not parse IPv4 address: {IPv4}");
}
/// <summary>
@@ -95,13 +93,13 @@ namespace MQTTnet.Server.Configuration
{
if (IPv6 == "*")
{
address = IPAddress.Parse("::");
address = IPAddress.IPv6Any;
return true;
}
if (IPv6 == "localhost")
{
address = IPAddress.Parse("::1");
address = IPAddress.IPv6Loopback;
return true;
}
@@ -116,10 +114,8 @@ namespace MQTTnet.Server.Configuration
address = ip;
return true;
}
else
{
throw new System.Exception($"Could not parse IPv6 address: {IPv6}");
}
throw new System.Exception($"Could not parse IPv6 address: {IPv6}");
}
}
}
@@ -0,0 +1,61 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Threading.Tasks;
using System.Web;
using Microsoft.AspNetCore.Mvc;
using MQTTnet.Server.Mqtt;
using MQTTnet.Server.Status;
namespace MQTTnet.Server.Controllers
{
[ApiController]
public class ClientsController : ControllerBase
{
private readonly MqttServerService _mqttServerService;
public ClientsController(MqttServerService mqttServerService)
{
_mqttServerService = mqttServerService ?? throw new ArgumentNullException(nameof(mqttServerService));
}
[Route("api/v1/clients")]
[HttpGet]
public async Task<ActionResult<IList<IMqttClientStatus>>> GetClients()
{
return new ObjectResult(await _mqttServerService.GetClientStatusAsync());
}
[Route("api/v1/clients/{clientId}")]
[HttpGet]
public async Task<ActionResult<IMqttClientStatus>> GetClient(string clientId)
{
clientId = HttpUtility.UrlDecode(clientId);
var client = (await _mqttServerService.GetClientStatusAsync()).FirstOrDefault(c => c.ClientId == clientId);
if (client == null)
{
return new StatusCodeResult((int)HttpStatusCode.NotFound);
}
return new ObjectResult(client);
}
[Route("api/v1/clients/{clientId}")]
[HttpDelete]
public async Task<ActionResult> DeleteClient(string clientId)
{
clientId = HttpUtility.UrlDecode(clientId);
var client = (await _mqttServerService.GetClientStatusAsync()).FirstOrDefault(c => c.ClientId == clientId);
if (client == null)
{
return new StatusCodeResult((int)HttpStatusCode.NotFound);
}
await client.DisconnectAsync();
return StatusCode((int)HttpStatusCode.NoContent);
}
}
}
@@ -0,0 +1,62 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Threading.Tasks;
using System.Web;
using Microsoft.AspNetCore.Mvc;
using MQTTnet.Server.Mqtt;
namespace MQTTnet.Server.Controllers
{
[ApiController]
public class RetainedApplicationMessagesController : ControllerBase
{
private readonly MqttServerService _mqttServerService;
public RetainedApplicationMessagesController(MqttServerService mqttServerService)
{
_mqttServerService = mqttServerService ?? throw new ArgumentNullException(nameof(mqttServerService));
}
[Route("api/v1/retainedApplicationMessages")]
[HttpGet]
public async Task<ActionResult<IList<MqttApplicationMessage>>> GetRetainedApplicationMessages()
{
return new ObjectResult(await _mqttServerService.GetRetainedApplicationMessagesAsync());
}
[Route("api/v1/retainedApplicationMessages/{topic}")]
[HttpGet]
public async Task<ActionResult<MqttApplicationMessage>> GetRetainedApplicationMessage(string topic)
{
topic = HttpUtility.UrlDecode(topic);
var applicationMessage = (await _mqttServerService.GetRetainedApplicationMessagesAsync()).FirstOrDefault(c => c.Topic == topic);
if (applicationMessage == null)
{
return new StatusCodeResult((int)HttpStatusCode.NotFound);
}
return new ObjectResult(applicationMessage);
}
[Route("api/v1/retainedApplicationMessages")]
[HttpDelete]
public async Task<ActionResult> DeleteRetainedApplicationMessages()
{
await _mqttServerService.ClearRetainedApplicationMessagesAsync();
return StatusCode((int)HttpStatusCode.NoContent);
}
[Route("api/v1/retainedApplicationMessages/{topic}")]
[HttpDelete]
public async Task<ActionResult> DeleteRetainedApplicationMessage(string topic)
{
topic = HttpUtility.UrlDecode(topic);
await _mqttServerService.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(topic).WithRetainFlag().Build());
return StatusCode((int)HttpStatusCode.NoContent);
}
}
}
@@ -0,0 +1,11 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
namespace MQTTnet.Server.Controllers
{
public class ScriptsController
{
}
}
@@ -0,0 +1,77 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Threading.Tasks;
using System.Web;
using Microsoft.AspNetCore.Mvc;
using MQTTnet.Server.Mqtt;
using MQTTnet.Server.Status;
namespace MQTTnet.Server.Controllers
{
[ApiController]
public class SessionsController : ControllerBase
{
private readonly MqttServerService _mqttServerService;
public SessionsController(MqttServerService mqttServerService)
{
_mqttServerService = mqttServerService ?? throw new ArgumentNullException(nameof(mqttServerService));
}
[Route("api/v1/sessions")]
[HttpGet]
public async Task<ActionResult<IList<IMqttSessionStatus>>> GetSessions()
{
return new ObjectResult(await _mqttServerService.GetSessionStatusAsync());
}
[Route("api/v1/sessions/{clientId}")]
[HttpGet]
public async Task<ActionResult<IMqttClientStatus>> GetSession(string clientId)
{
clientId = HttpUtility.UrlDecode(clientId);
var session = (await _mqttServerService.GetSessionStatusAsync()).FirstOrDefault(c => c.ClientId == clientId);
if (session == null)
{
return new StatusCodeResult((int)HttpStatusCode.NotFound);
}
return new ObjectResult(session);
}
[Route("api/v1/sessions/{clientId}")]
[HttpDelete]
public async Task<ActionResult> DeleteSession(string clientId)
{
clientId = HttpUtility.UrlDecode(clientId);
var session = (await _mqttServerService.GetSessionStatusAsync()).FirstOrDefault(c => c.ClientId == clientId);
if (session == null)
{
return new StatusCodeResult((int)HttpStatusCode.NotFound);
}
await session.DeleteAsync();
return StatusCode((int)HttpStatusCode.NoContent);
}
[Route("api/v1/sessions/{clientId}/pendingApplicationMessages")]
[HttpDelete]
public async Task<ActionResult> DeletePendingApplicationMessages(string clientId)
{
clientId = HttpUtility.UrlDecode(clientId);
var session = (await _mqttServerService.GetSessionStatusAsync()).FirstOrDefault(c => c.ClientId == clientId);
if (session == null)
{
return new StatusCodeResult((int)HttpStatusCode.NotFound);
}
await session.ClearPendingApplicationMessagesAsync();
return StatusCode((int)HttpStatusCode.NoContent);
}
}
}
@@ -1,42 +0,0 @@
using System.Collections.Generic;
using Microsoft.AspNetCore.Mvc;
namespace MQTTnet.Server.Controllers
{
[Route("api/[controller]")]
[ApiController]
public class ValuesController : ControllerBase
{
// GET api/values
[HttpGet]
public ActionResult<IEnumerable<string>> Get()
{
return new string[] { "value1", "value2" };
}
// GET api/values/5
[HttpGet("{id}")]
public ActionResult<string> Get(int id)
{
return "value";
}
// POST api/values
[HttpPost]
public void Post([FromBody] string value)
{
}
// PUT api/values/5
[HttpPut("{id}")]
public void Put(int id, [FromBody] string value)
{
}
// DELETE api/values/5
[HttpDelete("{id}")]
public void Delete(int id)
{
}
}
}
@@ -1,4 +1,5 @@
using System;
using System.Threading;
using Microsoft.Extensions.Logging;
using MQTTnet.Diagnostics;
@@ -14,7 +15,7 @@ namespace MQTTnet.Server.Logging
}
public event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;
public IMqttNetChildLogger CreateChildLogger(string source = null)
{
return new MqttNetChildLoggerWrapper(source, this);
@@ -24,8 +25,15 @@ namespace MQTTnet.Server.Logging
{
var convertedLogLevel = ConvertLogLevel(logLevel);
_logger.Log(convertedLogLevel, exception, message, parameters);
}
var logMessagePublishedEvent = LogMessagePublished;
if (logMessagePublishedEvent != null)
{
var logMessage = new MqttNetLogMessage(null, DateTime.UtcNow, Thread.CurrentThread.ManagedThreadId, source, logLevel, message, exception);
logMessagePublishedEvent.Invoke(this, new MqttNetLogMessagePublishedEventArgs(logMessage));
}
}
private static LogLevel ConvertLogLevel(MqttNetLogLevel logLevel)
{
switch (logLevel)
+2 -2
View File
@@ -40,8 +40,8 @@
<PackageReference Include="Microsoft.AspNetCore.Razor.Design" Version="2.2.0" PrivateAssets="All" />
<PackageReference Include="Microsoft.AspNetCore.StaticFiles" Version="2.2.0" />
<PackageReference Include="Microsoft.AspNetCore.WebSockets" Version="2.2.1" />
<PackageReference Include="Swashbuckle.AspNetCore.SwaggerGen" Version="4.0.1" />
<PackageReference Include="Swashbuckle.AspNetCore.SwaggerUI" Version="4.0.1" />
<PackageReference Include="Swashbuckle.AspNetCore.SwaggerGen" Version="5.0.0-rc2" />
<PackageReference Include="Swashbuckle.AspNetCore.SwaggerUI" Version="5.0.0-rc2" />
</ItemGroup>
<ItemGroup>
@@ -46,21 +46,5 @@ namespace MQTTnet.Server.Mqtt
return Task.CompletedTask;
}
// TODO: Create dump(object) method in wrapper (creates JSON and prints it).
public class PythonMqttApplicationMessageInterceptorContext
{
public bool accept_connection;
public bool accept_publish;
public string client_id;
public string topic;
public int qos;
public bool retain;
}
}
}
@@ -10,10 +10,12 @@ using Microsoft.AspNetCore.Http;
using Microsoft.Extensions.Logging;
using MQTTnet.Adapter;
using MQTTnet.AspNetCore;
using MQTTnet.Client.Publishing;
using MQTTnet.Implementations;
using MQTTnet.Protocol;
using MQTTnet.Server.Configuration;
using MQTTnet.Server.Scripting;
using MQTTnet.Server.Status;
namespace MQTTnet.Server.Mqtt
{
@@ -92,6 +94,33 @@ namespace MQTTnet.Server.Mqtt
return _webSocketServerAdapter.RunWebSocketConnectionAsync(webSocket, httpContext);
}
public Task<IList<IMqttClientStatus>> GetClientStatusAsync()
{
return _mqttServer.GetClientStatusAsync();
}
public Task<IList<IMqttSessionStatus>> GetSessionStatusAsync()
{
return _mqttServer.GetSessionStatusAsync();
}
public Task ClearRetainedApplicationMessagesAsync()
{
return _mqttServer.ClearRetainedApplicationMessagesAsync();
}
public Task<IList<MqttApplicationMessage>> GetRetainedApplicationMessagesAsync()
{
return _mqttServer.GetRetainedApplicationMessagesAsync();
}
public Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
return _mqttServer.PublishAsync(applicationMessage);
}
private void Publish(PythonDictionary parameters)
{
try
@@ -152,6 +181,7 @@ namespace MQTTnet.Server.Mqtt
if (_settings.TcpEndPoint.Enabled)
{
options.WithDefaultEndpoint();
if (_settings.TcpEndPoint.TryReadIPv4(out var address4))
{
options.WithDefaultEndpointBoundIPAddress(address4);
+20 -16
View File
@@ -1,4 +1,5 @@
using System;
using System.Reflection;
using Microsoft.AspNetCore;
using Microsoft.AspNetCore.Hosting;
@@ -12,7 +13,8 @@ namespace MQTTnet.Server
{
PrintLogo();
CreateWebHostBuilder(args).Build().Run();
WebHost.CreateDefaultBuilder(args).UseStartup<Startup>().Build().Run();
return 0;
}
catch (Exception exception)
@@ -22,23 +24,25 @@ namespace MQTTnet.Server
}
}
private static IWebHostBuilder CreateWebHostBuilder(string[] args) =>
WebHost.CreateDefaultBuilder(args)
.UseKestrel()
.UseStartup<Startup>();
private static void PrintLogo()
{
Console.ResetColor();
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine(@"
__ __ ____ _______ _______ _ _____
| \/ |/ __ \__ __|__ __| | | / ____|
| \ / | | | | | | | |_ __ ___| |_ | (___ ___ _ ____ _____ _ __
| |\/| | | | | | | | | '_ \ / _ \ __| \___ \ / _ \ '__\ \ / / _ \ '__|
| | | | |__| | | | | | | | | __/ |_ ____) | __/ | \ V / __/ |
|_| |_|\___\_\ |_| |_|_| |_|\___|\__| |_____/ \___|_| \_/ \___|_|");
Console.ForegroundColor = ConsoleColor.White;
Console.BackgroundColor = ConsoleColor.Red;
const string LogoText =
@"| |
| __ __ ____ _______ _______ _ _____ |
| | \/ |/ __ \__ __|__ __| | | / ____| |
| | \ / | | | | | | | |_ __ ___| |_ | (___ ___ _ ____ _____ _ __ |
| | |\/| | | | | | | | | '_ \ / _ \ __| \___ \ / _ \ '__\ \ / / _ \ '__| |
| | | | | |__| | | | | | | | | __/ |_ ____) | __/ | \ V / __/ | |
| |_| |_|\___\_\ |_| |_|_| |_|\___|\__| |_____/ \___|_| \_/ \___|_| |
| |
| |";
Console.WriteLine(LogoText);
Console.ResetColor();
Console.WriteLine();
Console.ForegroundColor = ConsoleColor.White;
Console.WriteLine(@"
-- The official MQTT server implementation of MQTTnet --
@@ -49,8 +53,8 @@ namespace MQTTnet.Server
https://github.com/chkr1011/MQTTnet");
Console.ForegroundColor = ConsoleColor.White;
Console.WriteLine(@"
Version: 1.0.0-alpha2
Console.WriteLine($@"
Version: {Assembly.GetExecutingAssembly().GetName().Version}
License: MIT (read LICENSE file)
Sponsoring: https://opencollective.com/mqttnet
Support: https://github.com/chkr1011/MQTTnet/issues
+42 -1
View File
@@ -5,12 +5,14 @@ using Microsoft.AspNetCore.Mvc;
using Microsoft.EntityFrameworkCore.Internal;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.OpenApi.Models;
using Microsoft.Scripting.Utils;
using MQTTnet.Server.Configuration;
using MQTTnet.Server.Logging;
using MQTTnet.Server.Mqtt;
using MQTTnet.Server.Scripting;
using MQTTnet.Server.Scripting.DataSharing;
using Swashbuckle.AspNetCore.SwaggerUI;
namespace MQTTnet.Server
{
@@ -55,11 +57,28 @@ namespace MQTTnet.Server
pythonScriptHostService.Configure();
mqttServerService.Configure();
application.UseSwagger(o => o.RouteTemplate = "/api/{documentName}/swagger.json");
application.UseSwaggerUI(o =>
{
o.RoutePrefix = "api";
o.DocumentTitle = "MQTTnet.Server API";
o.SwaggerEndpoint("/api/v1/swagger.json", "MQTTnet.Server API v1");
o.DisplayRequestDuration();
o.DocExpansion(DocExpansion.List);
o.DefaultModelRendering(ModelRendering.Model);
});
}
public void ConfigureServices(IServiceCollection services)
{
services.AddMvc().SetCompatibilityVersion(CompatibilityVersion.Version_2_2);
services.AddMvc()
.SetCompatibilityVersion(CompatibilityVersion.Version_2_2)
.AddJsonOptions(options =>
{
options.SerializerSettings.Converters.Add(new Newtonsoft.Json.Converters.StringEnumConverter());
});
services.AddSingleton(ReadSettings());
@@ -79,6 +98,28 @@ namespace MQTTnet.Server
services.AddSingleton<MqttConnectionValidator>();
services.AddSingleton<MqttSubscriptionInterceptor>();
services.AddSingleton<MqttApplicationMessageInterceptor>();
services.AddSwaggerGen(c =>
{
c.DescribeAllEnumsAsStrings();
c.SwaggerDoc("v1", new OpenApiInfo
{
Title = "MQTTnet.Server API",
Version = "v1",
Description = "The public API for the MQTT broker MQTTnet.Server.",
License = new OpenApiLicense
{
Name = "MIT",
Url = new Uri("https://github.com/chkr1011/MQTTnet/blob/master/README.md")
},
Contact = new OpenApiContact
{
Name = "MQTTnet.Server",
Email = string.Empty,
Url = new Uri("https://github.com/chkr1011/MQTTnet")
},
});
});
}
private SettingsModel ReadSettings()
+5 -5
View File
@@ -18,14 +18,14 @@
*/
"TcpEndPoint": {
"Enabled": true,
"IPv4": "localhost",
"IPv6": "localhost",
"IPv4": "*",
"IPv6": "*",
"Port": 1883
},
"EncryptedTcpEndPoint": {
"Enabled": false,
"IPv4": "localhost",
"IPv6": "localhost",
"IPv4": "*",
"IPv6": "*",
"Port": 8883,
"CertificatePath": "/absolute/path/to/pfx"
},
@@ -45,7 +45,7 @@
"Filename": "RetainedApplicationMessages.json",
"WriteInterval": 10 // In seconds.
},
"EnableDebugLogging": false
"EnableDebugLogging": true
},
"Logging": {
"LogLevel": {
+1 -1
View File
@@ -6,7 +6,7 @@ namespace MQTTnet.Adapter
{
public interface IMqttServerAdapter : IDisposable
{
Action<MqttServerAdapterClientAcceptedEventArgs> ClientAcceptedHandler { get; set; }
Func<IMqttChannelAdapter, Task> ClientHandler { get; set; }
Task StartAsync(IMqttServerOptions options);
Task StopAsync();
@@ -1,17 +0,0 @@
using System;
using System.Threading.Tasks;
namespace MQTTnet.Adapter
{
public class MqttServerAdapterClientAcceptedEventArgs : EventArgs
{
public MqttServerAdapterClientAcceptedEventArgs(IMqttChannelAdapter channelAdapter)
{
ChannelAdapter = channelAdapter ?? throw new ArgumentNullException(nameof(channelAdapter));
}
public IMqttChannelAdapter ChannelAdapter { get; }
public Task SessionTask { get; set; }
}
}
+3 -3
View File
@@ -4,16 +4,16 @@ namespace MQTTnet.Adapter
{
public class ReceivedMqttPacket
{
public ReceivedMqttPacket(byte fixedHeader, MqttPacketBodyReader body, int totalLength)
public ReceivedMqttPacket(byte fixedHeader, IMqttPacketBodyReader body, int totalLength)
{
FixedHeader = fixedHeader;
Body = body;
TotalLength = totalLength;
}
public byte FixedHeader { get; }
public byte FixedHeader { get; set; }
public MqttPacketBodyReader Body { get; }
public IMqttPacketBodyReader Body { get; }
public int TotalLength { get; }
}
+1 -1
View File
@@ -325,7 +325,7 @@ namespace MQTTnet.Client
}
catch (MqttCommunicationTimedOutException)
{
_logger.Warning(null, "Timeout while waiting for packet of type '{0}'.", typeof(TResponsePacket).Namespace);
_logger.Warning(null, "Timeout while waiting for packet of type '{0}'.", typeof(TResponsePacket).Name);
throw;
}
}
@@ -15,6 +15,11 @@ namespace MQTTnet.Client.Options
public MqttClientOptionsBuilder WithProtocolVersion(MqttProtocolVersion value)
{
if (value == MqttProtocolVersion.Unknown)
{
throw new ArgumentException("Protocol version is invalid.");
}
_options.ProtocolVersion = value;
return this;
}
@@ -8,7 +8,7 @@ namespace MQTTnet.Client.Options
public int? Port { get; set; }
public int BufferSize { get; set; } = 1024;
public int BufferSize { get; set; } = 65536;
public bool DualMode { get; set; } = true;
@@ -1,8 +1,17 @@
using System.Collections.Generic;
using MQTTnet.Packets;
namespace MQTTnet.Client.Publishing
{
public class MqttClientPublishResult
{
public ushort? PacketIdentifier { get; set; }
public MqttClientPublishReasonCode ReasonCode { get; set; } = MqttClientPublishReasonCode.Success;
public string ReasonString { get; set; }
public List<MqttUserProperty> UserProperties { get; set; }
}
}
@@ -12,6 +12,8 @@ namespace MQTTnet.Formatter
{
MqttPublishPacket CreatePublishPacket(MqttApplicationMessage applicationMessage);
MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket);
MqttApplicationMessage CreateApplicationMessage(MqttPublishPacket publishPacket);
MqttClientAuthenticateResult CreateClientConnectResult(MqttConnAckPacket connAckPacket);
@@ -0,0 +1,31 @@
using System;
namespace MQTTnet.Formatter
{
public interface IMqttPacketBodyReader
{
ulong Length { get; }
ulong Offset { get; }
bool EndOfStream { get; }
byte ReadByte();
byte[] ReadRemainingData();
ushort ReadTwoByteInteger();
string ReadStringWithLengthPrefix();
byte[] ReadWithLengthPrefix();
uint ReadFourByteInteger();
uint ReadVariableLengthInteger();
bool ReadBoolean();
void Seek(ulong position);
}
}
@@ -0,0 +1,20 @@
namespace MQTTnet.Formatter
{
public interface IMqttPacketWriter
{
int Length { get; }
void WriteWithLengthPrefix(string value);
void Write(byte returnCode);
void WriteWithLengthPrefix(byte[] payload);
void Write(ushort keepAlivePeriod);
void Write(IMqttPacketWriter propertyWriter);
void WriteVariableLengthInteger(uint length);
void Write(byte[] payload, int v, int length);
void Reset(int v);
void Seek(int v);
void FreeBuffer();
byte[] GetBuffer();
}
}
@@ -1,11 +1,12 @@
using System;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Text;
using MQTTnet.Exceptions;
namespace MQTTnet.Formatter
{
public class MqttPacketBodyReader
public class MqttPacketBodyReader : IMqttPacketBodyReader
{
private readonly byte[] _buffer;
private readonly ulong _initialOffset;
@@ -68,9 +69,9 @@ namespace MQTTnet.Formatter
throw new MqttProtocolViolationException("Boolean values can be 0 or 1 only.");
}
public ArraySegment<byte> ReadRemainingData()
public byte[] ReadRemainingData()
{
return new ArraySegment<byte>(_buffer, (int)Offset, (int)(_length - Offset));
return new ArraySegment<byte>(_buffer, (int)Offset, (int)(_length - Offset)).ToArray();
}
public ushort ReadTwoByteInteger()
@@ -95,7 +96,12 @@ namespace MQTTnet.Formatter
return (uint)(byte0 << 24 | byte1 << 16 | byte2 << 8 | byte3);
}
public ArraySegment<byte> ReadWithLengthPrefix()
public byte[] ReadWithLengthPrefix()
{
return ReadSegmentWithLengthPrefix().ToArray();
}
private ArraySegment<byte> ReadSegmentWithLengthPrefix()
{
var length = ReadTwoByteInteger();
@@ -140,7 +146,7 @@ namespace MQTTnet.Formatter
public string ReadStringWithLengthPrefix()
{
var buffer = ReadWithLengthPrefix();
var buffer = ReadSegmentWithLengthPrefix();
return Encoding.UTF8.GetString(buffer.Array, buffer.Offset, buffer.Count);
}
}
@@ -12,14 +12,26 @@ namespace MQTTnet.Formatter
private IMqttPacketFormatter _formatter;
public MqttPacketFormatterAdapter()
: this(new MqttPacketWriter())
{
}
public MqttPacketFormatterAdapter(MqttProtocolVersion protocolVersion)
: this(protocolVersion, new MqttPacketWriter())
{
}
public MqttPacketFormatterAdapter(MqttProtocolVersion protocolVersion)
public MqttPacketFormatterAdapter(MqttProtocolVersion protocolVersion, IMqttPacketWriter writer)
: this(writer)
{
UseProtocolVersion(protocolVersion);
}
public MqttPacketFormatterAdapter(IMqttPacketWriter writer)
{
Writer = writer;
}
public MqttProtocolVersion ProtocolVersion { get; private set; } = MqttProtocolVersion.Unknown;
public IMqttDataConverter DataConverter
@@ -31,6 +43,8 @@ namespace MQTTnet.Formatter
return _formatter.DataConverter;
}
}
public IMqttPacketWriter Writer { get; }
public ArraySegment<byte> Encode(MqttBasePacket packet)
{
@@ -75,28 +89,30 @@ namespace MQTTnet.Formatter
}
ProtocolVersion = protocolVersion;
_formatter = GetMqttPacketFormatter(protocolVersion, Writer);
}
public static IMqttPacketFormatter GetMqttPacketFormatter(MqttProtocolVersion protocolVersion, IMqttPacketWriter writer)
{
if (protocolVersion == MqttProtocolVersion.Unknown)
{
throw new InvalidOperationException("MQTT protocol version is invalid.");
}
switch (protocolVersion)
{
case MqttProtocolVersion.V500:
{
_formatter = new MqttV500PacketFormatter();
break;
return new MqttV500PacketFormatter(writer);
}
case MqttProtocolVersion.V311:
{
_formatter = new MqttV311PacketFormatter();
break;
return new MqttV311PacketFormatter(writer);
}
case MqttProtocolVersion.V310:
{
_formatter = new MqttV310PacketFormatter();
break;
return new MqttV310PacketFormatter(writer);
}
default:
{
throw new NotSupportedException();
+29 -14
View File
@@ -11,7 +11,7 @@ namespace MQTTnet.Formatter
/// same as for the original MemoryStream in .net. Also this implementation allows accessing the internal
/// buffer for all platforms and .net framework versions (which is not available at the regular MemoryStream).
/// </summary>
public class MqttPacketWriter
public class MqttPacketWriter : IMqttPacketWriter
{
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);
@@ -19,7 +19,6 @@ namespace MQTTnet.Formatter
public static int InitialBufferSize = 128;
public static int MaxBufferSize = 4096;
// TODO: Consider using the ArrayPool here together with FreeBuffer.
private byte[] _buffer = new byte[InitialBufferSize];
private int _offset;
@@ -33,6 +32,19 @@ namespace MQTTnet.Formatter
return (byte)fixedHeader;
}
public static int GetLengthOfVariableInteger(uint value)
{
var result = 0;
var x = value;
do
{
x = x / 128;
result++;
} while (x > 0);
return result;
}
public static ArraySegment<byte> EncodeVariableLengthInteger(uint value)
{
if (value == 0)
@@ -79,24 +91,21 @@ namespace MQTTnet.Formatter
else
{
WriteWithLengthPrefix(Encoding.UTF8.GetBytes(value));
}
}
}
public void WriteWithLengthPrefix(byte[] value)
{
if (value == null) throw new ArgumentNullException(nameof(value));
EnsureAdditionalCapacity(value.Length + 2);
if (value.Length == 0)
if (value == null || value.Length == 0)
{
Write(ZeroTwoByteIntegerArray);
Write(ZeroTwoByteIntegerArray);
}
else
{
EnsureAdditionalCapacity(value.Length + 2);
Write((ushort)value.Length);
Write(value, 0, value.Length);
}
}
}
public void Write(byte @byte)
@@ -131,17 +140,23 @@ namespace MQTTnet.Formatter
Array.Copy(buffer, offset, _buffer, _offset, count);
IncreasePosition(count);
}
public void Write(MqttPacketWriter propertyWriter)
public void Write(IMqttPacketWriter propertyWriter)
{
if (propertyWriter == null) throw new ArgumentNullException(nameof(propertyWriter));
if (propertyWriter.Length == 0)
if (propertyWriter is MqttPacketWriter writer)
{
if (writer.Length == 0)
{
return;
}
Write(writer._buffer, 0, writer.Length);
return;
}
Write(propertyWriter._buffer, 0, propertyWriter.Length);
throw new InvalidOperationException($"{nameof(propertyWriter)} must be of type {typeof(MqttPacketWriter).Name}");
}
public void Reset(int length)
@@ -33,6 +33,15 @@ namespace MQTTnet.Formatter.V3
};
}
public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket)
{
return new MqttPubAckPacket
{
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = MqttPubAckReasonCode.Success
};
}
public MqttApplicationMessage CreateApplicationMessage(MqttPublishPacket publishPacket)
{
if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
@@ -190,6 +199,7 @@ namespace MQTTnet.Formatter.V3
{
return new MqttClientPublishResult
{
PacketIdentifier = pubAckPacket?.PacketIdentifier,
ReasonCode = MqttClientPublishReasonCode.Success
};
}
@@ -206,6 +216,7 @@ namespace MQTTnet.Formatter.V3
return new MqttClientPublishResult
{
PacketIdentifier = pubCompPacket.PacketIdentifier,
ReasonCode = MqttClientPublishReasonCode.Success
};
}
@@ -2,7 +2,6 @@
using System.Linq;
using MQTTnet.Adapter;
using MQTTnet.Exceptions;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
@@ -11,8 +10,20 @@ namespace MQTTnet.Formatter.V3
public class MqttV310PacketFormatter : IMqttPacketFormatter
{
private const int FixedHeaderSize = 1;
private readonly IMqttPacketWriter _packetWriter;
public MqttV310PacketFormatter()
: this(new MqttPacketWriter())
{
}
public MqttV310PacketFormatter(IMqttPacketWriter packetWriter)
{
_packetWriter = packetWriter;
}
private readonly MqttPacketWriter _packetWriter = new MqttPacketWriter();
public IMqttDataConverter DataConverter { get; } = new MqttV310DataConverter();
@@ -27,17 +38,18 @@ namespace MQTTnet.Formatter.V3
var fixedHeader = EncodePacket(packet, _packetWriter);
var remainingLength = (uint)(_packetWriter.Length - 5);
var remainingLengthBuffer = MqttPacketWriter.EncodeVariableLengthInteger(remainingLength);
var remainingLengthSize = MqttPacketWriter.GetLengthOfVariableInteger(remainingLength);
var headerSize = FixedHeaderSize + remainingLengthBuffer.Count;
var headerSize = FixedHeaderSize + remainingLengthSize;
var headerOffset = 5 - headerSize;
// Position cursor on correct offset on beginning of array (has leading 0x0)
_packetWriter.Seek(headerOffset);
_packetWriter.Write(fixedHeader);
_packetWriter.Write(remainingLengthBuffer.Array, remainingLengthBuffer.Offset, remainingLengthBuffer.Count);
_packetWriter.WriteVariableLengthInteger(remainingLength);
var buffer = _packetWriter.GetBuffer();
return new ArraySegment<byte>(buffer, headerOffset, _packetWriter.Length - headerOffset);
}
@@ -56,17 +68,17 @@ namespace MQTTnet.Formatter.V3
case MqttControlPacketType.Connect: return DecodeConnectPacket(receivedMqttPacket.Body);
case MqttControlPacketType.ConnAck: return DecodeConnAckPacket(receivedMqttPacket.Body);
case MqttControlPacketType.Disconnect: return new MqttDisconnectPacket();
case MqttControlPacketType.Publish: return DecodePublish(receivedMqttPacket);
case MqttControlPacketType.PubAck: return DecodePubAck(receivedMqttPacket.Body);
case MqttControlPacketType.PubRec: return DecodePubRec(receivedMqttPacket.Body);
case MqttControlPacketType.PubRel: return DecodePubRel(receivedMqttPacket.Body);
case MqttControlPacketType.PubComp: return DecodePubComp(receivedMqttPacket.Body);
case MqttControlPacketType.Publish: return DecodePublishPacket(receivedMqttPacket);
case MqttControlPacketType.PubAck: return DecodePubAckPacket(receivedMqttPacket.Body);
case MqttControlPacketType.PubRec: return DecodePubRecPacket(receivedMqttPacket.Body);
case MqttControlPacketType.PubRel: return DecodePubRelPacket(receivedMqttPacket.Body);
case MqttControlPacketType.PubComp: return DecodePubCompPacket(receivedMqttPacket.Body);
case MqttControlPacketType.PingReq: return new MqttPingReqPacket();
case MqttControlPacketType.PingResp: return new MqttPingRespPacket();
case MqttControlPacketType.Subscribe: return DecodeSubscribe(receivedMqttPacket.Body);
case MqttControlPacketType.SubAck: return DecodeSubAck(receivedMqttPacket.Body);
case MqttControlPacketType.Unsubscibe: return DecodeUnsubscribe(receivedMqttPacket.Body);
case MqttControlPacketType.UnsubAck: return DecodeUnsubAck(receivedMqttPacket.Body);
case MqttControlPacketType.Subscribe: return DecodeSubscribePacket(receivedMqttPacket.Body);
case MqttControlPacketType.SubAck: return DecodeSubAckPacket(receivedMqttPacket.Body);
case MqttControlPacketType.Unsubscibe: return DecodeUnsubscribePacket(receivedMqttPacket.Body);
case MqttControlPacketType.UnsubAck: return DecodeUnsubAckPacket(receivedMqttPacket.Body);
default: throw new MqttProtocolViolationException($"Packet type ({controlPacketType}) not supported.");
}
@@ -77,7 +89,7 @@ namespace MQTTnet.Formatter.V3
_packetWriter.FreeBuffer();
}
private byte EncodePacket(MqttBasePacket packet, MqttPacketWriter packetWriter)
private byte EncodePacket(MqttBasePacket packet, IMqttPacketWriter packetWriter)
{
switch (packet)
{
@@ -100,7 +112,7 @@ namespace MQTTnet.Formatter.V3
}
}
private static MqttBasePacket DecodeUnsubAck(MqttPacketBodyReader body)
private static MqttBasePacket DecodeUnsubAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -110,7 +122,7 @@ namespace MQTTnet.Formatter.V3
};
}
private static MqttBasePacket DecodePubComp(MqttPacketBodyReader body)
private static MqttBasePacket DecodePubCompPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -120,7 +132,7 @@ namespace MQTTnet.Formatter.V3
};
}
private static MqttBasePacket DecodePubRel(MqttPacketBodyReader body)
private static MqttBasePacket DecodePubRelPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -130,7 +142,7 @@ namespace MQTTnet.Formatter.V3
};
}
private static MqttBasePacket DecodePubRec(MqttPacketBodyReader body)
private static MqttBasePacket DecodePubRecPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -140,7 +152,7 @@ namespace MQTTnet.Formatter.V3
};
}
private static MqttBasePacket DecodePubAck(MqttPacketBodyReader body)
private static MqttBasePacket DecodePubAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -150,7 +162,7 @@ namespace MQTTnet.Formatter.V3
};
}
private static MqttBasePacket DecodeUnsubscribe(MqttPacketBodyReader body)
private static MqttBasePacket DecodeUnsubscribePacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -167,7 +179,7 @@ namespace MQTTnet.Formatter.V3
return packet;
}
private static MqttBasePacket DecodeSubscribe(MqttPacketBodyReader body)
private static MqttBasePacket DecodeSubscribePacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -190,7 +202,7 @@ namespace MQTTnet.Formatter.V3
return packet;
}
private static MqttBasePacket DecodePublish(ReceivedMqttPacket receivedMqttPacket)
private static MqttBasePacket DecodePublishPacket(ReceivedMqttPacket receivedMqttPacket)
{
ThrowIfBodyIsEmpty(receivedMqttPacket.Body);
@@ -211,23 +223,36 @@ namespace MQTTnet.Formatter.V3
PacketIdentifier = packetIdentifier,
Retain = retain,
Topic = topic,
Payload = receivedMqttPacket.Body.ReadRemainingData().ToArray(),
QualityOfServiceLevel = qualityOfServiceLevel,
Dup = dup
};
if (!receivedMqttPacket.Body.EndOfStream)
{
packet.Payload = receivedMqttPacket.Body.ReadRemainingData().ToArray();
}
return packet;
}
private MqttBasePacket DecodeConnectPacket(MqttPacketBodyReader body)
private MqttBasePacket DecodeConnectPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
var packet = new MqttConnectPacket
var protocolName = body.ReadStringWithLengthPrefix();
var protocolVersion = body.ReadByte();
if (protocolName != "MQTT" && protocolName != "MQIsdp")
{
ProtocolName = body.ReadStringWithLengthPrefix(),
ProtocolLevel = body.ReadByte()
};
throw new MqttProtocolViolationException("MQTT protocol name do not match MQTT v3.");
}
if (protocolVersion != 3 && protocolVersion != 4)
{
throw new MqttProtocolViolationException("MQTT protocol version do not match MQTT v3.");
}
var packet = new MqttConnectPacket();
var connectFlags = body.ReadByte();
if ((connectFlags & 0x1) > 0)
@@ -251,7 +276,7 @@ namespace MQTTnet.Formatter.V3
packet.WillMessage = new MqttApplicationMessage
{
Topic = body.ReadStringWithLengthPrefix(),
Payload = body.ReadWithLengthPrefix().ToArray(),
Payload = body.ReadWithLengthPrefix(),
QualityOfServiceLevel = (MqttQualityOfServiceLevel)willQoS,
Retain = willRetain
};
@@ -271,7 +296,7 @@ namespace MQTTnet.Formatter.V3
return packet;
}
private static MqttBasePacket DecodeSubAck(MqttPacketBodyReader body)
private static MqttBasePacket DecodeSubAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -288,7 +313,7 @@ namespace MQTTnet.Formatter.V3
return packet;
}
protected virtual MqttBasePacket DecodeConnAckPacket(MqttPacketBodyReader body)
protected virtual MqttBasePacket DecodeConnAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -319,7 +344,7 @@ namespace MQTTnet.Formatter.V3
}
}
protected virtual byte EncodeConnectPacket(MqttConnectPacket packet, MqttPacketWriter packetWriter)
protected virtual byte EncodeConnectPacket(MqttConnectPacket packet, IMqttPacketWriter packetWriter)
{
ValidateConnectPacket(packet);
@@ -381,7 +406,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Connect);
}
protected virtual byte EncodeConnAckPacket(MqttConnAckPacket packet, MqttPacketWriter packetWriter)
protected virtual byte EncodeConnAckPacket(MqttConnAckPacket packet, IMqttPacketWriter packetWriter)
{
packetWriter.Write(0); // Reserved.
packetWriter.Write((byte)packet.ReturnCode.Value);
@@ -389,7 +414,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck);
}
private static byte EncodePubRelPacket(MqttPubRelPacket packet, MqttPacketWriter packetWriter)
private static byte EncodePubRelPacket(MqttPubRelPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.PacketIdentifier.HasValue)
{
@@ -401,7 +426,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02);
}
private static byte EncodePublishPacket(MqttPublishPacket packet, MqttPacketWriter packetWriter)
private static byte EncodePublishPacket(MqttPublishPacket packet, IMqttPacketWriter packetWriter)
{
ValidatePublishPacket(packet);
@@ -446,7 +471,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader);
}
private static byte EncodePubAckPacket(MqttPubAckPacket packet, MqttPacketWriter packetWriter)
private static byte EncodePubAckPacket(MqttPubAckPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.PacketIdentifier.HasValue)
{
@@ -458,7 +483,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubAck);
}
private static byte EncodePubRecPacket(MqttPubRecPacket packet, MqttPacketWriter packetWriter)
private static byte EncodePubRecPacket(MqttPubRecPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.PacketIdentifier.HasValue)
{
@@ -470,7 +495,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRec);
}
private static byte EncodePubCompPacket(MqttPubCompPacket packet, MqttPacketWriter packetWriter)
private static byte EncodePubCompPacket(MqttPubCompPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.PacketIdentifier.HasValue)
{
@@ -482,7 +507,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubComp);
}
private static byte EncodeSubscribePacket(MqttSubscribePacket packet, MqttPacketWriter packetWriter)
private static byte EncodeSubscribePacket(MqttSubscribePacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.8.3-3].");
@@ -505,7 +530,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Subscribe, 0x02);
}
private static byte EncodeSubAckPacket(MqttSubAckPacket packet, MqttPacketWriter packetWriter)
private static byte EncodeSubAckPacket(MqttSubAckPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.PacketIdentifier.HasValue)
{
@@ -525,7 +550,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.SubAck);
}
private static byte EncodeUnsubscribePacket(MqttUnsubscribePacket packet, MqttPacketWriter packetWriter)
private static byte EncodeUnsubscribePacket(MqttUnsubscribePacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.10.3-2].");
@@ -547,7 +572,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Unsubscibe, 0x02);
}
private static byte EncodeUnsubAckPacket(MqttUnsubAckPacket packet, MqttPacketWriter packetWriter)
private static byte EncodeUnsubAckPacket(MqttUnsubAckPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.PacketIdentifier.HasValue)
{
@@ -564,7 +589,7 @@ namespace MQTTnet.Formatter.V3
}
// ReSharper disable once ParameterOnlyUsedForPreconditionCheck.Local
protected static void ThrowIfBodyIsEmpty(MqttPacketBodyReader body)
protected static void ThrowIfBodyIsEmpty(IMqttPacketBodyReader body)
{
if (body == null || body.Length == 0)
{
@@ -6,7 +6,17 @@ namespace MQTTnet.Formatter.V3
{
public class MqttV311PacketFormatter : MqttV310PacketFormatter
{
protected override byte EncodeConnectPacket(MqttConnectPacket packet, MqttPacketWriter packetWriter)
public MqttV311PacketFormatter()
: base()
{
}
public MqttV311PacketFormatter(IMqttPacketWriter packetWriter)
: base(packetWriter)
{
}
protected override byte EncodeConnectPacket(MqttConnectPacket packet, IMqttPacketWriter packetWriter)
{
ValidateConnectPacket(packet);
@@ -68,7 +78,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Connect);
}
protected override byte EncodeConnAckPacket(MqttConnAckPacket packet, MqttPacketWriter packetWriter)
protected override byte EncodeConnAckPacket(MqttConnAckPacket packet, IMqttPacketWriter packetWriter)
{
byte connectAcknowledgeFlags = 0x0;
if (packet.IsSessionPresent)
@@ -82,7 +92,7 @@ namespace MQTTnet.Formatter.V3
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck);
}
protected override MqttBasePacket DecodeConnAckPacket(MqttPacketBodyReader body)
protected override MqttBasePacket DecodeConnAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -46,6 +46,15 @@ namespace MQTTnet.Formatter.V5
return packet;
}
public MqttPubAckPacket CreatePubAckPacket(MqttPublishPacket publishPacket)
{
return new MqttPubAckPacket
{
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = MqttPubAckReasonCode.Success
};
}
public MqttApplicationMessage CreateApplicationMessage(MqttPublishPacket publishPacket)
{
return new MqttApplicationMessage
@@ -180,13 +189,22 @@ namespace MQTTnet.Formatter.V5
public MqttClientPublishResult CreatePublishResult(MqttPubAckPacket pubAckPacket)
{
var result = new MqttClientPublishResult();
if (pubAckPacket?.ReasonCode != null)
var result = new MqttClientPublishResult
{
result.ReasonCode = (MqttClientPublishReasonCode)pubAckPacket.ReasonCode;
}
ReasonCode = MqttClientPublishReasonCode.Success,
ReasonString = pubAckPacket?.Properties?.ReasonString,
UserProperties = pubAckPacket?.Properties?.UserProperties
};
if (pubAckPacket != null)
{
// QoS 0 has no response. So we treat it as a success always.
// Both enums have the same values. So it can be easily converted.
result.ReasonCode = (MqttClientPublishReasonCode)pubAckPacket.ReasonCode;
result.PacketIdentifier = pubAckPacket.PacketIdentifier;
}
return result;
}
@@ -200,17 +218,24 @@ namespace MQTTnet.Formatter.V5
};
}
// The PUBCOMP is the last packet in QoS 2. So we use the results from that instead of PUBREC.
if (pubCompPacket.ReasonCode == MqttPubCompReasonCode.PacketIdentifierNotFound)
{
return new MqttClientPublishResult
{
ReasonCode = MqttClientPublishReasonCode.UnspecifiedError
PacketIdentifier = pubCompPacket.PacketIdentifier,
ReasonCode = MqttClientPublishReasonCode.UnspecifiedError,
ReasonString = pubCompPacket.Properties?.ReasonString,
UserProperties = pubCompPacket.Properties?.UserProperties
};
}
var result = new MqttClientPublishResult
{
ReasonCode = MqttClientPublishReasonCode.Success
PacketIdentifier = pubCompPacket.PacketIdentifier,
ReasonCode = MqttClientPublishReasonCode.Success,
ReasonString = pubCompPacket.Properties?.ReasonString,
UserProperties = pubCompPacket.Properties?.UserProperties
};
if (pubRecPacket.ReasonCode.HasValue)
@@ -41,7 +41,7 @@ namespace MQTTnet.Formatter.V5
}
}
private static MqttBasePacket DecodeConnectPacket(MqttPacketBodyReader body)
private static MqttBasePacket DecodeConnectPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -134,7 +134,7 @@ namespace MQTTnet.Formatter.V5
if (packet.WillMessage != null)
{
packet.WillMessage.Topic = body.ReadStringWithLengthPrefix();
packet.WillMessage.Payload = body.ReadWithLengthPrefix().ToArray();
packet.WillMessage.Payload = body.ReadWithLengthPrefix();
}
if (usernameFlag)
@@ -150,7 +150,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
private static MqttBasePacket DecodeConnAckPacket(MqttPacketBodyReader body)
private static MqttBasePacket DecodeConnAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -239,7 +239,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
private static MqttBasePacket DecodeDisconnectPacket(MqttPacketBodyReader body)
private static MqttBasePacket DecodeDisconnectPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -277,7 +277,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
private static MqttBasePacket DecodeSubscribePacket(MqttPacketBodyReader body)
private static MqttBasePacket DecodeSubscribePacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -327,7 +327,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
private static MqttBasePacket DecodeSubAckPacket(MqttPacketBodyReader body)
private static MqttBasePacket DecodeSubAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -363,7 +363,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
private static MqttBasePacket DecodeUnsubscribePacket(MqttPacketBodyReader body)
private static MqttBasePacket DecodeUnsubscribePacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -394,7 +394,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
private static MqttBasePacket DecodeUnsubAckPacket(MqttPacketBodyReader body)
private static MqttBasePacket DecodeUnsubAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -440,7 +440,7 @@ namespace MQTTnet.Formatter.V5
return new MqttPingRespPacket();
}
private static MqttBasePacket DecodePublishPacket(byte header, MqttPacketBodyReader body)
private static MqttBasePacket DecodePublishPacket(byte header, IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -503,12 +503,15 @@ namespace MQTTnet.Formatter.V5
}
}
packet.Payload = body.ReadRemainingData().ToArray();
if (!body.EndOfStream)
{
packet.Payload = body.ReadRemainingData().ToArray();
}
return packet;
}
private static MqttBasePacket DecodePubAckPacket(MqttPacketBodyReader body)
private static MqttBasePacket DecodePubAckPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -546,7 +549,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
private static MqttBasePacket DecodePubRecPacket(MqttPacketBodyReader body)
private static MqttBasePacket DecodePubRecPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -584,7 +587,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
private static MqttBasePacket DecodePubRelPacket(MqttPacketBodyReader body)
private static MqttBasePacket DecodePubRelPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -622,7 +625,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
private static MqttBasePacket DecodePubCompPacket(MqttPacketBodyReader body)
private static MqttBasePacket DecodePubCompPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -660,7 +663,7 @@ namespace MQTTnet.Formatter.V5
return packet;
}
private static MqttBasePacket DecodeAuthPacket(MqttPacketBodyReader body)
private static MqttBasePacket DecodeAuthPacket(IMqttPacketBodyReader body)
{
ThrowIfBodyIsEmpty(body);
@@ -706,7 +709,7 @@ namespace MQTTnet.Formatter.V5
}
// ReSharper disable once ParameterOnlyUsedForPreconditionCheck.Local
private static void ThrowIfBodyIsEmpty(MqttPacketBodyReader body)
private static void ThrowIfBodyIsEmpty(IMqttPacketBodyReader body)
{
if (body == null || body.Length == 0)
{
@@ -8,7 +8,19 @@ namespace MQTTnet.Formatter.V5
{
public class MqttV500PacketEncoder
{
private readonly MqttPacketWriter _packetWriter = new MqttPacketWriter();
private readonly IMqttPacketWriter _packetWriter;
public MqttV500PacketEncoder()
: this(new MqttPacketWriter())
{
}
public MqttV500PacketEncoder(IMqttPacketWriter packetWriter)
{
_packetWriter = packetWriter;
}
public ArraySegment<byte> Encode(MqttBasePacket packet)
{
@@ -21,15 +33,15 @@ namespace MQTTnet.Formatter.V5
var fixedHeader = EncodePacket(packet, _packetWriter);
var remainingLength = (uint)(_packetWriter.Length - 5);
var remainingLengthBuffer = MqttPacketWriter.EncodeVariableLengthInteger(remainingLength);
var headerSize = 1 + remainingLengthBuffer.Count;
var remainingLengthSize = MqttPacketWriter.GetLengthOfVariableInteger(remainingLength);
var headerSize = 1 + remainingLengthSize;
var headerOffset = 5 - headerSize;
// Position cursor on correct offset on beginning of array (has leading 0x0)
_packetWriter.Seek(headerOffset);
_packetWriter.Write(fixedHeader);
_packetWriter.Write(remainingLengthBuffer.Array, remainingLengthBuffer.Offset, remainingLengthBuffer.Count);
_packetWriter.WriteVariableLengthInteger(remainingLength);
var buffer = _packetWriter.GetBuffer();
return new ArraySegment<byte>(buffer, headerOffset, _packetWriter.Length - headerOffset);
@@ -40,7 +52,7 @@ namespace MQTTnet.Formatter.V5
_packetWriter.FreeBuffer();
}
private static byte EncodePacket(MqttBasePacket packet, MqttPacketWriter packetWriter)
private static byte EncodePacket(MqttBasePacket packet, IMqttPacketWriter packetWriter)
{
switch (packet)
{
@@ -64,7 +76,7 @@ namespace MQTTnet.Formatter.V5
}
}
private static byte EncodeConnectPacket(MqttConnectPacket packet, MqttPacketWriter packetWriter)
private static byte EncodeConnectPacket(MqttConnectPacket packet, IMqttPacketWriter packetWriter)
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter));
@@ -150,7 +162,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Connect);
}
private static byte EncodeConnAckPacket(MqttConnAckPacket packet, MqttPacketWriter packetWriter)
private static byte EncodeConnAckPacket(MqttConnAckPacket packet, IMqttPacketWriter packetWriter)
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter));
@@ -195,7 +207,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.ConnAck);
}
private static byte EncodePublishPacket(MqttPublishPacket packet, MqttPacketWriter packetWriter)
private static byte EncodePublishPacket(MqttPublishPacket packet, IMqttPacketWriter packetWriter)
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter));
@@ -261,7 +273,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader);
}
private static byte EncodePubAckPacket(MqttPubAckPacket packet, MqttPacketWriter packetWriter)
private static byte EncodePubAckPacket(MqttPubAckPacket packet, IMqttPacketWriter packetWriter)
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter));
@@ -273,7 +285,7 @@ namespace MQTTnet.Formatter.V5
if (!packet.ReasonCode.HasValue)
{
throw new MqttProtocolViolationException("PubAck packet must contain a connect reason.");
throw new MqttProtocolViolationException("PubAck packet must contain a reason code.");
}
packetWriter.Write(packet.PacketIdentifier.Value);
@@ -294,7 +306,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubAck);
}
private static byte EncodePubRecPacket(MqttPubRecPacket packet, MqttPacketWriter packetWriter)
private static byte EncodePubRecPacket(MqttPubRecPacket packet, IMqttPacketWriter packetWriter)
{
ThrowIfPacketIdentifierIsInvalid(packet);
@@ -322,7 +334,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRec);
}
private static byte EncodePubRelPacket(MqttPubRelPacket packet, MqttPacketWriter packetWriter)
private static byte EncodePubRelPacket(MqttPubRelPacket packet, IMqttPacketWriter packetWriter)
{
ThrowIfPacketIdentifierIsInvalid(packet);
@@ -350,7 +362,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubRel, 0x02);
}
private static byte EncodePubCompPacket(MqttPubCompPacket packet, MqttPacketWriter packetWriter)
private static byte EncodePubCompPacket(MqttPubCompPacket packet, IMqttPacketWriter packetWriter)
{
ThrowIfPacketIdentifierIsInvalid(packet);
@@ -378,9 +390,9 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PubComp);
}
private static byte EncodeSubscribePacket(MqttSubscribePacket packet, MqttPacketWriter packetWriter)
private static byte EncodeSubscribePacket(MqttSubscribePacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.8.3-3].");
if (packet.TopicFilters?.Any() != true) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.8.3-3].");
ThrowIfPacketIdentifierIsInvalid(packet);
@@ -425,14 +437,11 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Subscribe, 0x02);
}
private static byte EncodeSubAckPacket(MqttSubAckPacket packet, MqttPacketWriter packetWriter)
private static byte EncodeSubAckPacket(MqttSubAckPacket packet, IMqttPacketWriter packetWriter)
{
ThrowIfPacketIdentifierIsInvalid(packet);
if (packet.ReasonCodes?.Any() != true) throw new MqttProtocolViolationException("At least one reason code must be set[MQTT - 3.8.3 - 3].");
if (packet.ReasonCodes == null || !packet.ReasonCodes.Any())
{
throw new MqttProtocolViolationException("At least one reason code must be set[MQTT - 3.8.3 - 3].");
}
ThrowIfPacketIdentifierIsInvalid(packet);
packetWriter.Write(packet.PacketIdentifier.Value);
@@ -453,9 +462,9 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.SubAck);
}
private static byte EncodeUnsubscribePacket(MqttUnsubscribePacket packet, MqttPacketWriter packetWriter)
private static byte EncodeUnsubscribePacket(MqttUnsubscribePacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.TopicFilters.Any()) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.10.3-2].");
if (packet.TopicFilters?.Any() != true) throw new MqttProtocolViolationException("At least one topic filter must be set [MQTT-3.10.3-2].");
ThrowIfPacketIdentifierIsInvalid(packet);
@@ -477,15 +486,12 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Unsubscibe, 0x02);
}
private static byte EncodeUnsubAckPacket(MqttUnsubAckPacket packet, MqttPacketWriter packetWriter)
private static byte EncodeUnsubAckPacket(MqttUnsubAckPacket packet, IMqttPacketWriter packetWriter)
{
if (packet.ReasonCodes?.Any() != true) throw new MqttProtocolViolationException("At least one reason code must be set[MQTT - 3.8.3 - 3].");
ThrowIfPacketIdentifierIsInvalid(packet);
if (packet.ReasonCodes == null || !packet.ReasonCodes.Any())
{
throw new MqttProtocolViolationException("At least one reason code must be set[MQTT - 3.8.3 - 3].");
}
packetWriter.Write(packet.PacketIdentifier.Value);
var propertiesWriter = new MqttV500PropertiesWriter();
@@ -505,7 +511,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.UnsubAck);
}
private static byte EncodeDisconnectPacket(MqttDisconnectPacket packet, MqttPacketWriter packetWriter)
private static byte EncodeDisconnectPacket(MqttDisconnectPacket packet, IMqttPacketWriter packetWriter)
{
if (!packet.ReasonCode.HasValue)
{
@@ -538,7 +544,7 @@ namespace MQTTnet.Formatter.V5
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.PingResp);
}
private static byte EncodeAuthPacket(MqttAuthPacket packet, MqttPacketWriter packetWriter)
private static byte EncodeAuthPacket(MqttAuthPacket packet, IMqttPacketWriter packetWriter)
{
packetWriter.Write((byte)packet.ReasonCode);
@@ -6,9 +6,19 @@ namespace MQTTnet.Formatter.V5
{
public class MqttV500PacketFormatter : IMqttPacketFormatter
{
private readonly MqttV500PacketEncoder _encoder = new MqttV500PacketEncoder();
private readonly MqttV500PacketEncoder _encoder;
private readonly MqttV500PacketDecoder _decoder = new MqttV500PacketDecoder();
public MqttV500PacketFormatter()
{
_encoder = new MqttV500PacketEncoder();
}
public MqttV500PacketFormatter(IMqttPacketWriter writer)
{
_encoder = new MqttV500PacketEncoder(writer);
}
public IMqttDataConverter DataConverter { get; } = new MqttV500DataConverter();
public ArraySegment<byte> Encode(MqttBasePacket mqttPacket)
@@ -9,11 +9,11 @@ namespace MQTTnet.Formatter.V5
{
public class MqttV500PropertiesReader
{
private readonly MqttPacketBodyReader _body;
private readonly IMqttPacketBodyReader _body;
private readonly uint _length;
private readonly ulong _targetOffset;
public MqttV500PropertiesReader(MqttPacketBodyReader body)
public MqttV500PropertiesReader(IMqttPacketBodyReader body)
{
_body = body ?? throw new ArgumentNullException(nameof(body));
@@ -75,7 +75,7 @@ namespace MQTTnet.Formatter.V5
public byte[] ReadAuthenticationData()
{
return _body.ReadWithLengthPrefix().ToArray();
return _body.ReadWithLengthPrefix();
}
public bool? ReadRetainAvailable()
@@ -165,7 +165,7 @@ namespace MQTTnet.Formatter.V5
public byte[] ReadCorrelationData()
{
return _body.ReadWithLengthPrefix().ToArray();
return _body.ReadWithLengthPrefix();
}
public string ReadContentType()
@@ -63,7 +63,7 @@ namespace MQTTnet.Formatter.V5
Write(MqttPropertyId.AuthenticationMethod, value);
}
public void WriteToPacket(MqttPacketWriter packetWriter)
public void WriteToPacket(IMqttPacketWriter packetWriter)
{
if (packetWriter == null) throw new ArgumentNullException(nameof(packetWriter));
@@ -4,6 +4,7 @@ using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Runtime.InteropServices.WindowsRuntime;
using System.Security.Authentication;
using System.Threading;
using System.Threading.Tasks;
using Windows.Networking;
@@ -38,22 +39,13 @@ namespace MQTTnet.Implementations
CreateStreams();
IsSecureConnection = socket.Information.ProtectionLevel >= SocketProtectionLevel.Tls12;
Endpoint = _socket.Information.RemoteAddress + ":" + _socket.Information.RemotePort;
}
public static Func<MqttClientTcpOptions, IEnumerable<ChainValidationResult>> CustomIgnorableServerCertificateErrorsResolver { get; set; }
public string Endpoint
{
get
{
if (_socket?.Information != null)
{
return _socket.Information.RemoteAddress + ":" + _socket.Information.RemotePort;
}
return null;
}
}
public string Endpoint { get; private set; }
public bool IsSecureConnection { get; }
@@ -66,7 +58,7 @@ namespace MQTTnet.Implementations
_socket.Control.KeepAlive = true;
}
if (!_options.TlsOptions.UseTls)
if (_options.TlsOptions?.UseTls != true)
{
await _socket.ConnectAsync(new HostName(_options.Server), _options.GetPort().ToString());
}
@@ -79,9 +71,21 @@ namespace MQTTnet.Implementations
_socket.Control.IgnorableServerCertificateErrors.Add(ignorableChainValidationResult);
}
await _socket.ConnectAsync(new HostName(_options.Server), _options.GetPort().ToString(), SocketProtectionLevel.Tls12);
var socketProtectionLevel = SocketProtectionLevel.Tls12;
if (_options.TlsOptions.SslProtocol == SslProtocols.Tls11)
{
socketProtectionLevel = SocketProtectionLevel.Tls11;
}
else if (_options.TlsOptions.SslProtocol == SslProtocols.Tls)
{
socketProtectionLevel = SocketProtectionLevel.Tls10;
}
await _socket.ConnectAsync(new HostName(_options.Server), _options.GetPort().ToString(), socketProtectionLevel);
}
Endpoint = _socket.Information.RemoteAddress + ":" + _socket.Information.RemotePort;
CreateStreams();
}
@@ -28,11 +28,12 @@ namespace MQTTnet.Implementations
IsSecureConnection = clientOptions.ChannelOptions?.TlsOptions?.UseTls == true;
}
public MqttTcpChannel(Stream stream)
public MqttTcpChannel(Stream stream, string endpoint)
{
_stream = stream ?? throw new ArgumentNullException(nameof(stream));
IsSecureConnection = stream is SslStream;
Endpoint = endpoint;
}
public string Endpoint { get; private set; }
@@ -23,7 +23,7 @@ namespace MQTTnet.Implementations
_logger = logger.CreateChildLogger(nameof(MqttTcpServerAdapter));
}
public Action<MqttServerAdapterClientAcceptedEventArgs> ClientAcceptedHandler { get; set; }
public Func<IMqttChannelAdapter, Task> ClientHandler { get; set; }
public async Task StartAsync(IMqttServerOptions options)
{
@@ -39,7 +39,7 @@ namespace MQTTnet.Implementations
_listener.Control.NoDelay = options.DefaultEndpointOptions.NoDelay;
_listener.Control.KeepAlive = true;
_listener.Control.QualityOfService = SocketQualityOfService.LowLatency;
_listener.ConnectionReceived += AcceptDefaultEndpointConnectionsAsync;
_listener.ConnectionReceived += OnConnectionReceivedAsync;
await _listener.BindServiceNameAsync(options.DefaultEndpointOptions.Port.ToString(), SocketProtectionLevel.PlainSocket);
}
@@ -54,30 +54,51 @@ namespace MQTTnet.Implementations
{
if (_listener != null)
{
_listener.ConnectionReceived -= AcceptDefaultEndpointConnectionsAsync;
_listener.ConnectionReceived -= OnConnectionReceivedAsync;
}
_listener?.Dispose();
_listener = null;
return Task.FromResult(0);
}
public void Dispose()
{
StopAsync().GetAwaiter().GetResult();
_listener?.Dispose();
_listener = null;
}
private void AcceptDefaultEndpointConnectionsAsync(StreamSocketListener sender, StreamSocketListenerConnectionReceivedEventArgs args)
private async void OnConnectionReceivedAsync(StreamSocketListener sender, StreamSocketListenerConnectionReceivedEventArgs args)
{
try
{
var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(args.Socket, _options), new MqttPacketFormatterAdapter(), _logger);
ClientAcceptedHandler?.Invoke(new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
var clientHandler = ClientHandler;
if (clientHandler != null)
{
using (var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(args.Socket, _options), new MqttPacketFormatterAdapter(), _logger))
{
await clientHandler(clientAdapter).ConfigureAwait(false);
}
}
}
catch (Exception exception)
{
_logger.Error(exception, "Error while accepting connection at default endpoint.");
if (exception is ObjectDisposedException)
{
// It can happen that the listener socket is accessed after the cancellation token is already set and the listener socket is disposed.
return;
}
_logger.Error(exception, "Error while handling client connection.");
}
finally
{
try
{
args.Socket.Dispose();
}
catch (Exception exception)
{
_logger.Error(exception, "Error while cleaning up client connection");
}
}
}
}
@@ -26,7 +26,7 @@ namespace MQTTnet.Implementations
_logger = logger.CreateChildLogger(nameof(MqttTcpServerAdapter));
}
public Action<MqttServerAdapterClientAcceptedEventArgs> ClientAcceptedHandler { get; set; }
public Func<IMqttChannelAdapter, Task> ClientHandler { get; set; }
public Task StartAsync(IMqttServerOptions options)
{
@@ -89,7 +89,7 @@ namespace MQTTnet.Implementations
_cancellationTokenSource.Token,
_logger)
{
ClientAcceptedHandler = OnClientAccepted
ClientHandler = OnClientAcceptedAsync
};
listenerV4.Start();
@@ -105,7 +105,7 @@ namespace MQTTnet.Implementations
_cancellationTokenSource.Token,
_logger)
{
ClientAcceptedHandler = OnClientAccepted
ClientHandler = OnClientAcceptedAsync
};
listenerV6.Start();
@@ -113,9 +113,15 @@ namespace MQTTnet.Implementations
}
}
private void OnClientAccepted(MqttServerAdapterClientAcceptedEventArgs eventArgs)
private Task OnClientAcceptedAsync(IMqttChannelAdapter channelAdapter)
{
ClientAcceptedHandler?.Invoke(eventArgs);
var clientHandler = ClientHandler;
if (clientHandler == null)
{
return Task.FromResult(0);
}
return clientHandler(channelAdapter);
}
}
}
@@ -43,7 +43,7 @@ namespace MQTTnet.Implementations
}
}
public Action<MqttServerAdapterClientAcceptedEventArgs> ClientAcceptedHandler { get; set; }
public Func<IMqttChannelAdapter, Task> ClientHandler { get; set; }
public void Start()
{
@@ -82,7 +82,9 @@ namespace MQTTnet.Implementations
#else
var clientSocket = await _socket.AcceptAsync().ConfigureAwait(false);
#endif
await HandleClientConnection(clientSocket).ConfigureAwait(false);
#pragma warning disable 4014
Task.Run(() => TryHandleClientConnectionAsync(clientSocket), cancellationToken);
#pragma warning restore 4014
}
catch (Exception exception)
{
@@ -92,12 +94,20 @@ namespace MQTTnet.Implementations
}
}
private async Task HandleClientConnection(Socket clientSocket)
private async Task TryHandleClientConnectionAsync(Socket clientSocket)
{
Stream stream = null;
string remoteEndPoint = null;
try
{
remoteEndPoint = clientSocket.RemoteEndPoint.ToString();
_logger.Verbose("Client '{0}' accepted by TCP listener '{1}, {2}'.",
remoteEndPoint,
_socket.LocalEndPoint,
_addressFamily == AddressFamily.InterNetwork ? "ipv4" : "ipv6");
clientSocket.NoDelay = _options.NoDelay;
stream = new NetworkStream(clientSocket, true);
@@ -109,13 +119,14 @@ namespace MQTTnet.Implementations
stream = sslStream;
}
_logger.Verbose("Client '{0}' accepted by TCP listener '{1}, {2}'.",
clientSocket.RemoteEndPoint,
_socket.LocalEndPoint,
_addressFamily == AddressFamily.InterNetwork ? "ipv4" : "ipv6");
var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(stream), new MqttPacketFormatterAdapter(), _logger);
ClientAcceptedHandler?.Invoke(new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
var clientHandler = ClientHandler;
if (clientHandler != null)
{
using (var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(stream, remoteEndPoint), new MqttPacketFormatterAdapter(), _logger))
{
await clientHandler(clientAdapter).ConfigureAwait(false);
}
}
}
catch (Exception exception)
{
@@ -125,20 +136,29 @@ namespace MQTTnet.Implementations
return;
}
if (exception is SocketException socketException && socketException.SocketErrorCode == SocketError.OperationAborted)
if (exception is SocketException socketException &&
socketException.SocketErrorCode == SocketError.OperationAborted)
{
return;
}
_logger.Error(exception, "Error while handling client connection.");
}
finally
{
try
{
// Dispose already allocated resources.
stream?.Dispose();
clientSocket?.Dispose();
_logger.Verbose("Client '{0}' disconnected at TCP listener '{1}, {2}'.",
remoteEndPoint,
_socket.LocalEndPoint,
_addressFamily == AddressFamily.InterNetwork ? "ipv4" : "ipv6");
}
catch (Exception disposeException)
{
throw new AggregateException(exception, disposeException);
_logger.Error(disposeException, "Error while cleaning up client connection");
}
}
}
@@ -1,5 +1,4 @@
using System;
using System.Globalization;
using System.Net;
using System.Net.WebSockets;
using System.Security.Cryptography.X509Certificates;
+1 -1
View File
@@ -14,7 +14,7 @@ namespace MQTTnet
public bool Retain { get; set; }
public List<MqttUserProperty> UserProperties { get; set; } = new List<MqttUserProperty>();
public List<MqttUserProperty> UserProperties { get; set; }
public string ContentType { get; set; }
@@ -1,5 +1,4 @@
using System;
using System.Collections.Generic;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
@@ -11,8 +10,6 @@ namespace MQTTnet
{
public class MqttApplicationMessageBuilder
{
private readonly List<MqttUserProperty> _userProperties = new List<MqttUserProperty>();
private MqttQualityOfServiceLevel _qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
private string _topic;
private byte[] _payload;
@@ -24,6 +21,7 @@ namespace MQTTnet
private uint? _subscriptionIdentifier;
private uint? _messageExpiryInterval;
private MqttPayloadFormatIndicator? _payloadFormatIndicator;
private List<MqttUserProperty> _userProperties;
public MqttApplicationMessageBuilder WithTopic(string topic)
{
@@ -70,7 +68,7 @@ namespace MQTTnet
if (payload.Length == 0)
{
_payload = new byte[0];
_payload = null;
}
else
{
@@ -89,7 +87,7 @@ namespace MQTTnet
return this;
}
_payload = string.IsNullOrEmpty(payload) ? new byte[0] : Encoding.UTF8.GetBytes(payload);
_payload = string.IsNullOrEmpty(payload) ? null : Encoding.UTF8.GetBytes(payload);
return this;
}
@@ -128,6 +126,11 @@ namespace MQTTnet
/// </summary>
public MqttApplicationMessageBuilder WithUserProperty(string name, string value)
{
if (_userProperties == null)
{
_userProperties = new List<MqttUserProperty>();
}
_userProperties.Add(new MqttUserProperty(name, value));
return this;
}
@@ -205,7 +208,7 @@ namespace MQTTnet
var applicationMessage = new MqttApplicationMessage
{
Topic = _topic,
Payload = _payload ?? new byte[0],
Payload = _payload,
QualityOfServiceLevel = _qualityOfServiceLevel,
Retain = _retain,
ContentType = _contentType,
@@ -217,8 +220,11 @@ namespace MQTTnet
PayloadFormatIndicator = _payloadFormatIndicator
};
applicationMessage.UserProperties.AddRange(_userProperties);
if (_userProperties?.Any() == true)
{
applicationMessage.UserProperties = _userProperties;
}
return applicationMessage;
}
}
+1 -5
View File
@@ -2,10 +2,6 @@
{
public class MqttConnectPacket : MqttBasePacket
{
public string ProtocolName { get; set; }
public byte? ProtocolLevel { get; set; }
public string ClientId { get; set; }
public string Username { get; set; }
@@ -33,7 +29,7 @@
password = "****";
}
return string.Concat("Connect: [ProtocolLevel=", ProtocolLevel, "] [ClientId=", ClientId, "] [Username=", Username, "] [Password=", password, "] [KeepAlivePeriod=", KeepAlivePeriod, "] [CleanSession=", CleanSession, "]");
return string.Concat("Connect: [ClientId=", ClientId, "] [Username=", Username, "] [Password=", password, "] [KeepAlivePeriod=", KeepAlivePeriod, "] [CleanSession=", CleanSession, "]");
}
}
}
+2 -2
View File
@@ -19,8 +19,8 @@ namespace MQTTnet.Server
Task<IList<IMqttClientStatus>> GetClientStatusAsync();
Task<IList<IMqttSessionStatus>> GetSessionStatusAsync();
Task<IList<MqttApplicationMessage>> GetRetainedMessagesAsync();
Task ClearRetainedMessagesAsync();
Task<IList<MqttApplicationMessage>> GetRetainedApplicationMessagesAsync();
Task ClearRetainedApplicationMessagesAsync();
Task SubscribeAsync(string clientId, ICollection<TopicFilter> topicFilters);
Task UnsubscribeAsync(string clientId, ICollection<string> topicFilters);
+3 -1
View File
@@ -4,6 +4,8 @@ namespace MQTTnet.Server
{
public interface IMqttServerOptions
{
string ClientId { get; set; }
bool EnablePersistentSessions { get; }
int MaxPendingMessagesPerClient { get; }
@@ -19,6 +21,6 @@ namespace MQTTnet.Server
MqttServerTcpEndpointOptions DefaultEndpointOptions { get; }
MqttServerTlsTcpEndpointOptions TlsEndpointOptions { get; }
IMqttServerStorage Storage { get; }
IMqttServerStorage Storage { get; }
}
}
+4 -18
View File
@@ -27,12 +27,12 @@ namespace MQTTnet.Server
private readonly IMqttNetChildLogger _logger;
private readonly IMqttServerOptions _serverOptions;
private Task<MqttClientDisconnectType> _packageReceiverTask;
private readonly IMqttChannelAdapter _channelAdapter;
private readonly IMqttDataConverter _dataConverter;
private readonly string _endpoint;
private readonly MqttConnectPacket _connectPacket;
private Task<MqttClientDisconnectType> _packageReceiverTask;
private DateTime _lastPacketReceivedTimestamp;
private DateTime _lastNonKeepAlivePacketReceivedTimestamp;
@@ -78,7 +78,7 @@ namespace MQTTnet.Server
StopInternal();
var task = _packageReceiverTask;
if (task != null && !task.IsCompleted)
if (task != null)
{
await task.ConfigureAwait(false);
}
@@ -103,13 +103,6 @@ namespace MQTTnet.Server
status.BytesReceived = _channelAdapter.BytesReceived;
}
//public void ClearPendingApplicationMessages()
//{
// Session.ApplicationMessagesQueue.Clear();
// //_applicationMessagesQueue.Clear();
//}
public void Dispose()
{
_cancellationToken.Dispose();
@@ -330,15 +323,8 @@ namespace MQTTnet.Server
var applicationMessage = _dataConverter.CreateApplicationMessage(publishPacket);
_sessionsManager.DispatchApplicationMessage(applicationMessage, this);
// TODO: Create ACK packet.
var response = new MqttPubAckPacket
{
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = MqttPubAckReasonCode.Success
};
return SendAsync(response);
var pubAckPacket = _dataConverter.CreatePubAckPacket(publishPacket);
return SendAsync(pubAckPacket);
}
private Task HandleIncomingPublishPacketWithQoS2Async(MqttPublishPacket publishPacket)
@@ -57,9 +57,9 @@ namespace MQTTnet.Server
}
}
public Task HandleConnectionAsync(IMqttChannelAdapter clientAdapter)
public Task HandleClientAsync(IMqttChannelAdapter clientAdapter)
{
return HandleConnectionAsync(clientAdapter, _cancellationToken);
return HandleClientAsync(clientAdapter, _cancellationToken);
}
public Task<IList<IMqttClientStatus>> GetClientStatusAsync()
@@ -186,7 +186,10 @@ namespace MQTTnet.Server
{
if (interceptorContext.CloseConnection)
{
await queuedApplicationMessage.Sender.StopAsync().ConfigureAwait(false);
if (sender != null)
{
await sender.StopAsync().ConfigureAwait(false);
}
}
if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish)
@@ -207,7 +210,7 @@ namespace MQTTnet.Server
foreach (var clientSession in _sessions.Values)
{
clientSession.EnqueueApplicationMessage(
queuedApplicationMessage.ApplicationMessage,
applicationMessage,
sender?.ClientId,
false);
}
@@ -221,7 +224,7 @@ namespace MQTTnet.Server
}
}
private async Task HandleConnectionAsync(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
private async Task HandleClientAsync(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
{
var disconnectType = MqttClientDisconnectType.NotClean;
var clientId = string.Empty;
@@ -274,17 +277,9 @@ namespace MQTTnet.Server
{
_connections.TryRemove(clientId, out _);
////connection?.ReferenceCounter.Decrement();
////if (connection?.ReferenceCounter.HasReferences == true)
////{
//// disconnectType = MqttClientDisconnectType.Takeover;
////}
////else
if (!_options.EnablePersistentSessions)
{
if (!_options.EnablePersistentSessions)
{
await DeleteSessionAsync(clientId).ConfigureAwait(false);
}
await DeleteSessionAsync(clientId).ConfigureAwait(false);
}
await TryCleanupChannelAsync(channelAdapter).ConfigureAwait(false);
@@ -369,7 +364,13 @@ namespace MQTTnet.Server
return null;
}
var interceptorContext = new MqttApplicationMessageInterceptorContext(sender?.ClientId, applicationMessage);
var senderClientId = sender?.ClientId;
if (sender == null)
{
senderClientId = _options.ClientId;
}
var interceptorContext = new MqttApplicationMessageInterceptorContext(senderClientId, applicationMessage);
await interceptor.InterceptApplicationMessagePublishAsync(interceptorContext).ConfigureAwait(false);
return interceptorContext;
}
@@ -384,10 +385,6 @@ namespace MQTTnet.Server
{
_logger.Error(exception, "Error while disconnecting client channel.");
}
finally
{
channelAdapter.Dispose();
}
}
}
}
@@ -61,8 +61,9 @@ namespace MQTTnet.Server
using (await _messagesLock.WaitAsync().ConfigureAwait(false))
{
var saveIsRequired = false;
var hasPayload = applicationMessage.Payload != null && applicationMessage.Payload.Length > 0;
if (applicationMessage.Payload?.Length == 0)
if (!hasPayload)
{
saveIsRequired = _messages.Remove(applicationMessage.Topic);
_logger.Verbose("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);
+6 -6
View File
@@ -77,7 +77,7 @@ namespace MQTTnet.Server
return _clientSessionsManager.GetSessionStatusAsync();
}
public Task<IList<MqttApplicationMessage>> GetRetainedMessagesAsync()
public Task<IList<MqttApplicationMessage>> GetRetainedApplicationMessagesAsync()
{
return _retainedMessagesManager.GetMessagesAsync();
}
@@ -127,7 +127,7 @@ namespace MQTTnet.Server
foreach (var adapter in _adapters)
{
adapter.ClientAcceptedHandler = OnClientAccepted;
adapter.ClientHandler = OnHandleClient;
await adapter.StartAsync(Options).ConfigureAwait(false);
}
@@ -155,7 +155,7 @@ namespace MQTTnet.Server
foreach (var adapter in _adapters)
{
adapter.ClientAcceptedHandler = null;
adapter.ClientHandler = null;
await adapter.StopAsync().ConfigureAwait(false);
}
@@ -179,14 +179,14 @@ namespace MQTTnet.Server
}
}
public Task ClearRetainedMessagesAsync()
public Task ClearRetainedApplicationMessagesAsync()
{
return _retainedMessagesManager?.ClearMessagesAsync();
}
private void OnClientAccepted(MqttServerAdapterClientAcceptedEventArgs eventArgs)
private Task OnHandleClient(IMqttChannelAdapter channelAdapter)
{
eventArgs.SessionTask = _clientSessionsManager.HandleConnectionAsync(eventArgs.ChannelAdapter);
return _clientSessionsManager.HandleClientAsync(channelAdapter);
}
}
}
@@ -10,6 +10,70 @@ namespace MQTTnet.Server
{
public static class MqttServerExtensions
{
public static IMqttServer UseClientConnectedHandler(this IMqttServer server, Func<MqttServerClientConnectedEventArgs, Task> handler)
{
if (server == null) throw new ArgumentNullException(nameof(server));
if (handler == null)
{
return server.UseClientConnectedHandler((IMqttServerClientConnectedHandler)null);
}
return server.UseClientConnectedHandler(new MqttServerClientConnectedHandlerDelegate(handler));
}
public static IMqttServer UseClientConnectedHandler(this IMqttServer server, Action<MqttServerClientConnectedEventArgs> handler)
{
if (server == null) throw new ArgumentNullException(nameof(server));
if (handler == null)
{
return server.UseClientConnectedHandler((IMqttServerClientConnectedHandler)null);
}
return server.UseClientConnectedHandler(new MqttServerClientConnectedHandlerDelegate(handler));
}
public static IMqttServer UseClientConnectedHandler(this IMqttServer server, IMqttServerClientConnectedHandler handler)
{
if (server == null) throw new ArgumentNullException(nameof(server));
server.ClientConnectedHandler = handler;
return server;
}
public static IMqttServer UseClientDisconnectedHandler(this IMqttServer server, Func<MqttServerClientDisconnectedEventArgs, Task> handler)
{
if (server == null) throw new ArgumentNullException(nameof(server));
if (handler == null)
{
return server.UseClientDisconnectedHandler((IMqttServerClientDisconnectedHandler)null);
}
return server.UseClientDisconnectedHandler(new MqttServerClientDisconnectedHandlerDelegate(handler));
}
public static IMqttServer UseClientDisconnectedHandler(this IMqttServer server, Action<MqttServerClientDisconnectedEventArgs> handler)
{
if (server == null) throw new ArgumentNullException(nameof(server));
if (handler == null)
{
return server.UseClientDisconnectedHandler((IMqttServerClientDisconnectedHandler)null);
}
return server.UseClientDisconnectedHandler(new MqttServerClientDisconnectedHandlerDelegate(handler));
}
public static IMqttServer UseClientDisconnectedHandler(this IMqttServer server, IMqttServerClientDisconnectedHandler handler)
{
if (server == null) throw new ArgumentNullException(nameof(server));
server.ClientDisconnectedHandler = handler;
return server;
}
public static IMqttServer UseApplicationMessageReceivedHandler(this IMqttServer server, Func<MqttApplicationMessageReceivedEventArgs, Task> handler)
{
if (server == null) throw new ArgumentNullException(nameof(server));
@@ -8,6 +8,8 @@ namespace MQTTnet.Server
public MqttServerTlsTcpEndpointOptions TlsEndpointOptions { get; } = new MqttServerTlsTcpEndpointOptions();
public string ClientId { get; set; }
public bool EnablePersistentSessions { get; set; }
public int MaxPendingMessagesPerClient { get; set; } = 250;
@@ -147,6 +147,15 @@ namespace MQTTnet.Server
return this;
}
/// <summary>
/// Gets or sets the client ID which is used when publishing messages from the server directly.
/// </summary>
public MqttServerOptionsBuilder WithClientId(string value)
{
_options.ClientId = value;
return this;
}
public IMqttServerOptions Build()
{
return _options;
@@ -6,8 +6,6 @@ namespace MQTTnet.Server.Status
{
string ClientId { get; set; }
bool IsConnected { get; }
long PendingApplicationMessagesCount { get; set; }
Task ClearPendingApplicationMessagesAsync();
@@ -17,8 +17,6 @@ namespace MQTTnet.Server.Status
public string ClientId { get; set; }
public long PendingApplicationMessagesCount { get; set; }
public bool IsConnected { get; set; }
public DateTime CreatedTimestamp { get; set; }
@@ -14,7 +14,7 @@
<ItemGroup>
<ProjectReference Include="..\..\Source\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj" />
<ProjectReference Include="..\..\Source\MQTTnet\MQTTnet.csproj" />
<ProjectReference Include="..\MQTTnet.Core.Tests\MQTTnet.Tests.csproj" />
</ItemGroup>
</Project>
@@ -0,0 +1,22 @@
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Formatter;
using MQTTnet.Tests;
namespace MQTTnet.AspNetCore.Tests
{
[TestClass]
public class MqttPacketSerializerTestsWithSpanBasedReader : MqttPacketSerializer_Tests
{
protected override IMqttPacketBodyReader ReaderFactory(byte[] data)
{
var result = new SpanBasedMqttPacketBodyReader();
result.SetBuffer(data);
return result;
}
protected override IMqttPacketWriter WriterFactory()
{
return new SpanBasedMqttPacketWriter();
}
}
}
@@ -23,22 +23,25 @@ namespace MQTTnet.AspNetCore.Tests
var consumed = part.Start;
var observed = part.Start;
var result = false;
var read = 0;
var reader = new SpanBasedMqttPacketBodyReader();
part = sequence.Slice(sequence.Start, 0); // empty message should fail
result = serializer.TryDecode(part, out packet, out consumed, out observed);
result = serializer.TryDecode(reader, part, out packet, out consumed, out observed, out read);
Assert.IsFalse(result);
part = sequence.Slice(sequence.Start, 1); // partial fixed header should fail
result = serializer.TryDecode(part, out packet, out consumed, out observed);
result = serializer.TryDecode(reader, part, out packet, out consumed, out observed, out read);
Assert.IsFalse(result);
part = sequence.Slice(sequence.Start, 4); // partial body should fail
result = serializer.TryDecode(part, out packet, out consumed, out observed);
result = serializer.TryDecode(reader, part, out packet, out consumed, out observed, out read);
Assert.IsFalse(result);
part = sequence; // complete msg should work
result = serializer.TryDecode(part, out packet, out consumed, out observed);
result = serializer.TryDecode(reader, part, out packet, out consumed, out observed, out read);
Assert.IsTrue(result);
}
}
@@ -3,20 +3,17 @@ using BenchmarkDotNet.Configs;
using BenchmarkDotNet.Jobs;
using BenchmarkDotNet.Validators;
namespace MQTTnet.Benchmarks
namespace MQTTnet.Benchmarks.Configurations
{
/// <summary>
/// this options may be used to run benchmarks in debugmode and attach a performance profiler
/// https://benchmarkdotnet.org/Configs/Configs.htm
/// </summary>
public class AllowNonOptimized : ManualConfig
public class AllowNonOptimized : BaseConfig
{
public AllowNonOptimized()
{
Add(JitOptimizationsValidator.DontFailOnError); // ALLOW NON-OPTIMIZED DLLS
Add(DefaultConfig.Instance.GetLoggers().ToArray()); // manual config has no loggers by default
Add(DefaultConfig.Instance.GetExporters().ToArray()); // manual config has no exporters by default
Add(DefaultConfig.Instance.GetColumnProviders().ToArray()); // manual config has no columns by default
Add(Job.InProcess);
}
}
@@ -0,0 +1,15 @@
using BenchmarkDotNet.Configs;
using System.Linq;
namespace MQTTnet.Benchmarks.Configurations
{
public class BaseConfig : ManualConfig
{
public BaseConfig()
{
Add(DefaultConfig.Instance.GetLoggers().ToArray()); // manual config has no loggers by default
Add(DefaultConfig.Instance.GetExporters().ToArray()); // manual config has no exporters by default
Add(DefaultConfig.Instance.GetColumnProviders().ToArray()); // manual config has no columns by default
}
}
}
@@ -0,0 +1,17 @@
using BenchmarkDotNet.Configs;
using BenchmarkDotNet.Environments;
using BenchmarkDotNet.Jobs;
using BenchmarkDotNet.Toolchains.CsProj;
namespace MQTTnet.Benchmarks.Configurations
{
public class RuntimeCompareConfig : BaseConfig
{
public RuntimeCompareConfig()
{
Add(Job.Default.With(Runtime.Clr));
Add(Job.Default.With(Runtime.Core).With(CsProjCoreToolchain.NetCoreApp21));
}
}
}
@@ -3,7 +3,7 @@
<PropertyGroup>
<OutputType>Exe</OutputType>
<DebugType>Full</DebugType>
<TargetFramework Condition=" '$(OS)' == 'Windows_NT' ">net461</TargetFramework>
<TargetFrameworks Condition=" '$(OS)' == 'Windows_NT' ">net461;netcoreapp2.1</TargetFrameworks>
<TargetFramework Condition=" '$(OS)' != 'Windows_NT' ">netcoreapp2.1</TargetFramework>
<LangVersion>7.2</LangVersion>
</PropertyGroup>
@@ -21,8 +21,16 @@ namespace MQTTnet.Benchmarks
{
var factory = new MqttFactory();
var tcpServer = new MqttTcpServerAdapter(new MqttNetLogger().CreateChildLogger());
tcpServer.ClientAcceptedHandler += args => _serverChannel = (IMqttChannel)args.ChannelAdapter.GetType().GetField("_channel", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance).GetValue(args.ChannelAdapter);
tcpServer.ClientHandler += args =>
{
_serverChannel =
(IMqttChannel)args.GetType().GetField("_channel",
System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance)
.GetValue(args);
return Task.CompletedTask;
};
_mqttServer = factory.CreateMqttServer(new[] { tcpServer }, new MqttNetLogger());
var serverOptions = new MqttServerOptionsBuilder().Build();
+2 -1
View File
@@ -1,5 +1,6 @@
using System;
using BenchmarkDotNet.Running;
using MQTTnet.Benchmarks.Configurations;
using MQTTnet.Diagnostics;
namespace MQTTnet.Benchmarks
@@ -43,7 +44,7 @@ namespace MQTTnet.Benchmarks
BenchmarkRunner.Run<TcpPipesBenchmark>();
break;
case '8':
BenchmarkRunner.Run<MessageProcessingMqttConnectionContextBenchmark>(new AllowNonOptimized());
BenchmarkRunner.Run<MessageProcessingMqttConnectionContextBenchmark>(new RuntimeCompareConfig()/*new AllowNonOptimized()*/);
break;
}
@@ -84,12 +84,12 @@ namespace MQTTnet.Benchmarks
public Task ConnectAsync(CancellationToken cancellationToken)
{
throw new NotImplementedException();
throw new NotSupportedException();
}
public Task DisconnectAsync(CancellationToken cancellationToken)
{
throw new NotImplementedException();
throw new NotSupportedException();
}
public Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
@@ -102,7 +102,7 @@ namespace MQTTnet.Benchmarks
public Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
throw new NotImplementedException();
throw new NotSupportedException();
}
public void Dispose()
@@ -0,0 +1,55 @@
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
using MQTTnet.Client.Options;
using MQTTnet.Client.Subscribing;
using MQTTnet.Formatter;
using MQTTnet.Tests.Mockups;
namespace MQTTnet.Tests.MQTTv5
{
[TestClass]
public class Feature_Tests
{
[TestMethod]
public async Task Use_User_Properties()
{
using (var testEnvironment = new TestEnvironment())
{
await testEnvironment.StartServerAsync();
var sender = await testEnvironment.ConnectClientAsync(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500));
var receiver = await testEnvironment.ConnectClientAsync(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500));
var message = new MqttApplicationMessageBuilder()
.WithTopic("A")
.WithUserProperty("x", "1")
.WithUserProperty("y", "2")
.WithUserProperty("z", "3")
.WithUserProperty("z", "4"); // z is here two times to test list of items
await receiver.SubscribeAsync(new MqttClientSubscribeOptions
{
TopicFilters = new List<TopicFilter>
{
new TopicFilter { Topic = "#" }
}
}, CancellationToken.None);
MqttApplicationMessage receivedMessage = null;
receiver.UseApplicationMessageReceivedHandler(e => receivedMessage = e.ApplicationMessage);
await sender.PublishAsync(message.Build(), CancellationToken.None);
await Task.Delay(1000);
Assert.IsNotNull(receivedMessage);
Assert.AreEqual("A", receivedMessage.Topic);
Assert.AreEqual(4, receivedMessage.UserProperties.Count);
}
}
}
}
@@ -1,61 +0,0 @@
using System;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Adapter;
using MQTTnet.Client;
using MQTTnet.Client.Options;
using MQTTnet.Diagnostics;
using MQTTnet.Server;
namespace MQTTnet.Tests.Mockups
{
public class TestMqttServerAdapter : IMqttServerAdapter
{
public Action<MqttServerAdapterClientAcceptedEventArgs> ClientAcceptedHandler { get; set; }
public async Task<IMqttClient> ConnectTestClient(string clientId, MqttApplicationMessage willMessage = null)
{
var adapterA = new TestMqttCommunicationAdapter();
var adapterB = new TestMqttCommunicationAdapter();
adapterA.Partner = adapterB;
adapterB.Partner = adapterA;
var client = new MqttClient(
new TestMqttCommunicationAdapterFactory(adapterA),
new MqttNetLogger());
FireClientAcceptedEvent(adapterB);
var options = new MqttClientOptions
{
ClientId = clientId,
WillMessage = willMessage,
ChannelOptions = new MqttClientTcpOptions()
};
await client.ConnectAsync(options);
SpinWait.SpinUntil(() => client.IsConnected);
return client;
}
private void FireClientAcceptedEvent(IMqttChannelAdapter adapter)
{
ClientAcceptedHandler?.Invoke(new MqttServerAdapterClientAcceptedEventArgs(adapter));
}
public Task StartAsync(IMqttServerOptions options)
{
return Task.FromResult(0);
}
public Task StopAsync()
{
return Task.FromResult(0);
}
public void Dispose()
{
}
}
}
@@ -2,12 +2,14 @@ using System;
using System.Collections.Generic;
using System.Linq;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Client.Options;
using MQTTnet.Client.Subscribing;
using MQTTnet.Exceptions;
using MQTTnet.Protocol;
using MQTTnet.Server;
@@ -18,6 +20,34 @@ namespace MQTTnet.Tests
[TestClass]
public class Client_Tests
{
[TestMethod]
public async Task PacketIdentifier_In_Publish_Result()
{
using (var testEnvironment = new TestEnvironment())
{
await testEnvironment.StartServerAsync();
var client = await testEnvironment.ConnectClientAsync();
var result = await client.PublishAsync("a", "a", MqttQualityOfServiceLevel.AtMostOnce);
Assert.AreEqual(null, result.PacketIdentifier);
result = await client.PublishAsync("b", "b", MqttQualityOfServiceLevel.AtMostOnce);
Assert.AreEqual(null, result.PacketIdentifier);
result = await client.PublishAsync("a", "a", MqttQualityOfServiceLevel.AtLeastOnce);
Assert.AreEqual((ushort)1, result.PacketIdentifier);
result = await client.PublishAsync("b", "b", MqttQualityOfServiceLevel.AtLeastOnce);
Assert.AreEqual((ushort)2, result.PacketIdentifier);
result = await client.PublishAsync("a", "a", MqttQualityOfServiceLevel.ExactlyOnce);
Assert.AreEqual((ushort)3, result.PacketIdentifier);
result = await client.PublishAsync("b", "b", MqttQualityOfServiceLevel.ExactlyOnce);
Assert.AreEqual((ushort)4, result.PacketIdentifier);
}
}
[TestMethod]
public async Task Invalid_Connect_Throws_Exception()
{
@@ -339,5 +369,36 @@ namespace MQTTnet.Tests
Assert.AreEqual("1", receivedPayload);
}
}
[TestMethod]
public async Task No_Payload()
{
using (var testEnvironment = new TestEnvironment())
{
await testEnvironment.StartServerAsync();
var sender = await testEnvironment.ConnectClientAsync();
var receiver = await testEnvironment.ConnectClientAsync();
var message = new MqttApplicationMessageBuilder()
.WithTopic("A");
await receiver.SubscribeAsync(new MqttClientSubscribeOptions
{
TopicFilters = new List<TopicFilter> { new TopicFilter { Topic = "#" } }
}, CancellationToken.None);
MqttApplicationMessage receivedMessage = null;
receiver.UseApplicationMessageReceivedHandler(e => receivedMessage = e.ApplicationMessage);
await sender.PublishAsync(message.Build(), CancellationToken.None);
await Task.Delay(1000);
Assert.IsNotNull(receivedMessage);
Assert.AreEqual("A", receivedMessage.Topic);
Assert.AreEqual(null, receivedMessage.Payload);
}
}
}
}
@@ -231,6 +231,32 @@ namespace MQTTnet.Tests
SerializeAndCompare(p, "Ow4ABUEvQi9DAHtIRUxMTw==");
}
[TestMethod]
public void SerializeV500_MqttPublishPacket()
{
var prop = new MqttPublishPacketProperties();
prop.ResponseTopic = "/Response";
prop.UserProperties.Add(new MqttUserProperty("Foo", "Bar"));
var p = new MqttPublishPacket
{
PacketIdentifier = 123,
Dup = true,
Retain = true,
Payload = Encoding.ASCII.GetBytes("HELLO"),
QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce,
Topic = "A/B/C",
Properties = prop
};
var deserialized = Roundtrip(p, MqttProtocolVersion.V500);
Assert.AreEqual(prop.ResponseTopic, deserialized.Properties.ResponseTopic);
Assert.IsTrue(deserialized.Properties.UserProperties.Any(x => x.Name == "Foo"));
}
[TestMethod]
public void DeserializeV311_MqttPublishPacket()
{
@@ -528,21 +554,21 @@ namespace MQTTnet.Tests
Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(data)));
}
private static void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
protected virtual IMqttPacketWriter WriterFactory()
{
IMqttPacketFormatter serializer;
if (protocolVersion == MqttProtocolVersion.V311)
{
serializer = new MqttV311PacketFormatter();
}
else if (protocolVersion == MqttProtocolVersion.V310)
{
serializer = new MqttV310PacketFormatter();
}
else
{
throw new NotSupportedException();
}
return new MqttPacketWriter();
}
protected virtual IMqttPacketBodyReader ReaderFactory(byte[] data)
{
return new MqttPacketBodyReader(data, 0, data.Length);
}
private void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
{
var writer = WriterFactory();
var serializer = MqttPacketFormatterAdapter.GetMqttPacketFormatter(protocolVersion, writer);
var buffer1 = serializer.Encode(packet);
@@ -554,7 +580,8 @@ namespace MQTTnet.Tests
using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, header.RemainingLength))
{
var deserializedPacket = serializer.Decode(new ReceivedMqttPacket(header.Flags, new MqttPacketBodyReader(bodyStream.ToArray(), 0, (int)bodyStream.Length), 0));
var reader = ReaderFactory(bodyStream.ToArray());
var deserializedPacket = serializer.Decode(new ReceivedMqttPacket(header.Flags, reader, 0));
var buffer2 = serializer.Encode(deserializedPacket);
Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(buffer2)));
@@ -562,23 +589,13 @@ namespace MQTTnet.Tests
}
}
private static T Roundtrip<T>(T packet, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
private T Roundtrip<T>(T packet, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
where T : MqttBasePacket
{
IMqttPacketFormatter serializer;
if (protocolVersion == MqttProtocolVersion.V311)
{
serializer = new MqttV311PacketFormatter();
}
else if (protocolVersion == MqttProtocolVersion.V310)
{
serializer = new MqttV310PacketFormatter();
}
else
{
throw new NotSupportedException();
}
var writer = WriterFactory();
var serializer = MqttPacketFormatterAdapter.GetMqttPacketFormatter(protocolVersion, writer);
var buffer1 = serializer.Encode(packet);
using (var headerStream = new MemoryStream(Join(buffer1)))
@@ -590,7 +607,8 @@ namespace MQTTnet.Tests
using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, (int)header.RemainingLength))
{
return (T)serializer.Decode(new ReceivedMqttPacket(header.Flags, new MqttPacketBodyReader(bodyStream.ToArray(), 0, (int)bodyStream.Length), 0));
var reader = ReaderFactory(bodyStream.ToArray());
return (T)serializer.Decode(new ReceivedMqttPacket(header.Flags, reader, 0));
}
}
}
@@ -15,11 +15,11 @@ namespace MQTTnet.Tests
public async Task Dispose_Channel_While_Used()
{
var ct = new CancellationTokenSource();
var serverSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
try
{
var serverSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
serverSocket.Bind(new IPEndPoint(IPAddress.Any, 50000));
serverSocket.Bind(new IPEndPoint(IPAddress.Any, 50001));
serverSocket.Listen(0);
#pragma warning disable 4014
@@ -35,11 +35,11 @@ namespace MQTTnet.Tests
}, ct.Token);
var clientSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await clientSocket.ConnectAsync(IPAddress.Loopback, 50000);
await clientSocket.ConnectAsync(IPAddress.Loopback, 50001);
await Task.Delay(100, ct.Token);
var tcpChannel = new MqttTcpChannel(new NetworkStream(clientSocket, true));
var tcpChannel = new MqttTcpChannel(new NetworkStream(clientSocket, true), "test");
var buffer = new byte[1];
await tcpChannel.ReadAsync(buffer, 0, 1, ct.Token);
@@ -68,6 +68,7 @@ namespace MQTTnet.Tests
finally
{
ct.Cancel(false);
serverSocket.Dispose();
}
}
}
+26 -33
View File
@@ -4,9 +4,7 @@ using System.Diagnostics;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
using MQTTnet.Client.Options;
using MQTTnet.Client.Receiving;
using MQTTnet.Server;
using MQTTnet.Tests.Mockups;
namespace MQTTnet.Tests
{
@@ -16,42 +14,37 @@ namespace MQTTnet.Tests
[TestMethod]
public async Task Round_Trip_Time()
{
var factory = new MqttFactory();
var server = factory.CreateMqttServer();
var receiverClient = factory.CreateMqttClient();
var senderClient = factory.CreateMqttClient();
await server.StartAsync(new MqttServerOptionsBuilder().Build());
await receiverClient.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost").Build());
await receiverClient.SubscribeAsync("#");
await senderClient.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost").Build());
TaskCompletionSource<string> response = null;
receiverClient.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(args =>
using (var testEnvironment = new TestEnvironment())
{
response?.SetResult(args.ApplicationMessage.ConvertPayloadToString());
});
await testEnvironment.StartServerAsync();
var receiverClient = await testEnvironment.ConnectClientAsync();
var senderClient = await testEnvironment.ConnectClientAsync();
var times = new List<TimeSpan>();
var stopwatch = Stopwatch.StartNew();
await receiverClient.SubscribeAsync("#");
TaskCompletionSource<string> response = null;
await Task.Delay(1000);
receiverClient.UseApplicationMessageReceivedHandler(e =>
{
response?.SetResult(e.ApplicationMessage.ConvertPayloadToString());
});
for (var i = 0; i < 100; i++)
{
response = new TaskCompletionSource<string>();
await senderClient.PublishAsync("test", DateTime.UtcNow.Ticks.ToString());
response.Task.GetAwaiter().GetResult();
var times = new List<TimeSpan>();
var stopwatch = Stopwatch.StartNew();
stopwatch.Stop();
times.Add(stopwatch.Elapsed);
stopwatch.Restart();
await Task.Delay(1000);
for (var i = 0; i < 100; i++)
{
response = new TaskCompletionSource<string>();
await senderClient.PublishAsync("test", DateTime.UtcNow.Ticks.ToString());
response.Task.GetAwaiter().GetResult();
stopwatch.Stop();
times.Add(stopwatch.Elapsed);
stopwatch.Restart();
}
}
await server.StopAsync();
}
}
}
+94 -3
View File
@@ -54,6 +54,20 @@ namespace MQTTnet.Tests
1);
}
[TestMethod]
public async Task Use_Clean_Session()
{
using (var testEnvironment = new TestEnvironment())
{
await testEnvironment.StartServerAsync();
var client = testEnvironment.CreateClient();
var connectResult = await client.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("localhost", testEnvironment.ServerPort).WithCleanSession().Build());
Assert.IsFalse(connectResult.IsSessionPresent);
}
}
[TestMethod]
public async Task Will_Message_Do_Not_Send()
{
@@ -61,7 +75,7 @@ namespace MQTTnet.Tests
{
var receivedMessagesCount = 0;
await testEnvironment.StartServerAsync(new MqttServerOptionsBuilder());
await testEnvironment.StartServerAsync();
var willMessage = new MqttApplicationMessageBuilder().WithTopic("My/last/will").WithAtMostOnceQoS().Build();
@@ -437,7 +451,7 @@ namespace MQTTnet.Tests
await Task.Delay(1000);
var retainedMessages = await server.GetRetainedMessagesAsync();
var retainedMessages = await server.GetRetainedApplicationMessagesAsync();
Assert.AreEqual(ClientCount, retainedMessages.Count);
@@ -540,7 +554,7 @@ namespace MQTTnet.Tests
}
[TestMethod]
public async Task Clear_Retained_Message()
public async Task Clear_Retained_Message_With_Empty_Payload()
{
using (var testEnvironment = new TestEnvironment())
{
@@ -567,6 +581,55 @@ namespace MQTTnet.Tests
}
}
[TestMethod]
public async Task Clear_Retained_Message_With_Null_Payload()
{
using (var testEnvironment = new TestEnvironment())
{
var receivedMessagesCount = 0;
await testEnvironment.StartServerAsync();
var c1 = await testEnvironment.ConnectClientAsync();
await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build());
await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload((byte[])null).WithRetainFlag().Build());
await c1.DisconnectAsync();
var c2 = await testEnvironment.ConnectClientAsync();
c2.UseApplicationMessageReceivedHandler(c => Interlocked.Increment(ref receivedMessagesCount));
await Task.Delay(200);
await c2.SubscribeAsync(new TopicFilter { Topic = "retained", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
await Task.Delay(500);
Assert.AreEqual(0, receivedMessagesCount);
}
}
[TestMethod]
public async Task Intercept_Application_Message()
{
using (var testEnvironment = new TestEnvironment())
{
await testEnvironment.StartServerAsync(
new MqttServerOptionsBuilder().WithApplicationMessageInterceptor(
c => { c.ApplicationMessage = new MqttApplicationMessage {Topic = "new_topic" }; }));
string receivedTopic = null;
var c1 = await testEnvironment.ConnectClientAsync();
await c1.SubscribeAsync("#");
c1.UseApplicationMessageReceivedHandler(a => { receivedTopic = a.ApplicationMessage.Topic; });
await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("original_topic").Build());
await Task.Delay(500);
Assert.AreEqual("new_topic", receivedTopic);
}
}
[TestMethod]
public async Task Persist_Retained_Message()
{
@@ -586,6 +649,34 @@ namespace MQTTnet.Tests
}
}
[TestMethod]
public async Task Publish_After_Client_Connects()
{
using (var testEnvironment = new TestEnvironment())
{
var server = await testEnvironment.StartServerAsync();
server.UseClientConnectedHandler(async e =>
{
await server.PublishAsync("/test/1", "true", MqttQualityOfServiceLevel.ExactlyOnce, false);
});
string receivedTopic = null;
var c1 = await testEnvironment.ConnectClientAsync();
c1.UseApplicationMessageReceivedHandler(e => { receivedTopic = e.ApplicationMessage.Topic; });
await c1.SubscribeAsync("#");
await testEnvironment.ConnectClientAsync();
await testEnvironment.ConnectClientAsync();
await testEnvironment.ConnectClientAsync();
await testEnvironment.ConnectClientAsync();
await Task.Delay(500);
Assert.AreEqual("/test/1", receivedTopic);
}
}
[TestMethod]
public async Task Intercept_Message()
{
+21 -1
View File
@@ -1,6 +1,8 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Net.Security;
using System.Security.Cryptography.X509Certificates;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Client;
@@ -81,7 +83,7 @@ namespace MQTTnet.TestApp.NetCore
{
Task.Run(PerformanceTest.RunQoS0Test);
}
Thread.Sleep(Timeout.Infinite);
}
}
@@ -118,4 +120,22 @@ namespace MQTTnet.TestApp.NetCore
return Task.FromResult(retainedMessages);
}
}
public class WikiCode
{
public void Code()
{
//Validate certificate.
var options = new MqttClientOptionsBuilder()
.WithTls(new MqttClientOptionsBuilderTlsParameters
{
CertificateValidationCallback = (X509Certificate x, X509Chain y, SslPolicyErrors z, IMqttClientOptions o) =>
{
// TODO: Check conditions of certificate by using above parameters.
return true;
}
})
.Build();
}
}
}
@@ -19,37 +19,37 @@ namespace MQTTnet.TestApp.NetCore
// iot.eclipse.org
await ExecuteTestAsync("iot.eclipse.org TCP",
new MqttClientOptionsBuilder().WithTcpServer("iot.eclipse.org", 1883).WithProtocolVersion(MqttProtocolVersion.V500).Build());
new MqttClientOptionsBuilder().WithTcpServer("iot.eclipse.org", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
await ExecuteTestAsync("iot.eclipse.org WS",
new MqttClientOptionsBuilder().WithWebSocketServer("iot.eclipse.org:80/mqtt").WithProtocolVersion(MqttProtocolVersion.V500).Build());
new MqttClientOptionsBuilder().WithWebSocketServer("iot.eclipse.org:80/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).Build());
await ExecuteTestAsync("iot.eclipse.org WS TLS",
new MqttClientOptionsBuilder().WithWebSocketServer("iot.eclipse.org:443/mqtt").WithProtocolVersion(MqttProtocolVersion.V500).WithTls().Build());
new MqttClientOptionsBuilder().WithWebSocketServer("iot.eclipse.org:443/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).WithTls().Build());
// test.mosquitto.org
await ExecuteTestAsync("test.mosquitto.org TCP",
new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 1883).WithProtocolVersion(MqttProtocolVersion.V500).Build());
new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
await ExecuteTestAsync("test.mosquitto.org TCP TLS",
new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 8883).WithProtocolVersion(MqttProtocolVersion.V500).WithTls().Build());
new MqttClientOptionsBuilder().WithTcpServer("test.mosquitto.org", 8883).WithProtocolVersion(MqttProtocolVersion.V311).WithTls().Build());
await ExecuteTestAsync("test.mosquitto.org WS",
new MqttClientOptionsBuilder().WithWebSocketServer("test.mosquitto.org:8080/mqtt").WithProtocolVersion(MqttProtocolVersion.V500).Build());
new MqttClientOptionsBuilder().WithWebSocketServer("test.mosquitto.org:8080/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).Build());
await ExecuteTestAsync("test.mosquitto.org WS TLS",
new MqttClientOptionsBuilder().WithWebSocketServer("test.mosquitto.org:8081/mqtt").WithProtocolVersion(MqttProtocolVersion.V500).WithTls().Build());
new MqttClientOptionsBuilder().WithWebSocketServer("test.mosquitto.org:8081/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).WithTls().Build());
// broker.hivemq.com
await ExecuteTestAsync("broker.hivemq.com TCP",
new MqttClientOptionsBuilder().WithTcpServer("broker.hivemq.com", 1883).WithProtocolVersion(MqttProtocolVersion.V500).Build());
new MqttClientOptionsBuilder().WithTcpServer("broker.hivemq.com", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
await ExecuteTestAsync("broker.hivemq.com WS",
new MqttClientOptionsBuilder().WithWebSocketServer("broker.hivemq.com:8000/mqtt").WithProtocolVersion(MqttProtocolVersion.V500).Build());
new MqttClientOptionsBuilder().WithWebSocketServer("broker.hivemq.com:8000/mqtt").WithProtocolVersion(MqttProtocolVersion.V311).Build());
// mqtt.swifitch.cz
await ExecuteTestAsync("mqtt.swifitch.cz",
new MqttClientOptionsBuilder().WithTcpServer("mqtt.swifitch.cz", 1883).WithProtocolVersion(MqttProtocolVersion.V500).Build());
new MqttClientOptionsBuilder().WithTcpServer("mqtt.swifitch.cz", 1883).WithProtocolVersion(MqttProtocolVersion.V311).Build());
// CloudMQTT
var configFile = Path.Combine("E:\\CloudMqttTestConfig.json");
@@ -58,13 +58,13 @@ namespace MQTTnet.TestApp.NetCore
var config = JsonConvert.DeserializeObject<MqttConfig>(File.ReadAllText(configFile));
await ExecuteTestAsync("CloudMQTT TCP",
new MqttClientOptionsBuilder().WithTcpServer(config.Server, config.Port).WithCredentials(config.Username, config.Password).WithProtocolVersion(MqttProtocolVersion.V500).Build());
new MqttClientOptionsBuilder().WithTcpServer(config.Server, config.Port).WithCredentials(config.Username, config.Password).WithProtocolVersion(MqttProtocolVersion.V311).Build());
await ExecuteTestAsync("CloudMQTT TCP TLS",
new MqttClientOptionsBuilder().WithTcpServer(config.Server, config.SslPort).WithCredentials(config.Username, config.Password).WithTls().WithProtocolVersion(MqttProtocolVersion.V500).Build());
new MqttClientOptionsBuilder().WithTcpServer(config.Server, config.SslPort).WithCredentials(config.Username, config.Password).WithTls().WithProtocolVersion(MqttProtocolVersion.V311).Build());
await ExecuteTestAsync("CloudMQTT WS TLS",
new MqttClientOptionsBuilder().WithWebSocketServer(config.Server + ":" + config.SslWebSocketPort + "/mqtt").WithCredentials(config.Username, config.Password).WithTls().WithProtocolVersion(MqttProtocolVersion.V500).Build());
new MqttClientOptionsBuilder().WithWebSocketServer(config.Server + ":" + config.SslWebSocketPort + "/mqtt").WithCredentials(config.Username, config.Password).WithTls().WithProtocolVersion(MqttProtocolVersion.V311).Build());
}
Write("Finished.", ConsoleColor.White);
@@ -94,7 +94,7 @@ namespace MQTTnet.TestApp.NetCore
throw new Exception("Message invalid.");
}
await client.UnsubscribeAsync("test");
await client.UnsubscribeAsync(topic);
await client.DisconnectAsync();
Write("[OK]\n", ConsoleColor.Green);
@@ -15,28 +15,14 @@ namespace MQTTnet.TestApp.UniversalWindows
Suspending += OnSuspending;
}
/// <summary>
/// Invoked when the application is launched normally by the end user. Other entry points
/// will be used such as when the application is launched to open a specific file.
/// </summary>
/// <param name="e">Details about the launch request and process.</param>
protected override void OnLaunched(LaunchActivatedEventArgs e)
{
// Do not repeat app initialization when the Window already has content,
// just ensure that the window is active
if (!(Window.Current.Content is Frame rootFrame))
{
// Create a Frame to act as the navigation context and navigate to the first page
rootFrame = new Frame();
rootFrame.NavigationFailed += OnNavigationFailed;
if (e.PreviousExecutionState == ApplicationExecutionState.Terminated)
{
//TODO: Load state from previously suspended application
}
// Place the frame in the current Window
Window.Current.Content = rootFrame;
}
@@ -44,37 +30,21 @@ namespace MQTTnet.TestApp.UniversalWindows
{
if (rootFrame.Content == null)
{
// When the navigation stack isn't restored navigate to the first page,
// configuring the new page by passing required information as a navigation
// parameter
rootFrame.Navigate(typeof(MainPage), e.Arguments);
}
// Ensure the current window is active
Window.Current.Activate();
}
}
/// <summary>
/// Invoked when Navigation to a certain page fails
/// </summary>
/// <param name="sender">The Frame which failed navigation</param>
/// <param name="e">Details about the navigation failure</param>
void OnNavigationFailed(object sender, NavigationFailedEventArgs e)
{
throw new Exception("Failed to load Page " + e.SourcePageType.FullName);
}
/// <summary>
/// Invoked when application execution is being suspended. Application state is saved
/// without knowing whether the application will be terminated or resumed with the contents
/// of memory still intact.
/// </summary>
/// <param name="sender">The source of the suspend request.</param>
/// <param name="e">Details about the suspend request.</param>
private void OnSuspending(object sender, SuspendingEventArgs e)
{
var deferral = e.SuspendingOperation.GetDeferral();
//TODO: Save application state and stop any background activity
deferral.Complete();
}
}
@@ -1,6 +1,8 @@
using System;
using System.Collections.Concurrent;
using System.Collections.ObjectModel;
using System.Net.Security;
using System.Security.Cryptography.X509Certificates;
using System.Text;
using System.Threading.Tasks;
using Windows.Security.Cryptography.Certificates;
@@ -119,12 +121,15 @@ namespace MQTTnet.TestApp.UniversalWindows
throw new InvalidOperationException();
}
options.Credentials = new MqttClientCredentials
if (!string.IsNullOrEmpty(User.Text))
{
Username = User.Text,
Password = Password.Text
};
options.Credentials = new MqttClientCredentials
{
Username = User.Text,
Password = Password.Text
};
}
options.CleanSession = CleanSession.IsChecked == true;
options.KeepAlivePeriod = TimeSpan.FromSeconds(double.Parse(KeepAliveInterval.Text));
@@ -592,7 +597,7 @@ namespace MQTTnet.TestApp.UniversalWindows
};
}
}
// ----------------------------------
{
var options = new MqttServerOptions();
@@ -711,6 +716,25 @@ namespace MQTTnet.TestApp.UniversalWindows
var mqttServer = new MqttFactory().CreateMqttServer();
}
{
// Setup subscription interceptor.
var options = new MqttServerOptionsBuilder()
.WithSubscriptionInterceptor(context =>
{
if (context.TopicFilter.Topic.StartsWith("admin/foo/bar") && context.ClientId != "theAdmin")
{
context.AcceptSubscription = false;
}
if (context.TopicFilter.Topic.StartsWith("the/secret/stuff") && context.ClientId != "Imperator")
{
context.AcceptSubscription = false;
context.CloseConnection = true;
}
})
.Build();
}
{
// Setup and start a managed MQTT client.
var options = new ManagedMqttClientOptionsBuilder()