Browse Source

Add low level client.

release/3.x.x
Christian Kratky 4 years ago
parent
commit
04ccdcd562
46 changed files with 719 additions and 426 deletions
  1. +1
    -0
      Build/MQTTnet.nuspec
  2. +4
    -5
      Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs
  3. +7
    -7
      Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs
  4. +9
    -9
      Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs
  5. +12
    -12
      Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs
  6. +43
    -43
      Source/MQTTnet.Server/Logging/MqttNetChildLoggerWrapper.cs
  7. +12
    -7
      Source/MQTTnet.Server/Logging/MqttNetLoggerWrapper.cs
  8. +1
    -1
      Source/MQTTnet/Adapter/IMqttClientAdapterFactory.cs
  9. +33
    -33
      Source/MQTTnet/Adapter/MqttChannelAdapter.cs
  10. +5
    -5
      Source/MQTTnet/Client/MqttClient.cs
  11. +8
    -8
      Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs
  12. +0
    -17
      Source/MQTTnet/Diagnostics/IMqttNetChildLogger.cs
  13. +2
    -2
      Source/MQTTnet/Diagnostics/IMqttNetLogger.cs
  14. +0
    -51
      Source/MQTTnet/Diagnostics/MqttNetChildLogger.cs
  15. +4
    -0
      Source/MQTTnet/Diagnostics/MqttNetLogMessagePublishedEventArgs.cs
  16. +12
    -6
      Source/MQTTnet/Diagnostics/MqttNetLogger.cs
  17. +35
    -0
      Source/MQTTnet/Diagnostics/MqttNetLoggerExtensions.cs
  18. +1
    -1
      Source/MQTTnet/Diagnostics/TargetFrameworkProvider.cs
  19. +4
    -4
      Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs
  20. +6
    -6
      Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs
  21. +22
    -21
      Source/MQTTnet/Implementations/MqttTcpServerAdapter.cs
  22. +22
    -22
      Source/MQTTnet/Implementations/MqttTcpServerListener.cs
  23. +3
    -3
      Source/MQTTnet/Internal/TaskExtensions.cs
  24. +19
    -0
      Source/MQTTnet/LowLevelClient/ILowLevelMqttClient.cs
  25. +128
    -0
      Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs
  26. +35
    -7
      Source/MQTTnet/MqttFactory.cs
  27. +3
    -3
      Source/MQTTnet/Server/IMqttRetainedMessagesManager.cs
  28. +7
    -7
      Source/MQTTnet/Server/MqttClientConnection.cs
  29. +7
    -7
      Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs
  30. +5
    -5
      Source/MQTTnet/Server/MqttClientSession.cs
  31. +8
    -8
      Source/MQTTnet/Server/MqttClientSessionsManager.cs
  32. +6
    -6
      Source/MQTTnet/Server/MqttRetainedMessagesManager.cs
  33. +2
    -2
      Source/MQTTnet/Server/MqttServer.cs
  34. +5
    -5
      Source/MQTTnet/Server/MqttServerEventDispatcher.cs
  35. +9
    -8
      Source/MQTTnet/TopicFilter.cs
  36. +1
    -1
      Tests/MQTTnet.Benchmarks/LoggerBenchmark.cs
  37. +1
    -1
      Tests/MQTTnet.Benchmarks/Program.cs
  38. +111
    -0
      Tests/MQTTnet.Core.Tests/LowLevelMqttClient_Tests.cs
  39. +27
    -22
      Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs
  40. +30
    -21
      Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs
  41. +10
    -10
      Tests/MQTTnet.Core.Tests/Mockups/TestLogger.cs
  42. +2
    -2
      Tests/MQTTnet.Core.Tests/Mockups/TestMqttCommunicationAdapterFactory.cs
  43. +28
    -21
      Tests/MQTTnet.Core.Tests/Mockups/TestServerWrapper.cs
  44. +10
    -8
      Tests/MQTTnet.Core.Tests/MqttFactory_Tests.cs
  45. +18
    -18
      Tests/MQTTnet.Core.Tests/Server_Tests.cs
  46. +1
    -1
      Tests/MQTTnet.TestApp.NetCore/Program.cs

+ 1
- 0
Build/MQTTnet.nuspec View File

@@ -21,6 +21,7 @@
* [Server] Added interceptor for unsubscriptions. * [Server] Added interceptor for unsubscriptions.
* [MQTTnet.Server] Added interceptor for unsubscriptions. * [MQTTnet.Server] Added interceptor for unsubscriptions.
* [MQTTnet.AspNetCore] improved compatibility with AspNetCore 3.1 * [MQTTnet.AspNetCore] improved compatibility with AspNetCore 3.1
* [LowLevelMqttClient] Added low level MQTT client in order to provide more flexibility when using the MQTT protocol. This client requires detailed knowledge about the MQTT protocol.
</releaseNotes> </releaseNotes>
<copyright>Copyright Christian Kratky 2016-2019</copyright> <copyright>Copyright Christian Kratky 2016-2019</copyright>
<tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin</tags> <tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin</tags>


+ 4
- 5
Source/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs View File

@@ -1,17 +1,16 @@
using System;
using System.Net;
using MQTTnet.Adapter;
using MQTTnet.Adapter;
using MQTTnet.AspNetCore.Client.Tcp; using MQTTnet.AspNetCore.Client.Tcp;
using MQTTnet.Client;
using MQTTnet.Client.Options; using MQTTnet.Client.Options;
using MQTTnet.Diagnostics; using MQTTnet.Diagnostics;
using MQTTnet.Formatter; using MQTTnet.Formatter;
using System;
using System.Net;


namespace MQTTnet.AspNetCore.Client namespace MQTTnet.AspNetCore.Client
{ {
public class MqttClientConnectionContextFactory : IMqttClientAdapterFactory public class MqttClientConnectionContextFactory : IMqttClientAdapterFactory
{ {
public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger)
public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger)
{ {
if (options == null) throw new ArgumentNullException(nameof(options)); if (options == null) throw new ArgumentNullException(nameof(options));




+ 7
- 7
Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs View File

@@ -1,20 +1,20 @@
using System;
using System.Net.WebSockets;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Http;
using MQTTnet.Adapter; using MQTTnet.Adapter;
using MQTTnet.Diagnostics; using MQTTnet.Diagnostics;
using MQTTnet.Formatter; using MQTTnet.Formatter;
using MQTTnet.Implementations; using MQTTnet.Implementations;
using MQTTnet.Server; using MQTTnet.Server;
using System;
using System.Net.WebSockets;
using System.Threading.Tasks;


namespace MQTTnet.AspNetCore namespace MQTTnet.AspNetCore
{ {
public class MqttWebSocketServerAdapter : IMqttServerAdapter public class MqttWebSocketServerAdapter : IMqttServerAdapter
{ {
private readonly IMqttNetChildLogger _logger;
private readonly IMqttNetLogger _logger;


public MqttWebSocketServerAdapter(IMqttNetChildLogger logger)
public MqttWebSocketServerAdapter(IMqttNetLogger logger)
{ {
if (logger == null) throw new ArgumentNullException(nameof(logger)); if (logger == null) throw new ArgumentNullException(nameof(logger));


@@ -38,7 +38,7 @@ namespace MQTTnet.AspNetCore
if (webSocket == null) throw new ArgumentNullException(nameof(webSocket)); if (webSocket == null) throw new ArgumentNullException(nameof(webSocket));


var endpoint = $"{httpContext.Connection.RemoteIpAddress}:{httpContext.Connection.RemotePort}"; var endpoint = $"{httpContext.Connection.RemoteIpAddress}:{httpContext.Connection.RemotePort}";
var clientCertificate = await httpContext.Connection.GetClientCertificateAsync().ConfigureAwait(false); var clientCertificate = await httpContext.Connection.GetClientCertificateAsync().ConfigureAwait(false);
try try
{ {


+ 9
- 9
Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs View File

@@ -1,9 +1,4 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Client;
using MQTTnet.Client;
using MQTTnet.Client.Connecting; using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting; using MQTTnet.Client.Disconnecting;
using MQTTnet.Client.Publishing; using MQTTnet.Client.Publishing;
@@ -13,6 +8,11 @@ using MQTTnet.Exceptions;
using MQTTnet.Internal; using MQTTnet.Internal;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using MQTTnet.Server; using MQTTnet.Server;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;


namespace MQTTnet.Extensions.ManagedClient namespace MQTTnet.Extensions.ManagedClient
{ {
@@ -33,7 +33,7 @@ namespace MQTTnet.Extensions.ManagedClient
private readonly SemaphoreSlim _subscriptionsQueuedSignal = new SemaphoreSlim(0); private readonly SemaphoreSlim _subscriptionsQueuedSignal = new SemaphoreSlim(0);


private readonly IMqttClient _mqttClient; private readonly IMqttClient _mqttClient;
private readonly IMqttNetChildLogger _logger;
private readonly IMqttNetLogger _logger;


private readonly AsyncLock _messageQueueLock = new AsyncLock(); private readonly AsyncLock _messageQueueLock = new AsyncLock();


@@ -42,8 +42,8 @@ namespace MQTTnet.Extensions.ManagedClient
private Task _maintainConnectionTask; private Task _maintainConnectionTask;


private ManagedMqttClientStorageManager _storageManager; private ManagedMqttClientStorageManager _storageManager;
public ManagedMqttClient(IMqttClient mqttClient, IMqttNetChildLogger logger)
public ManagedMqttClient(IMqttClient mqttClient, IMqttNetLogger logger)
{ {
_mqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient)); _mqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient));




+ 12
- 12
Source/MQTTnet.Extensions.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs View File

@@ -1,15 +1,15 @@
using System;
using MQTTnet.Adapter;
using MQTTnet.Adapter;
using MQTTnet.Client.Options; using MQTTnet.Client.Options;
using MQTTnet.Diagnostics; using MQTTnet.Diagnostics;
using MQTTnet.Formatter; using MQTTnet.Formatter;
using MQTTnet.Implementations; using MQTTnet.Implementations;
using System;


namespace MQTTnet.Extensions.WebSocket4Net namespace MQTTnet.Extensions.WebSocket4Net
{ {
public class WebSocket4NetMqttClientAdapterFactory : IMqttClientAdapterFactory public class WebSocket4NetMqttClientAdapterFactory : IMqttClientAdapterFactory
{ {
public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger)
public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger)
{ {
if (options == null) throw new ArgumentNullException(nameof(options)); if (options == null) throw new ArgumentNullException(nameof(options));
if (logger == null) throw new ArgumentNullException(nameof(logger)); if (logger == null) throw new ArgumentNullException(nameof(logger));
@@ -17,19 +17,19 @@ namespace MQTTnet.Extensions.WebSocket4Net
switch (options.ChannelOptions) switch (options.ChannelOptions)
{ {
case MqttClientTcpOptions _: case MqttClientTcpOptions _:
{
return new MqttChannelAdapter(new MqttTcpChannel(options), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger);
}
{
return new MqttChannelAdapter(new MqttTcpChannel(options), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger);
}


case MqttClientWebSocketOptions webSocketOptions: case MqttClientWebSocketOptions webSocketOptions:
{
return new MqttChannelAdapter(new WebSocket4NetMqttChannel(options, webSocketOptions), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger);
}
{
return new MqttChannelAdapter(new WebSocket4NetMqttChannel(options, webSocketOptions), new MqttPacketFormatterAdapter(options.ProtocolVersion), logger);
}


default: default:
{
throw new NotSupportedException();
}
{
throw new NotSupportedException();
}
} }
} }
} }


+ 43
- 43
Source/MQTTnet.Server/Logging/MqttNetChildLoggerWrapper.cs View File

@@ -1,43 +1,43 @@
using System;
using MQTTnet.Diagnostics;
namespace MQTTnet.Server.Logging
{
public class MqttNetChildLoggerWrapper : IMqttNetChildLogger
{
private readonly MqttNetLoggerWrapper _logger;
private readonly string _source;
public MqttNetChildLoggerWrapper(string source, MqttNetLoggerWrapper logger)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_source = source;
}
public IMqttNetChildLogger CreateChildLogger(string source = null)
{
return _logger.CreateChildLogger(source);
}
public void Verbose(string message, params object[] parameters)
{
_logger.Publish(MqttNetLogLevel.Verbose, _source, message, parameters, null);
}
public void Info(string message, params object[] parameters)
{
_logger.Publish(MqttNetLogLevel.Info, _source, message, parameters, null);
}
public void Warning(Exception exception, string message, params object[] parameters)
{
_logger.Publish(MqttNetLogLevel.Warning, _source, message, parameters, exception);
}
public void Error(Exception exception, string message, params object[] parameters)
{
_logger.Publish(MqttNetLogLevel.Error, _source, message, parameters, exception);
}
}
}
//using MQTTnet.Diagnostics;
//using System;
//namespace MQTTnet.Server.Logging
//{
// public class MqttNetChildLoggerWrapper : IMqttNetChildLogger
// {
// private readonly MqttNetLoggerWrapper _logger;
// private readonly string _source;
// public MqttNetChildLoggerWrapper(string source, MqttNetLoggerWrapper logger)
// {
// _logger = logger ?? throw new ArgumentNullException(nameof(logger));
// _source = source;
// }
// public IMqttNetLogger CreateChildLogger(string source = null)
// {
// return _logger.CreateChildLogger(source);
// }
// public void Verbose(string message, params object[] parameters)
// {
// _logger.Publish(MqttNetLogLevel.Verbose, _source, message, parameters, null);
// }
// public void Info(string message, params object[] parameters)
// {
// _logger.Publish(MqttNetLogLevel.Info, _source, message, parameters, null);
// }
// public void Warning(Exception exception, string message, params object[] parameters)
// {
// _logger.Publish(MqttNetLogLevel.Warning, _source, message, parameters, exception);
// }
// public void Error(Exception exception, string message, params object[] parameters)
// {
// _logger.Publish(MqttNetLogLevel.Error, _source, message, parameters, exception);
// }
// }
//}

+ 12
- 7
Source/MQTTnet.Server/Logging/MqttNetLoggerWrapper.cs View File

@@ -1,7 +1,7 @@
using System;
using System.Threading;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging;
using MQTTnet.Diagnostics; using MQTTnet.Diagnostics;
using System;
using System.Threading;


namespace MQTTnet.Server.Logging namespace MQTTnet.Server.Logging
{ {
@@ -16,9 +16,9 @@ namespace MQTTnet.Server.Logging


public event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished; public event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;


public IMqttNetChildLogger CreateChildLogger(string source = null)
public IMqttNetLogger CreateChildLogger(string source = null)
{ {
return new MqttNetChildLoggerWrapper(source, this);
return new MqttNetLogger(source);
} }


public void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception) public void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception)
@@ -33,8 +33,13 @@ namespace MQTTnet.Server.Logging
logMessagePublishedEvent.Invoke(this, new MqttNetLogMessagePublishedEventArgs(logMessage)); logMessagePublishedEvent.Invoke(this, new MqttNetLogMessagePublishedEventArgs(logMessage));
} }
} }
private static LogLevel ConvertLogLevel(MqttNetLogLevel logLevel)

public void Publish(MqttNetLogLevel logLevel, string message, object[] parameters, Exception exception)
{
Publish(logLevel, null, message, parameters, exception);
}

static LogLevel ConvertLogLevel(MqttNetLogLevel logLevel)
{ {
switch (logLevel) switch (logLevel)
{ {


+ 1
- 1
Source/MQTTnet/Adapter/IMqttClientAdapterFactory.cs View File

@@ -5,6 +5,6 @@ namespace MQTTnet.Adapter
{ {
public interface IMqttClientAdapterFactory public interface IMqttClientAdapterFactory
{ {
IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger);
IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger);
} }
} }

+ 33
- 33
Source/MQTTnet/Adapter/MqttChannelAdapter.cs View File

@@ -1,3 +1,9 @@
using MQTTnet.Channel;
using MQTTnet.Diagnostics;
using MQTTnet.Exceptions;
using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Packets;
using System; using System;
using System.IO; using System.IO;
using System.Net.Sockets; using System.Net.Sockets;
@@ -5,32 +11,26 @@ using System.Runtime.InteropServices;
using System.Security.Cryptography.X509Certificates; using System.Security.Cryptography.X509Certificates;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Channel;
using MQTTnet.Diagnostics;
using MQTTnet.Exceptions;
using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Packets;


namespace MQTTnet.Adapter namespace MQTTnet.Adapter
{ {
public class MqttChannelAdapter : Disposable, IMqttChannelAdapter
public sealed class MqttChannelAdapter : Disposable, IMqttChannelAdapter
{ {
private const uint ErrorOperationAborted = 0x800703E3;
private const int ReadBufferSize = 4096; // TODO: Move buffer size to config
const uint ErrorOperationAborted = 0x800703E3;
const int ReadBufferSize = 4096; // TODO: Move buffer size to config


private readonly SemaphoreSlim _writerSemaphore = new SemaphoreSlim(1, 1);
readonly SemaphoreSlim _writerSemaphore = new SemaphoreSlim(1, 1);


private readonly IMqttNetChildLogger _logger;
private readonly IMqttChannel _channel;
private readonly MqttPacketReader _packetReader;
readonly IMqttNetLogger _logger;
readonly IMqttChannel _channel;
readonly MqttPacketReader _packetReader;


private readonly byte[] _fixedHeaderBuffer = new byte[2];
private long _bytesReceived;
private long _bytesSent;
readonly byte[] _fixedHeaderBuffer = new byte[2];


public MqttChannelAdapter(IMqttChannel channel, MqttPacketFormatterAdapter packetFormatterAdapter, IMqttNetChildLogger logger)
long _bytesReceived;
long _bytesSent;

public MqttChannelAdapter(IMqttChannel channel, MqttPacketFormatterAdapter packetFormatterAdapter, IMqttNetLogger logger)
{ {
if (logger == null) throw new ArgumentNullException(nameof(logger)); if (logger == null) throw new ArgumentNullException(nameof(logger));


@@ -55,7 +55,7 @@ namespace MQTTnet.Adapter


public Action ReadingPacketStartedCallback { get; set; } public Action ReadingPacketStartedCallback { get; set; }
public Action ReadingPacketCompletedCallback { get; set; } public Action ReadingPacketCompletedCallback { get; set; }
public async Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken) public async Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
{ {
ThrowIfDisposed(); ThrowIfDisposed();
@@ -207,7 +207,18 @@ namespace MQTTnet.Adapter
Interlocked.Exchange(ref _bytesSent, 0L); Interlocked.Exchange(ref _bytesSent, 0L);
} }


private async Task<ReceivedMqttPacket> ReceiveAsync(CancellationToken cancellationToken)
protected override void Dispose(bool disposing)
{
if (disposing)
{
_channel?.Dispose();
_writerSemaphore?.Dispose();
}

base.Dispose(disposing);
}

async Task<ReceivedMqttPacket> ReceiveAsync(CancellationToken cancellationToken)
{ {
var readFixedHeaderResult = await _packetReader.ReadFixedHeaderAsync(_fixedHeaderBuffer, cancellationToken).ConfigureAwait(false); var readFixedHeaderResult = await _packetReader.ReadFixedHeaderAsync(_fixedHeaderBuffer, cancellationToken).ConfigureAwait(false);


@@ -267,25 +278,14 @@ namespace MQTTnet.Adapter
} }
} }


protected override void Dispose(bool disposing)
{
if (disposing)
{
_channel?.Dispose();
_writerSemaphore?.Dispose();
}
base.Dispose(disposing);
}

private static bool IsWrappedException(Exception exception)
static bool IsWrappedException(Exception exception)
{ {
return exception is OperationCanceledException || return exception is OperationCanceledException ||
exception is MqttCommunicationTimedOutException || exception is MqttCommunicationTimedOutException ||
exception is MqttCommunicationException; exception is MqttCommunicationException;
} }


private static void WrapException(Exception exception)
static void WrapException(Exception exception)
{ {
if (exception is IOException && exception.InnerException is SocketException innerException) if (exception is IOException && exception.InnerException is SocketException innerException)
{ {


+ 5
- 5
Source/MQTTnet/Client/MqttClient.cs View File

@@ -1,7 +1,3 @@
using System;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Adapter; using MQTTnet.Adapter;
using MQTTnet.Client.Connecting; using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting; using MQTTnet.Client.Disconnecting;
@@ -17,6 +13,10 @@ using MQTTnet.Internal;
using MQTTnet.PacketDispatcher; using MQTTnet.PacketDispatcher;
using MQTTnet.Packets; using MQTTnet.Packets;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using System;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;


namespace MQTTnet.Client namespace MQTTnet.Client
{ {
@@ -29,7 +29,7 @@ namespace MQTTnet.Client
private readonly object _disconnectLock = new object(); private readonly object _disconnectLock = new object();


private readonly IMqttClientAdapterFactory _adapterFactory; private readonly IMqttClientAdapterFactory _adapterFactory;
private readonly IMqttNetChildLogger _logger;
private readonly IMqttNetLogger _logger;


private CancellationTokenSource _backgroundCancellationTokenSource; private CancellationTokenSource _backgroundCancellationTokenSource;
private Task _packetReceiverTask; private Task _packetReceiverTask;


+ 8
- 8
Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs View File

@@ -1,19 +1,19 @@
using System;
using MQTTnet.Client.ExtendedAuthenticationExchange;
using MQTTnet.Formatter;
using System;
using System.Linq; using System.Linq;
using System.Text; using System.Text;
using MQTTnet.Client.ExtendedAuthenticationExchange;
using MQTTnet.Formatter;


namespace MQTTnet.Client.Options namespace MQTTnet.Client.Options
{ {
public class MqttClientOptionsBuilder public class MqttClientOptionsBuilder
{ {
private readonly MqttClientOptions _options = new MqttClientOptions();
readonly MqttClientOptions _options = new MqttClientOptions();


private MqttClientTcpOptions _tcpOptions;
private MqttClientWebSocketOptions _webSocketOptions;
private MqttClientOptionsBuilderTlsParameters _tlsParameters;
private MqttClientWebSocketProxyOptions _proxyOptions;
MqttClientTcpOptions _tcpOptions;
MqttClientWebSocketOptions _webSocketOptions;
MqttClientOptionsBuilderTlsParameters _tlsParameters;
MqttClientWebSocketProxyOptions _proxyOptions;


public MqttClientOptionsBuilder WithProtocolVersion(MqttProtocolVersion value) public MqttClientOptionsBuilder WithProtocolVersion(MqttProtocolVersion value)
{ {


+ 0
- 17
Source/MQTTnet/Diagnostics/IMqttNetChildLogger.cs View File

@@ -1,17 +0,0 @@
using System;

namespace MQTTnet.Diagnostics
{
public interface IMqttNetChildLogger
{
IMqttNetChildLogger CreateChildLogger(string source = null);

void Verbose(string message, params object[] parameters);

void Info(string message, params object[] parameters);

void Warning(Exception exception, string message, params object[] parameters);

void Error(Exception exception, string message, params object[] parameters);
}
}

+ 2
- 2
Source/MQTTnet/Diagnostics/IMqttNetLogger.cs View File

@@ -6,8 +6,8 @@ namespace MQTTnet.Diagnostics
{ {
event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished; event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;


IMqttNetChildLogger CreateChildLogger(string source = null);
IMqttNetLogger CreateChildLogger(string source = null);


void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception);
void Publish(MqttNetLogLevel logLevel, string message, object[] parameters, Exception exception);
} }
} }

+ 0
- 51
Source/MQTTnet/Diagnostics/MqttNetChildLogger.cs View File

@@ -1,51 +0,0 @@
using System;

namespace MQTTnet.Diagnostics
{
public class MqttNetChildLogger : IMqttNetChildLogger
{
private readonly IMqttNetLogger _logger;
private readonly string _source;

public MqttNetChildLogger(IMqttNetLogger logger, string source)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_source = source;
}

public IMqttNetChildLogger CreateChildLogger(string source)
{
string childSource;
if (!string.IsNullOrEmpty(_source))
{
childSource = _source + "." + source;
}
else
{
childSource = source;
}

return new MqttNetChildLogger(_logger, childSource);
}

public void Verbose(string message, params object[] parameters)
{
_logger.Publish(MqttNetLogLevel.Verbose, _source, message, parameters, null);
}

public void Info(string message, params object[] parameters)
{
_logger.Publish(MqttNetLogLevel.Info, _source, message, parameters, null);
}

public void Warning(Exception exception, string message, params object[] parameters)
{
_logger.Publish(MqttNetLogLevel.Warning, _source, message, parameters, exception);
}

public void Error(Exception exception, string message, params object[] parameters)
{
_logger.Publish(MqttNetLogLevel.Error, _source, message, parameters, exception);
}
}
}

+ 4
- 0
Source/MQTTnet/Diagnostics/MqttNetLogMessagePublishedEventArgs.cs View File

@@ -7,8 +7,12 @@ namespace MQTTnet.Diagnostics
public MqttNetLogMessagePublishedEventArgs(MqttNetLogMessage logMessage) public MqttNetLogMessagePublishedEventArgs(MqttNetLogMessage logMessage)
{ {
TraceMessage = logMessage ?? throw new ArgumentNullException(nameof(logMessage)); TraceMessage = logMessage ?? throw new ArgumentNullException(nameof(logMessage));
LogMessage = logMessage ?? throw new ArgumentNullException(nameof(logMessage));
} }


[Obsolete("Use new proeprty LogMessage instead.")]
public MqttNetLogMessage TraceMessage { get; } public MqttNetLogMessage TraceMessage { get; }

public MqttNetLogMessage LogMessage { get; }
} }
} }

+ 12
- 6
Source/MQTTnet/Diagnostics/MqttNetLogger.cs View File

@@ -4,21 +4,27 @@ namespace MQTTnet.Diagnostics
{ {
public class MqttNetLogger : IMqttNetLogger public class MqttNetLogger : IMqttNetLogger
{ {
private readonly string _logId;
readonly string _logId;
readonly string _source;


public MqttNetLogger(string logId = null)
public MqttNetLogger(string source, string logId = null)
{ {
_source = source;
_logId = logId; _logId = logId;
} }


public MqttNetLogger()
{
}

public event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished; public event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;


public IMqttNetChildLogger CreateChildLogger(string source = null)
public IMqttNetLogger CreateChildLogger(string source = null)
{ {
return new MqttNetChildLogger(this, source);
return new MqttNetLogger(source, _logId);
} }


public void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception)
public void Publish(MqttNetLogLevel logLevel, string message, object[] parameters, Exception exception)
{ {
var hasLocalListeners = LogMessagePublished != null; var hasLocalListeners = LogMessagePublished != null;
var hasGlobalListeners = MqttNetGlobalLogger.HasListeners; var hasGlobalListeners = MqttNetGlobalLogger.HasListeners;
@@ -40,7 +46,7 @@ namespace MQTTnet.Diagnostics
} }
} }


var traceMessage = new MqttNetLogMessage(_logId, DateTime.UtcNow, Environment.CurrentManagedThreadId, source, logLevel, message, exception);
var traceMessage = new MqttNetLogMessage(_logId, DateTime.UtcNow, Environment.CurrentManagedThreadId, _source, logLevel, message, exception);


if (hasGlobalListeners) if (hasGlobalListeners)
{ {


+ 35
- 0
Source/MQTTnet/Diagnostics/MqttNetLoggerExtensions.cs View File

@@ -0,0 +1,35 @@
using System;

namespace MQTTnet.Diagnostics
{
public static class MqttNetLoggerExtensions
{
public static void Verbose(this IMqttNetLogger logger, string message, params object[] parameters)
{
if (logger is null) throw new ArgumentNullException(nameof(logger));

logger.Publish(MqttNetLogLevel.Verbose, message, parameters, null);
}

public static void Info(this IMqttNetLogger logger, string message, params object[] parameters)
{
if (logger is null) throw new ArgumentNullException(nameof(logger));

logger.Publish(MqttNetLogLevel.Info, message, parameters, null);
}

public static void Warning(this IMqttNetLogger logger, Exception exception, string message, params object[] parameters)
{
if (logger is null) throw new ArgumentNullException(nameof(logger));

logger.Publish(MqttNetLogLevel.Warning, message, parameters, exception);
}

public static void Error(this IMqttNetLogger logger, Exception exception, string message, params object[] parameters)
{
if (logger is null) throw new ArgumentNullException(nameof(logger));

logger.Publish(MqttNetLogLevel.Error, message, parameters, exception);
}
}
}

Source/MQTTnet/Diagnostics/TargetFrameworkInfoProvider.cs → Source/MQTTnet/Diagnostics/TargetFrameworkProvider.cs View File

@@ -1,6 +1,6 @@
namespace MQTTnet.Diagnostics namespace MQTTnet.Diagnostics
{ {
public static class TargetFrameworkInfoProvider
public static class TargetFrameworkProvider
{ {
public static string TargetFramework public static string TargetFramework
{ {

+ 4
- 4
Source/MQTTnet/Implementations/MqttClientAdapterFactory.cs View File

@@ -1,17 +1,17 @@
using System;
using MQTTnet.Adapter;
using MQTTnet.Adapter;
using MQTTnet.Client.Options; using MQTTnet.Client.Options;
using MQTTnet.Diagnostics; using MQTTnet.Diagnostics;
using MQTTnet.Formatter; using MQTTnet.Formatter;
using System;


namespace MQTTnet.Implementations namespace MQTTnet.Implementations
{ {
public class MqttClientAdapterFactory : IMqttClientAdapterFactory public class MqttClientAdapterFactory : IMqttClientAdapterFactory
{ {
public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger)
public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger)
{ {
if (options == null) throw new ArgumentNullException(nameof(options)); if (options == null) throw new ArgumentNullException(nameof(options));
switch (options.ChannelOptions) switch (options.ChannelOptions)
{ {
case MqttClientTcpOptions _: case MqttClientTcpOptions _:


+ 6
- 6
Source/MQTTnet/Implementations/MqttTcpServerAdapter.Uwp.cs View File

@@ -11,14 +11,14 @@ using System.Threading.Tasks;


namespace MQTTnet.Implementations namespace MQTTnet.Implementations
{ {
public class MqttTcpServerAdapter : IMqttServerAdapter
public sealed class MqttTcpServerAdapter : IMqttServerAdapter
{ {
private readonly IMqttNetChildLogger _logger;
readonly IMqttNetLogger _logger;


private IMqttServerOptions _options;
private StreamSocketListener _listener;
IMqttServerOptions _options;
StreamSocketListener _listener;


public MqttTcpServerAdapter(IMqttNetChildLogger logger)
public MqttTcpServerAdapter(IMqttNetLogger logger)
{ {
if (logger == null) throw new ArgumentNullException(nameof(logger)); if (logger == null) throw new ArgumentNullException(nameof(logger));


@@ -68,7 +68,7 @@ namespace MQTTnet.Implementations
_listener = null; _listener = null;
} }


private async void OnConnectionReceivedAsync(StreamSocketListener sender, StreamSocketListenerConnectionReceivedEventArgs args)
async void OnConnectionReceivedAsync(StreamSocketListener sender, StreamSocketListenerConnectionReceivedEventArgs args)
{ {
try try
{ {


+ 22
- 21
Source/MQTTnet/Implementations/MqttTcpServerAdapter.cs View File

@@ -1,4 +1,8 @@
#if !WINDOWS_UWP #if !WINDOWS_UWP
using MQTTnet.Adapter;
using MQTTnet.Diagnostics;
using MQTTnet.Internal;
using MQTTnet.Server;
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Net; using System.Net;
@@ -6,21 +10,17 @@ using System.Net.Sockets;
using System.Security.Cryptography.X509Certificates; using System.Security.Cryptography.X509Certificates;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Adapter;
using MQTTnet.Diagnostics;
using MQTTnet.Internal;
using MQTTnet.Server;


namespace MQTTnet.Implementations namespace MQTTnet.Implementations
{ {
public class MqttTcpServerAdapter : Disposable, IMqttServerAdapter
public sealed class MqttTcpServerAdapter : Disposable, IMqttServerAdapter
{ {
private readonly List<MqttTcpServerListener> _listeners = new List<MqttTcpServerListener>();
private readonly IMqttNetChildLogger _logger;
readonly List<MqttTcpServerListener> _listeners = new List<MqttTcpServerListener>();
readonly IMqttNetLogger _logger;


private CancellationTokenSource _cancellationTokenSource;
CancellationTokenSource _cancellationTokenSource;


public MqttTcpServerAdapter(IMqttNetChildLogger logger)
public MqttTcpServerAdapter(IMqttNetLogger logger)
{ {
if (logger == null) throw new ArgumentNullException(nameof(logger)); if (logger == null) throw new ArgumentNullException(nameof(logger));


@@ -59,7 +59,7 @@ namespace MQTTnet.Implementations
{ {
tlsCertificate = new X509Certificate2(options.TlsEndpointOptions.Certificate, options.TlsEndpointOptions.CertificateCredentials.Password); tlsCertificate = new X509Certificate2(options.TlsEndpointOptions.Certificate, options.TlsEndpointOptions.CertificateCredentials.Password);
} }
if (!tlsCertificate.HasPrivateKey) if (!tlsCertificate.HasPrivateKey)
{ {
throw new InvalidOperationException("The certificate for TLS encryption must contain the private key."); throw new InvalidOperationException("The certificate for TLS encryption must contain the private key.");
@@ -77,7 +77,17 @@ namespace MQTTnet.Implementations
return Task.FromResult(0); return Task.FromResult(0);
} }


private void Cleanup()
protected override void Dispose(bool disposing)
{
if (disposing)
{
Cleanup();
}

base.Dispose(disposing);
}

void Cleanup()
{ {
_cancellationTokenSource?.Cancel(false); _cancellationTokenSource?.Cancel(false);
_cancellationTokenSource?.Dispose(); _cancellationTokenSource?.Dispose();
@@ -91,16 +101,7 @@ namespace MQTTnet.Implementations
_listeners.Clear(); _listeners.Clear();
} }


protected override void Dispose(bool disposing)
{
if (disposing)
{
Cleanup();
}
base.Dispose(disposing);
}

private void RegisterListeners(MqttServerTcpEndpointBaseOptions options, X509Certificate2 tlsCertificate, CancellationToken cancellationToken)
void RegisterListeners(MqttServerTcpEndpointBaseOptions options, X509Certificate2 tlsCertificate, CancellationToken cancellationToken)
{ {
if (!options.BoundInterNetworkAddress.Equals(IPAddress.None)) if (!options.BoundInterNetworkAddress.Equals(IPAddress.None))
{ {


+ 22
- 22
Source/MQTTnet/Implementations/MqttTcpServerListener.cs View File

@@ -1,4 +1,9 @@
#if !WINDOWS_UWP #if !WINDOWS_UWP
using MQTTnet.Adapter;
using MQTTnet.Diagnostics;
using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Server;
using System; using System;
using System.IO; using System.IO;
using System.Net; using System.Net;
@@ -7,21 +12,16 @@ using System.Net.Sockets;
using System.Security.Cryptography.X509Certificates; using System.Security.Cryptography.X509Certificates;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Adapter;
using MQTTnet.Diagnostics;
using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Server;


namespace MQTTnet.Implementations namespace MQTTnet.Implementations
{ {
public class MqttTcpServerListener : IDisposable
public sealed class MqttTcpServerListener : IDisposable
{ {
private readonly IMqttNetChildLogger _logger;
private readonly AddressFamily _addressFamily;
private readonly MqttServerTcpEndpointBaseOptions _options;
private readonly MqttServerTlsTcpEndpointOptions _tlsOptions;
private readonly X509Certificate2 _tlsCertificate;
readonly IMqttNetLogger _logger;
readonly AddressFamily _addressFamily;
readonly MqttServerTcpEndpointBaseOptions _options;
readonly MqttServerTlsTcpEndpointOptions _tlsOptions;
readonly X509Certificate2 _tlsCertificate;


private Socket _socket; private Socket _socket;
private IPEndPoint _localEndPoint; private IPEndPoint _localEndPoint;
@@ -30,7 +30,7 @@ namespace MQTTnet.Implementations
AddressFamily addressFamily, AddressFamily addressFamily,
MqttServerTcpEndpointBaseOptions options, MqttServerTcpEndpointBaseOptions options,
X509Certificate2 tlsCertificate, X509Certificate2 tlsCertificate,
IMqttNetChildLogger logger)
IMqttNetLogger logger)
{ {
_addressFamily = addressFamily; _addressFamily = addressFamily;
_options = options; _options = options;
@@ -67,12 +67,12 @@ namespace MQTTnet.Implementations
{ {
_socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true); _socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
} }
if (_options.NoDelay) if (_options.NoDelay)
{ {
_socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.NoDelay, true); _socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.NoDelay, true);
} }
_socket.Bind(_localEndPoint); _socket.Bind(_localEndPoint);
_socket.Listen(_options.ConnectionBacklog); _socket.Listen(_options.ConnectionBacklog);


@@ -87,7 +87,7 @@ namespace MQTTnet.Implementations
throw; throw;
} }


_logger.Warning(exception,"Error while creating listener socket for local end point '{0}'.", _localEndPoint);
_logger.Warning(exception, "Error while creating listener socket for local end point '{0}'.", _localEndPoint);
return false; return false;
} }
} }
@@ -101,7 +101,7 @@ namespace MQTTnet.Implementations
#endif #endif
} }


private async Task AcceptClientConnectionsAsync(CancellationToken cancellationToken)
async Task AcceptClientConnectionsAsync(CancellationToken cancellationToken)
{ {
while (!cancellationToken.IsCancellationRequested) while (!cancellationToken.IsCancellationRequested)
{ {
@@ -116,7 +116,7 @@ namespace MQTTnet.Implementations
Task.Run(() => TryHandleClientConnectionAsync(clientSocket), cancellationToken).Forget(_logger); Task.Run(() => TryHandleClientConnectionAsync(clientSocket), cancellationToken).Forget(_logger);
} }
catch (OperationCanceledException) catch (OperationCanceledException)
{
{
} }
catch (Exception exception) catch (Exception exception)
{ {
@@ -128,14 +128,14 @@ namespace MQTTnet.Implementations
continue; continue;
} }
} }
_logger.Error(exception, $"Error while accepting connection at TCP listener {_localEndPoint} TLS={_tlsCertificate != null}."); _logger.Error(exception, $"Error while accepting connection at TCP listener {_localEndPoint} TLS={_tlsCertificate != null}.");
await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false); await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken).ConfigureAwait(false);
} }
} }
} }


private async Task TryHandleClientConnectionAsync(Socket clientSocket)
async Task TryHandleClientConnectionAsync(Socket clientSocket)
{ {
Stream stream = null; Stream stream = null;
string remoteEndPoint = null; string remoteEndPoint = null;
@@ -160,9 +160,9 @@ namespace MQTTnet.Implementations
var sslStream = new SslStream(stream, false, _tlsOptions.RemoteCertificateValidationCallback); var sslStream = new SslStream(stream, false, _tlsOptions.RemoteCertificateValidationCallback);


await sslStream.AuthenticateAsServerAsync( await sslStream.AuthenticateAsServerAsync(
_tlsCertificate,
_tlsOptions.ClientCertificateRequired,
_tlsOptions.SslProtocol,
_tlsCertificate,
_tlsOptions.ClientCertificateRequired,
_tlsOptions.SslProtocol,
_tlsOptions.CheckCertificateRevocation).ConfigureAwait(false); _tlsOptions.CheckCertificateRevocation).ConfigureAwait(false);


stream = sslStream; stream = sslStream;


+ 3
- 3
Source/MQTTnet/Internal/TaskExtensions.cs View File

@@ -1,11 +1,11 @@
using System.Threading.Tasks;
using MQTTnet.Diagnostics;
using MQTTnet.Diagnostics;
using System.Threading.Tasks;


namespace MQTTnet.Internal namespace MQTTnet.Internal
{ {
public static class TaskExtensions public static class TaskExtensions
{ {
public static void Forget(this Task task, IMqttNetChildLogger logger)
public static void Forget(this Task task, IMqttNetLogger logger)
{ {
task?.ContinueWith(t => task?.ContinueWith(t =>
{ {


+ 19
- 0
Source/MQTTnet/LowLevelClient/ILowLevelMqttClient.cs View File

@@ -0,0 +1,19 @@
using MQTTnet.Client.Options;
using MQTTnet.Packets;
using System;
using System.Threading;
using System.Threading.Tasks;

namespace MQTTnet.LowLevelClient
{
public interface ILowLevelMqttClient : IDisposable
{
Task ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken);

Task DisconnectAsync(CancellationToken cancellationToken);

Task SendAsync(MqttBasePacket packet, CancellationToken cancellationToken);

Task<MqttBasePacket> ReceiveAsync(CancellationToken cancellationToken);
}
}

+ 128
- 0
Source/MQTTnet/LowLevelClient/LowLevelMqttClient.cs View File

@@ -0,0 +1,128 @@
using MQTTnet.Adapter;
using MQTTnet.Client.Options;
using MQTTnet.Diagnostics;
using MQTTnet.Packets;
using System;
using System.Threading;
using System.Threading.Tasks;

namespace MQTTnet.LowLevelClient
{
public sealed class LowLevelMqttClient : ILowLevelMqttClient
{
readonly IMqttNetLogger _logger;
readonly IMqttClientAdapterFactory _clientAdapterFactory;

IMqttChannelAdapter _adapter;
IMqttClientOptions _options;

public LowLevelMqttClient(IMqttClientAdapterFactory clientAdapterFactory, IMqttNetLogger logger)
{
if (clientAdapterFactory is null) throw new ArgumentNullException(nameof(clientAdapterFactory));
if (logger is null) throw new ArgumentNullException(nameof(logger));

_clientAdapterFactory = clientAdapterFactory;
_logger = logger.CreateChildLogger(nameof(LowLevelMqttClient));
}

bool IsConnected => _adapter != null;

public async Task ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken)
{
if (options is null) throw new ArgumentNullException(nameof(options));

if (_adapter != null)
{
throw new InvalidOperationException("Low level MQTT client is already connected. Disconnect first before connecting again.");
}

var newAdapter = _clientAdapterFactory.CreateClientAdapter(options, _logger);

try
{
_logger.Verbose($"Trying to connect with server '{options.ChannelOptions}' (Timeout={options.CommunicationTimeout}).");
await newAdapter.ConnectAsync(options.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
_logger.Verbose("Connection with server established.");

_options = options;
}
catch (Exception)
{
_adapter.Dispose();
throw;
}

_adapter = newAdapter;
}

public async Task DisconnectAsync(CancellationToken cancellationToken)
{
if (_adapter == null)
{
return;
}

await SafeDisconnect(cancellationToken).ConfigureAwait(false);
_adapter = null;
}

public async Task SendAsync(MqttBasePacket packet, CancellationToken cancellationToken)
{
if (packet is null) throw new ArgumentNullException(nameof(packet));

if (_adapter == null)
{
throw new InvalidOperationException("Low level MQTT client is not connected.");
}

try
{
await _adapter.SendPacketAsync(packet, _options.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
}
catch (Exception)
{
await SafeDisconnect(cancellationToken).ConfigureAwait(false);
throw;
}
}

public async Task<MqttBasePacket> ReceiveAsync(CancellationToken cancellationToken)
{
if (_adapter == null)
{
throw new InvalidOperationException("Low level MQTT client is not connected.");
}

try
{
return await _adapter.ReceivePacketAsync(_options.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
}
catch (Exception)
{
await SafeDisconnect(cancellationToken).ConfigureAwait(false);
throw;
}
}

public void Dispose()
{
_adapter?.Dispose();
}

async Task SafeDisconnect(CancellationToken cancellationToken)
{
try
{
await _adapter.DisconnectAsync(_options.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
}
catch (Exception exception)
{
_logger.Error(exception, "Error while disconnecting.");
}
finally
{
_adapter.Dispose();
}
}
}
}

+ 35
- 7
Source/MQTTnet/MqttFactory.cs View File

@@ -1,18 +1,19 @@
using System;
using System.Collections.Generic;
using MQTTnet.Adapter;
using MQTTnet.Adapter;
using MQTTnet.Client; using MQTTnet.Client;
using MQTTnet.Diagnostics; using MQTTnet.Diagnostics;
using MQTTnet.Implementations; using MQTTnet.Implementations;
using MQTTnet.LowLevelClient;
using MQTTnet.Server; using MQTTnet.Server;
using System;
using System.Collections.Generic;


namespace MQTTnet namespace MQTTnet
{ {
public class MqttFactory : IMqttFactory
public sealed class MqttFactory : IMqttFactory
{ {
private IMqttClientAdapterFactory _clientAdapterFactory = new MqttClientAdapterFactory();
IMqttClientAdapterFactory _clientAdapterFactory = new MqttClientAdapterFactory();


public MqttFactory() : this(new MqttNetLogger())
public MqttFactory() : this(new MqttNetLogger(null, null))
{ {
} }


@@ -29,6 +30,33 @@ namespace MQTTnet
return this; return this;
} }


public ILowLevelMqttClient CreateLowLevelMqttClient()
{
return CreateLowLevelMqttClient(DefaultLogger);
}

public ILowLevelMqttClient CreateLowLevelMqttClient(IMqttNetLogger logger)
{
if (logger == null) throw new ArgumentNullException(nameof(logger));

return new LowLevelMqttClient(_clientAdapterFactory, logger);
}

public ILowLevelMqttClient CreateLowLevelMqttClient(IMqttClientAdapterFactory clientAdapterFactory)
{
if (clientAdapterFactory == null) throw new ArgumentNullException(nameof(clientAdapterFactory));

return new LowLevelMqttClient(_clientAdapterFactory, DefaultLogger);
}

public ILowLevelMqttClient CreateLowLevelMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory clientAdapterFactoryy)
{
if (logger == null) throw new ArgumentNullException(nameof(logger));
if (clientAdapterFactoryy == null) throw new ArgumentNullException(nameof(clientAdapterFactoryy));

return new LowLevelMqttClient(_clientAdapterFactory, logger);
}

public IMqttClient CreateMqttClient() public IMqttClient CreateMqttClient()
{ {
return CreateMqttClient(DefaultLogger); return CreateMqttClient(DefaultLogger);
@@ -79,7 +107,7 @@ namespace MQTTnet
public IMqttServer CreateMqttServer(IEnumerable<IMqttServerAdapter> serverAdapters) public IMqttServer CreateMqttServer(IEnumerable<IMqttServerAdapter> serverAdapters)
{ {
if (serverAdapters == null) throw new ArgumentNullException(nameof(serverAdapters)); if (serverAdapters == null) throw new ArgumentNullException(nameof(serverAdapters));
return new MqttServer(serverAdapters, DefaultLogger.CreateChildLogger()); return new MqttServer(serverAdapters, DefaultLogger.CreateChildLogger());
} }
} }

+ 3
- 3
Source/MQTTnet/Server/IMqttRetainedMessagesManager.cs View File

@@ -1,12 +1,12 @@
using System.Collections.Generic;
using MQTTnet.Diagnostics;
using System.Collections.Generic;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Diagnostics;


namespace MQTTnet.Server namespace MQTTnet.Server
{ {
public interface IMqttRetainedMessagesManager public interface IMqttRetainedMessagesManager
{ {
Task Start(IMqttServerOptions options, IMqttNetChildLogger logger);
Task Start(IMqttServerOptions options, IMqttNetLogger logger);


Task LoadMessagesAsync(); Task LoadMessagesAsync();




+ 7
- 7
Source/MQTTnet/Server/MqttClientConnection.cs View File

@@ -1,8 +1,4 @@
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Adapter;
using MQTTnet.Adapter;
using MQTTnet.Client; using MQTTnet.Client;
using MQTTnet.Diagnostics; using MQTTnet.Diagnostics;
using MQTTnet.Exceptions; using MQTTnet.Exceptions;
@@ -12,6 +8,10 @@ using MQTTnet.PacketDispatcher;
using MQTTnet.Packets; using MQTTnet.Packets;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using MQTTnet.Server.Status; using MQTTnet.Server.Status;
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;


namespace MQTTnet.Server namespace MQTTnet.Server
{ {
@@ -25,7 +25,7 @@ namespace MQTTnet.Server
private readonly MqttClientKeepAliveMonitor _keepAliveMonitor; private readonly MqttClientKeepAliveMonitor _keepAliveMonitor;
private readonly MqttClientSessionsManager _sessionsManager; private readonly MqttClientSessionsManager _sessionsManager;


private readonly IMqttNetChildLogger _logger;
private readonly IMqttNetLogger _logger;
private readonly IMqttServerOptions _serverOptions; private readonly IMqttServerOptions _serverOptions;


private readonly IMqttChannelAdapter _channelAdapter; private readonly IMqttChannelAdapter _channelAdapter;
@@ -49,7 +49,7 @@ namespace MQTTnet.Server
IMqttServerOptions serverOptions, IMqttServerOptions serverOptions,
MqttClientSessionsManager sessionsManager, MqttClientSessionsManager sessionsManager,
IMqttRetainedMessagesManager retainedMessagesManager, IMqttRetainedMessagesManager retainedMessagesManager,
IMqttNetChildLogger logger)
IMqttNetLogger logger)
{ {
Session = session ?? throw new ArgumentNullException(nameof(session)); Session = session ?? throw new ArgumentNullException(nameof(session));
_serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions)); _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));


+ 7
- 7
Source/MQTTnet/Server/MqttClientKeepAliveMonitor.cs View File

@@ -1,9 +1,9 @@
using System;
using MQTTnet.Diagnostics;
using MQTTnet.Internal;
using System;
using System.Diagnostics; using System.Diagnostics;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Diagnostics;
using MQTTnet.Internal;


namespace MQTTnet.Server namespace MQTTnet.Server
{ {
@@ -13,15 +13,15 @@ namespace MQTTnet.Server


private readonly string _clientId; private readonly string _clientId;
private readonly Func<Task> _keepAliveElapsedCallback; private readonly Func<Task> _keepAliveElapsedCallback;
private readonly IMqttNetChildLogger _logger;
private readonly IMqttNetLogger _logger;


private bool _isPaused; private bool _isPaused;


public MqttClientKeepAliveMonitor(string clientId, Func<Task> keepAliveElapsedCallback, IMqttNetChildLogger logger)
public MqttClientKeepAliveMonitor(string clientId, Func<Task> keepAliveElapsedCallback, IMqttNetLogger logger)
{ {
_clientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); _clientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
_keepAliveElapsedCallback = keepAliveElapsedCallback ?? throw new ArgumentNullException(nameof(keepAliveElapsedCallback)); _keepAliveElapsedCallback = keepAliveElapsedCallback ?? throw new ArgumentNullException(nameof(keepAliveElapsedCallback));
if (logger == null) throw new ArgumentNullException(nameof(logger)); if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateChildLogger(nameof(MqttClientKeepAliveMonitor)); _logger = logger.CreateChildLogger(nameof(MqttClientKeepAliveMonitor));
} }
@@ -32,7 +32,7 @@ namespace MQTTnet.Server
{ {
return; return;
} }
Task.Run(() => RunAsync(keepAlivePeriod, cancellationToken), cancellationToken).Forget(_logger); Task.Run(() => RunAsync(keepAlivePeriod, cancellationToken), cancellationToken).Forget(_logger);
} }




+ 5
- 5
Source/MQTTnet/Server/MqttClientSession.cs View File

@@ -1,18 +1,18 @@
using System;
using MQTTnet.Diagnostics;
using MQTTnet.Server.Status;
using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Diagnostics;
using MQTTnet.Server.Status;


namespace MQTTnet.Server namespace MQTTnet.Server
{ {
public class MqttClientSession public class MqttClientSession
{ {
private readonly IMqttNetChildLogger _logger;
private readonly IMqttNetLogger _logger;


private readonly DateTime _createdTimestamp = DateTime.UtcNow; private readonly DateTime _createdTimestamp = DateTime.UtcNow;


public MqttClientSession(string clientId, IDictionary<object, object> items, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions serverOptions, IMqttNetChildLogger logger)
public MqttClientSession(string clientId, IDictionary<object, object> items, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions serverOptions, IMqttNetLogger logger)
{ {
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
Items = items ?? throw new ArgumentNullException(nameof(items)); Items = items ?? throw new ArgumentNullException(nameof(items));


+ 8
- 8
Source/MQTTnet/Server/MqttClientSessionsManager.cs View File

@@ -1,15 +1,15 @@
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Adapter;
using MQTTnet.Adapter;
using MQTTnet.Diagnostics; using MQTTnet.Diagnostics;
using MQTTnet.Formatter; using MQTTnet.Formatter;
using MQTTnet.Internal; using MQTTnet.Internal;
using MQTTnet.Packets; using MQTTnet.Packets;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using MQTTnet.Server.Status; using MQTTnet.Server.Status;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;


namespace MQTTnet.Server namespace MQTTnet.Server
{ {
@@ -27,14 +27,14 @@ namespace MQTTnet.Server


private readonly IMqttRetainedMessagesManager _retainedMessagesManager; private readonly IMqttRetainedMessagesManager _retainedMessagesManager;
private readonly IMqttServerOptions _options; private readonly IMqttServerOptions _options;
private readonly IMqttNetChildLogger _logger;
private readonly IMqttNetLogger _logger;


public MqttClientSessionsManager( public MqttClientSessionsManager(
IMqttServerOptions options, IMqttServerOptions options,
IMqttRetainedMessagesManager retainedMessagesManager, IMqttRetainedMessagesManager retainedMessagesManager,
CancellationToken cancellationToken, CancellationToken cancellationToken,
MqttServerEventDispatcher eventDispatcher, MqttServerEventDispatcher eventDispatcher,
IMqttNetChildLogger logger)
IMqttNetLogger logger)
{ {
_cancellationToken = cancellationToken; _cancellationToken = cancellationToken;




+ 6
- 6
Source/MQTTnet/Server/MqttRetainedMessagesManager.cs View File

@@ -1,10 +1,10 @@
using System;
using MQTTnet.Diagnostics;
using MQTTnet.Implementations;
using MQTTnet.Internal;
using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq; using System.Linq;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Diagnostics;
using MQTTnet.Implementations;
using MQTTnet.Internal;


namespace MQTTnet.Server namespace MQTTnet.Server
{ {
@@ -14,10 +14,10 @@ namespace MQTTnet.Server
private readonly AsyncLock _messagesLock = new AsyncLock(); private readonly AsyncLock _messagesLock = new AsyncLock();
private readonly Dictionary<string, MqttApplicationMessage> _messages = new Dictionary<string, MqttApplicationMessage>(); private readonly Dictionary<string, MqttApplicationMessage> _messages = new Dictionary<string, MqttApplicationMessage>();


private IMqttNetChildLogger _logger;
private IMqttNetLogger _logger;
private IMqttServerOptions _options; private IMqttServerOptions _options;


public Task Start(IMqttServerOptions options, IMqttNetChildLogger logger)
public Task Start(IMqttServerOptions options, IMqttNetLogger logger)
{ {
if (logger == null) throw new ArgumentNullException(nameof(logger)); if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateChildLogger(nameof(MqttRetainedMessagesManager)); _logger = logger.CreateChildLogger(nameof(MqttRetainedMessagesManager));


+ 2
- 2
Source/MQTTnet/Server/MqttServer.cs View File

@@ -17,13 +17,13 @@ namespace MQTTnet.Server
{ {
private readonly MqttServerEventDispatcher _eventDispatcher; private readonly MqttServerEventDispatcher _eventDispatcher;
private readonly ICollection<IMqttServerAdapter> _adapters; private readonly ICollection<IMqttServerAdapter> _adapters;
private readonly IMqttNetChildLogger _logger;
private readonly IMqttNetLogger _logger;


private MqttClientSessionsManager _clientSessionsManager; private MqttClientSessionsManager _clientSessionsManager;
private IMqttRetainedMessagesManager _retainedMessagesManager; private IMqttRetainedMessagesManager _retainedMessagesManager;
private CancellationTokenSource _cancellationTokenSource; private CancellationTokenSource _cancellationTokenSource;


public MqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetChildLogger logger)
public MqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
{ {
if (adapters == null) throw new ArgumentNullException(nameof(adapters)); if (adapters == null) throw new ArgumentNullException(nameof(adapters));
_adapters = adapters.ToList(); _adapters = adapters.ToList();


+ 5
- 5
Source/MQTTnet/Server/MqttServerEventDispatcher.cs View File

@@ -1,15 +1,15 @@
using System;
using System.Threading.Tasks;
using MQTTnet.Client.Receiving;
using MQTTnet.Client.Receiving;
using MQTTnet.Diagnostics; using MQTTnet.Diagnostics;
using System;
using System.Threading.Tasks;


namespace MQTTnet.Server namespace MQTTnet.Server
{ {
public class MqttServerEventDispatcher public class MqttServerEventDispatcher
{ {
private readonly IMqttNetChildLogger _logger;
private readonly IMqttNetLogger _logger;


public MqttServerEventDispatcher(IMqttNetChildLogger logger)
public MqttServerEventDispatcher(IMqttNetLogger logger)
{ {
_logger = logger ?? throw new ArgumentNullException(nameof(logger)); _logger = logger ?? throw new ArgumentNullException(nameof(logger));
} }


+ 9
- 8
Source/MQTTnet/TopicFilter.cs View File

@@ -2,6 +2,7 @@


namespace MQTTnet namespace MQTTnet
{ {
// TODO: Consider renaming to "MqttTopicFilter"
public class TopicFilter public class TopicFilter
{ {
public string Topic { get; set; } public string Topic { get; set; }
@@ -26,16 +27,16 @@ namespace MQTTnet
public override string ToString() public override string ToString()
{ {
return string.Concat( return string.Concat(
"TopicFilter: [Topic=",
"TopicFilter: [Topic=",
Topic, Topic,
"] [QualityOfServiceLevel=",
"] [QualityOfServiceLevel=",
QualityOfServiceLevel, QualityOfServiceLevel,
"] [NoLocal=",
NoLocal,
"] [RetainAsPublished=",
RetainAsPublished,
"] [RetainHandling=",
RetainHandling,
"] [NoLocal=",
NoLocal,
"] [RetainAsPublished=",
RetainAsPublished,
"] [RetainHandling=",
RetainHandling,
"]"); "]");
} }
} }

+ 1
- 1
Tests/MQTTnet.Benchmarks/LoggerBenchmark.cs View File

@@ -10,7 +10,7 @@ namespace MQTTnet.Benchmarks
public class LoggerBenchmark public class LoggerBenchmark
{ {
private IMqttNetLogger _logger; private IMqttNetLogger _logger;
private IMqttNetChildLogger _childLogger;
private IMqttNetLogger _childLogger;
private bool _useHandler; private bool _useHandler;


[GlobalSetup] [GlobalSetup]


+ 1
- 1
Tests/MQTTnet.Benchmarks/Program.cs View File

@@ -9,7 +9,7 @@ namespace MQTTnet.Benchmarks
{ {
public static void Main(string[] args) public static void Main(string[] args)
{ {
Console.WriteLine($"MQTTnet - BenchmarkApp.{TargetFrameworkInfoProvider.TargetFramework}");
Console.WriteLine($"MQTTnet - BenchmarkApp.{TargetFrameworkProvider.TargetFramework}");
Console.WriteLine("1 = MessageProcessingBenchmark"); Console.WriteLine("1 = MessageProcessingBenchmark");
Console.WriteLine("2 = SerializerBenchmark"); Console.WriteLine("2 = SerializerBenchmark");
Console.WriteLine("3 = LoggerBenchmark"); Console.WriteLine("3 = LoggerBenchmark");


+ 111
- 0
Tests/MQTTnet.Core.Tests/LowLevelMqttClient_Tests.cs View File

@@ -0,0 +1,111 @@
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client.Options;
using MQTTnet.LowLevelClient;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Tests.Mockups;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace MQTTnet.Tests
{
[TestClass]
public class LowLevelMqttClient_Tests
{
public TestContext TestContext { get; set; }

[TestMethod]
public async Task Connect_And_Disconnect()
{
using (var testEnvironment = new TestEnvironment(TestContext))
{
var server = await testEnvironment.StartServerAsync();

var factory = new MqttFactory();
var lowLevelClient = factory.CreateLowLevelMqttClient();

await lowLevelClient.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build(), CancellationToken.None);

await lowLevelClient.DisconnectAsync(CancellationToken.None);
}
}

[TestMethod]
public async Task Authenticate()
{
using (var testEnvironment = new TestEnvironment(TestContext))
{
var server = await testEnvironment.StartServerAsync();

var factory = new MqttFactory();
var lowLevelClient = factory.CreateLowLevelMqttClient();

await lowLevelClient.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build(), CancellationToken.None);

var receivedPacket = await Authenticate(lowLevelClient).ConfigureAwait(false);

await lowLevelClient.DisconnectAsync(CancellationToken.None).ConfigureAwait(false);

Assert.IsNotNull(receivedPacket);
Assert.AreEqual(MqttConnectReturnCode.ConnectionAccepted, receivedPacket.ReturnCode);
}
}

[TestMethod]
public async Task Subscribe()
{
using (var testEnvironment = new TestEnvironment(TestContext))
{
var server = await testEnvironment.StartServerAsync();

var factory = new MqttFactory();
var lowLevelClient = factory.CreateLowLevelMqttClient();

await lowLevelClient.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", testEnvironment.ServerPort).Build(), CancellationToken.None);

await Authenticate(lowLevelClient).ConfigureAwait(false);

var receivedPacket = await Subscribe(lowLevelClient, "a").ConfigureAwait(false);

await lowLevelClient.DisconnectAsync(CancellationToken.None).ConfigureAwait(false);

Assert.IsNotNull(receivedPacket);
Assert.AreEqual(MqttSubscribeReturnCode.SuccessMaximumQoS0, receivedPacket.ReturnCodes[0]);
}
}

async Task<MqttConnAckPacket> Authenticate(ILowLevelMqttClient client)
{
await client.SendAsync(new MqttConnectPacket()
{
CleanSession = true,
ClientId = TestContext.TestName,
Username = "user",
Password = Encoding.UTF8.GetBytes("pass")
},
CancellationToken.None).ConfigureAwait(false);

return await client.ReceiveAsync(CancellationToken.None).ConfigureAwait(false) as MqttConnAckPacket;
}

async Task<MqttSubAckPacket> Subscribe(ILowLevelMqttClient client, string topic)
{
await client.SendAsync(new MqttSubscribePacket
{
PacketIdentifier = 1,
TopicFilters = new List<TopicFilter>
{
new TopicFilter
{
Topic = topic
}
}
},
CancellationToken.None).ConfigureAwait(false);

return await client.ReceiveAsync(CancellationToken.None).ConfigureAwait(false) as MqttSubAckPacket;
}
}
}

+ 27
- 22
Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs View File

@@ -1,6 +1,4 @@
using System.Threading;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client; using MQTTnet.Client;
using MQTTnet.Client.Connecting; using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting; using MQTTnet.Client.Disconnecting;
@@ -10,10 +8,12 @@ using MQTTnet.Client.Publishing;
using MQTTnet.Client.Receiving; using MQTTnet.Client.Receiving;
using MQTTnet.Client.Subscribing; using MQTTnet.Client.Subscribing;
using MQTTnet.Client.Unsubscribing; using MQTTnet.Client.Unsubscribing;
using System.Threading;
using System.Threading.Tasks;


namespace MQTTnet.Tests.Mockups namespace MQTTnet.Tests.Mockups
{ {
public class TestClientWrapper : IMqttClient
public sealed class TestClientWrapper : IMqttClient
{ {
public TestClientWrapper(IMqttClient implementation, TestContext testContext) public TestClientWrapper(IMqttClient implementation, TestContext testContext)
{ {
@@ -22,6 +22,7 @@ namespace MQTTnet.Tests.Mockups
} }


public IMqttClient Implementation { get; } public IMqttClient Implementation { get; }

public TestContext TestContext { get; } public TestContext TestContext { get; }


public bool IsConnected => Implementation.IsConnected; public bool IsConnected => Implementation.IsConnected;
@@ -34,28 +35,32 @@ namespace MQTTnet.Tests.Mockups


public Task<MqttClientAuthenticateResult> ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken) public Task<MqttClientAuthenticateResult> ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken)
{ {
switch (options)
if (TestContext != null)
{ {
case MqttClientOptionsBuilder builder:
{
var existingClientId = builder.Build().ClientId;
if (existingClientId != null && !existingClientId.StartsWith(TestContext.TestName))
switch (options)
{
case MqttClientOptionsBuilder builder:
{ {
builder.WithClientId(TestContext.TestName + existingClientId);
var existingClientId = builder.Build().ClientId;
if (existingClientId != null && !existingClientId.StartsWith(TestContext.TestName))
{
builder.WithClientId(TestContext.TestName + existingClientId);
}

break;
} }
}
break;
case MqttClientOptions op:
{
var existingClientId = op.ClientId;
if (existingClientId != null && !existingClientId.StartsWith(TestContext.TestName))

case MqttClientOptions op:
{ {
op.ClientId = TestContext.TestName + existingClientId;
var existingClientId = op.ClientId;
if (existingClientId != null && !existingClientId.StartsWith(TestContext.TestName))
{
op.ClientId = TestContext.TestName + existingClientId;
}

break;
} }
}
break;
default:
break;
}
} }


return Implementation.ConnectAsync(options, cancellationToken); return Implementation.ConnectAsync(options, cancellationToken);
@@ -81,7 +86,7 @@ namespace MQTTnet.Tests.Mockups
return Implementation.SendExtendedAuthenticationExchangeDataAsync(data, cancellationToken); return Implementation.SendExtendedAuthenticationExchangeDataAsync(data, cancellationToken);
} }


public Task<Client.Subscribing.MqttClientSubscribeResult> SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken)
public Task<MqttClientSubscribeResult> SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken)
{ {
return Implementation.SubscribeAsync(options, cancellationToken); return Implementation.SubscribeAsync(options, cancellationToken);
} }


+ 30
- 21
Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs View File

@@ -1,27 +1,27 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client; using MQTTnet.Client;
using MQTTnet.Client.Options; using MQTTnet.Client.Options;
using MQTTnet.Diagnostics; using MQTTnet.Diagnostics;
using MQTTnet.Internal; using MQTTnet.Internal;
using MQTTnet.Server; using MQTTnet.Server;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;


namespace MQTTnet.Tests.Mockups namespace MQTTnet.Tests.Mockups
{ {
public class TestEnvironment : Disposable
public sealed class TestEnvironment : Disposable
{ {
private readonly MqttFactory _mqttFactory = new MqttFactory();
private readonly List<IMqttClient> _clients = new List<IMqttClient>();
private readonly IMqttNetLogger _serverLogger = new MqttNetLogger("server");
private readonly IMqttNetLogger _clientLogger = new MqttNetLogger("client");
readonly MqttFactory _mqttFactory = new MqttFactory();
readonly List<IMqttClient> _clients = new List<IMqttClient>();
readonly IMqttNetLogger _serverLogger = new MqttNetLogger("server");
readonly IMqttNetLogger _clientLogger = new MqttNetLogger("client");


private readonly List<string> _serverErrors = new List<string>();
private readonly List<string> _clientErrors = new List<string>();
readonly List<string> _serverErrors = new List<string>();
readonly List<string> _clientErrors = new List<string>();


private readonly List<Exception> _exceptions = new List<Exception>();
readonly List<Exception> _exceptions = new List<Exception>();


public IMqttServer Server { get; private set; } public IMqttServer Server { get; private set; }


@@ -37,36 +37,42 @@ namespace MQTTnet.Tests.Mockups


public TestContext TestContext { get; } public TestContext TestContext { get; }


public TestEnvironment() : this(null)
{
}

public TestEnvironment(TestContext testContext) public TestEnvironment(TestContext testContext)
{ {
TestContext = testContext;

_serverLogger.LogMessagePublished += (s, e) => _serverLogger.LogMessagePublished += (s, e) =>
{ {
if (e.TraceMessage.Level == MqttNetLogLevel.Error)
if (e.LogMessage.Level == MqttNetLogLevel.Error)
{ {
lock (_serverErrors) lock (_serverErrors)
{ {
_serverErrors.Add(e.TraceMessage.ToString());
_serverErrors.Add(e.LogMessage.ToString());
} }
} }
}; };


_clientLogger.LogMessagePublished += (s, e) => _clientLogger.LogMessagePublished += (s, e) =>
{ {
lock (_clientErrors)
if (e.LogMessage.Level == MqttNetLogLevel.Error)
{ {
if (e.TraceMessage.Level == MqttNetLogLevel.Error)
lock (_clientErrors)
{ {
_clientErrors.Add(e.TraceMessage.ToString());
_clientErrors.Add(e.LogMessage.ToString());
} }
} }
}; };
TestContext = testContext;
} }


public IMqttClient CreateClient() public IMqttClient CreateClient()
{ {
var client = _mqttFactory.CreateMqttClient(_clientLogger); var client = _mqttFactory.CreateMqttClient(_clientLogger);
_clients.Add(client); _clients.Add(client);

return new TestClientWrapper(client, TestContext); return new TestClientWrapper(client, TestContext);
} }


@@ -90,15 +96,17 @@ namespace MQTTnet.Tests.Mockups


public Task<IMqttClient> ConnectClientAsync() public Task<IMqttClient> ConnectClientAsync()
{ {
return ConnectClientAsync(new MqttClientOptionsBuilder() );
return ConnectClientAsync(new MqttClientOptionsBuilder());
} }


public async Task<IMqttClient> ConnectClientAsync(MqttClientOptionsBuilder options) public async Task<IMqttClient> ConnectClientAsync(MqttClientOptionsBuilder options)
{ {
if (options == null) throw new ArgumentNullException(nameof(options)); if (options == null) throw new ArgumentNullException(nameof(options));


options = options.WithTcpServer("localhost", ServerPort);

var client = CreateClient(); var client = CreateClient();
await client.ConnectAsync(options.WithTcpServer("localhost", ServerPort).Build());
await client.ConnectAsync(options.Build());


return client; return client;
} }
@@ -150,6 +158,7 @@ namespace MQTTnet.Tests.Mockups
throw new Exception($"{_exceptions.Count} exceptions tracked.\r\n" + string.Join(Environment.NewLine, _exceptions)); throw new Exception($"{_exceptions.Count} exceptions tracked.\r\n" + string.Join(Environment.NewLine, _exceptions));
} }
} }

base.Dispose(disposing); base.Dispose(disposing);
} }




+ 10
- 10
Tests/MQTTnet.Core.Tests/Mockups/TestLogger.cs View File

@@ -1,20 +1,15 @@
using System;
using MQTTnet.Diagnostics;
using MQTTnet.Diagnostics;
using System;


namespace MQTTnet.Tests.Mockups namespace MQTTnet.Tests.Mockups
{ {
public class TestLogger : IMqttNetLogger, IMqttNetChildLogger
public class TestLogger : IMqttNetLogger
{ {
public event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished; public event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;


IMqttNetChildLogger IMqttNetLogger.CreateChildLogger(string source)
public IMqttNetLogger CreateChildLogger(string source = null)
{ {
return new MqttNetChildLogger(this, source);
}

IMqttNetChildLogger IMqttNetChildLogger.CreateChildLogger(string source)
{
return new MqttNetChildLogger(this, source);
return new TestLogger();
} }


public void Verbose(string message, params object[] parameters) public void Verbose(string message, params object[] parameters)
@@ -36,5 +31,10 @@ namespace MQTTnet.Tests.Mockups
public void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception) public void Publish(MqttNetLogLevel logLevel, string source, string message, object[] parameters, Exception exception)
{ {
} }

public void Publish(MqttNetLogLevel logLevel, string message, object[] parameters, Exception exception)
{
throw new NotImplementedException();
}
} }
} }

+ 2
- 2
Tests/MQTTnet.Core.Tests/Mockups/TestMqttCommunicationAdapterFactory.cs View File

@@ -12,8 +12,8 @@ namespace MQTTnet.Tests.Mockups
{ {
_adapter = adapter; _adapter = adapter;
} }
public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger)
public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetLogger logger)
{ {
return _adapter; return _adapter;
} }


+ 28
- 21
Tests/MQTTnet.Core.Tests/Mockups/TestServerWrapper.cs View File

@@ -1,16 +1,16 @@
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client.Publishing; using MQTTnet.Client.Publishing;
using MQTTnet.Client.Receiving; using MQTTnet.Client.Receiving;
using MQTTnet.Server; using MQTTnet.Server;
using MQTTnet.Server.Status; using MQTTnet.Server.Status;
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;


namespace MQTTnet.Tests.Mockups namespace MQTTnet.Tests.Mockups
{ {
public class TestServerWrapper : IMqttServer
public sealed class TestServerWrapper : IMqttServer
{ {
public TestServerWrapper(IMqttServer implementation, TestContext testContext, TestEnvironment testEnvironment) public TestServerWrapper(IMqttServer implementation, TestContext testContext, TestEnvironment testEnvironment)
{ {
@@ -60,22 +60,29 @@ namespace MQTTnet.Tests.Mockups


public Task StartAsync(IMqttServerOptions options) public Task StartAsync(IMqttServerOptions options)
{ {
switch (options)
if (TestContext != null)
{ {
case MqttServerOptionsBuilder builder:
if (builder.Build().ConnectionValidator == null)
{
builder.WithConnectionValidator(ConnectionValidator);
}
break;
case MqttServerOptions op:
if (op.ConnectionValidator == null)
{
op.ConnectionValidator = new MqttServerConnectionValidatorDelegate(ConnectionValidator);
}
break;
default:
break;
switch (options)
{
case MqttServerOptionsBuilder builder:
{
if (builder.Build().ConnectionValidator == null)
{
builder.WithConnectionValidator(ConnectionValidator);
}

break;
}
case MqttServerOptions op:
{
if (op.ConnectionValidator == null)
{
op.ConnectionValidator = new MqttServerConnectionValidatorDelegate(ConnectionValidator);
}

break;
}
}
} }


return Implementation.StartAsync(options); return Implementation.StartAsync(options);


+ 10
- 8
Tests/MQTTnet.Core.Tests/MqttFactory_Tests.cs View File

@@ -17,22 +17,23 @@ namespace MQTTnet.Tests


//This test compares //This test compares
//1. correct logID //1. correct logID
string logId = "logId";
var logId = "logId";
string invalidLogId = null; string invalidLogId = null;


//2. if the total log calls are the same for global and local //2. if the total log calls are the same for global and local
int globalLogCount = 0;
int localLogCount = 0;
var globalLogCount = 0;
var localLogCount = 0;


MqttNetLogger logger = new MqttNetLogger(logId);
var logger = new MqttNetLogger(logId);


//we have a theoretical bug here if a concurrent test is also logging //we have a theoretical bug here if a concurrent test is also logging
var globalLog = new EventHandler<MqttNetLogMessagePublishedEventArgs>((s, e) => var globalLog = new EventHandler<MqttNetLogMessagePublishedEventArgs>((s, e) =>
{ {
if (logId != e.TraceMessage.LogId)
if (logId != e.LogMessage.LogId)
{ {
invalidLogId = e.TraceMessage.LogId;
invalidLogId = e.LogMessage.LogId;
} }

Interlocked.Increment(ref globalLogCount); Interlocked.Increment(ref globalLogCount);
}); });


@@ -40,10 +41,11 @@ namespace MQTTnet.Tests


logger.LogMessagePublished += (s, e) => logger.LogMessagePublished += (s, e) =>
{ {
if (logId != e.TraceMessage.LogId)
if (logId != e.LogMessage.LogId)
{ {
invalidLogId = e.TraceMessage.LogId;
invalidLogId = e.LogMessage.LogId;
} }

Interlocked.Increment(ref localLogCount); Interlocked.Increment(ref localLogCount);
}; };




+ 18
- 18
Tests/MQTTnet.Core.Tests/Server_Tests.cs View File

@@ -1,11 +1,4 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Adapter; using MQTTnet.Adapter;
using MQTTnet.Client; using MQTTnet.Client;
using MQTTnet.Client.Connecting; using MQTTnet.Client.Connecting;
@@ -17,6 +10,13 @@ using MQTTnet.Implementations;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using MQTTnet.Server; using MQTTnet.Server;
using MQTTnet.Tests.Mockups; using MQTTnet.Tests.Mockups;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;


namespace MQTTnet.Tests namespace MQTTnet.Tests
{ {
@@ -54,7 +54,7 @@ namespace MQTTnet.Tests
MqttQualityOfServiceLevel.AtMostOnce, MqttQualityOfServiceLevel.AtMostOnce,
"A/B/C", "A/B/C",
MqttQualityOfServiceLevel.AtMostOnce, MqttQualityOfServiceLevel.AtMostOnce,
1,
1,
TestContext); TestContext);
} }


@@ -1016,7 +1016,7 @@ namespace MQTTnet.Tests
[TestMethod] [TestMethod]
public async Task Same_Client_Id_Connect_Disconnect_Event_Order() public async Task Same_Client_Id_Connect_Disconnect_Event_Order()
{ {
using (var testEnvironment = new TestEnvironment(TestContext))
using (var testEnvironment = new TestEnvironment())
{ {
var server = await testEnvironment.StartServerAsync(new MqttServerOptionsBuilder()); var server = await testEnvironment.StartServerAsync(new MqttServerOptionsBuilder());


@@ -1038,11 +1038,11 @@ namespace MQTTnet.Tests
} }
}); });


var clientOptions = new MqttClientOptionsBuilder()
.WithClientId("same_id");
var clientOptionsBuilder = new MqttClientOptionsBuilder()
.WithClientId(Guid.NewGuid().ToString());


// c // c
var c1 = await testEnvironment.ConnectClientAsync(clientOptions);
var c1 = await testEnvironment.ConnectClientAsync(clientOptionsBuilder);


await Task.Delay(500); await Task.Delay(500);


@@ -1050,7 +1050,7 @@ namespace MQTTnet.Tests
Assert.AreEqual("c", flow); Assert.AreEqual("c", flow);


// dc // dc
var c2 = await testEnvironment.ConnectClientAsync(clientOptions);
var c2 = await testEnvironment.ConnectClientAsync(clientOptionsBuilder);


c2.UseApplicationMessageReceivedHandler(_ => c2.UseApplicationMessageReceivedHandler(_ =>
{ {
@@ -1058,8 +1058,8 @@ namespace MQTTnet.Tests
{ {
events.Add("r"); events.Add("r");
} }

}); });

c2.SubscribeAsync("topic").Wait(); c2.SubscribeAsync("topic").Wait();


await Task.Delay(500); await Task.Delay(500);
@@ -1080,7 +1080,7 @@ namespace MQTTnet.Tests


Assert.AreEqual(false, c1.IsConnected); Assert.AreEqual(false, c1.IsConnected);
await c1.DisconnectAsync(); await c1.DisconnectAsync();
Assert.AreEqual (false, c1.IsConnected);
Assert.AreEqual(false, c1.IsConnected);


await Task.Delay(500); await Task.Delay(500);


@@ -1141,7 +1141,7 @@ namespace MQTTnet.Tests
await testEnvironment.ConnectClientAsync(); await testEnvironment.ConnectClientAsync();
} }
} }
[TestMethod] [TestMethod]
public async Task Close_Idle_Connection() public async Task Close_Idle_Connection()
{ {
@@ -1182,7 +1182,7 @@ namespace MQTTnet.Tests
// forever. This is security related. // forever. This is security related.
var client = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); var client = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await PlatformAbstractionLayer.ConnectAsync(client, "localhost", testEnvironment.ServerPort); await PlatformAbstractionLayer.ConnectAsync(client, "localhost", testEnvironment.ServerPort);
var buffer = Encoding.UTF8.GetBytes("Garbage"); var buffer = Encoding.UTF8.GetBytes("Garbage");
client.Send(buffer, buffer.Length, SocketFlags.None); client.Send(buffer, buffer.Length, SocketFlags.None);




+ 1
- 1
Tests/MQTTnet.TestApp.NetCore/Program.cs View File

@@ -17,7 +17,7 @@ namespace MQTTnet.TestApp.NetCore
{ {
public static void Main() public static void Main()
{ {
Console.WriteLine($"MQTTnet - TestApp.{TargetFrameworkInfoProvider.TargetFramework}");
Console.WriteLine($"MQTTnet - TestApp.{TargetFrameworkProvider.TargetFramework}");
Console.WriteLine("1 = Start client"); Console.WriteLine("1 = Start client");
Console.WriteLine("2 = Start server"); Console.WriteLine("2 = Start server");
Console.WriteLine("3 = Start performance test"); Console.WriteLine("3 = Start performance test");


Loading…
Cancel
Save