Compare commits

...
Author SHA1 Message Date
Christian ae84aa43d2 1846 server mqttserverstopasync doesnt indicate correct reason (#1872) 2023-11-25 16:17:16 +01:00
RidoandChristian b431a5d3fe Add ReasonCode to SendExtendedAuthenticationExchangeDataAsync (#1882)
* add reason code to ext auth

* Update ReleaseNotes.md

---------

Co-authored-by: Christian <6939810+chkr1011@users.noreply.github.com>
2023-11-25 15:45:10 +01:00
Christian b9683c2943 Improve client disconnect (#1873)
Imrpove client disconnect
2023-11-05 15:15:36 +01:00
26 changed files with 555 additions and 179 deletions
+5 -1
View File
@@ -1,3 +1,7 @@
* [Server] Fixed not working _UpdateRetainedMessageAsync_ public api (#1858, thanks to @kimdiego2098). * [Client] Added support for custom CA chain validation (#1851, thanks to @rido-min).
* [Client] Added support for custom CA chain validation (#1851, thanks to @rido-min). * [Client] Added support for custom CA chain validation (#1851, thanks to @rido-min).
* [Client] Fixed handling of unobserved tasks exceptions (#1871). * [Client] Fixed handling of unobserved tasks exceptions (#1871).
* [Client] Fixed not specified ReasonCode when using _SendExtendedAuthenticationExchangeDataAsync_ (#1882, thanks to @rido-min).
* [Server] Fixed not working _UpdateRetainedMessageAsync_ public api (#1858, thanks to @kimdiego2098).
* [Server] Added support for custom DISCONNECT packets when stopping the server or disconnect a client (BREAKING CHANGE!, #1846).
* [Server] Added new property to stop the server from accepting new connections even if it is running (#1846).
+14 -6
View File
@@ -2,6 +2,7 @@
// The .NET Foundation licenses this file to you under the MIT license. // The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information. // See the LICENSE file in the project root for more information.
using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
@@ -14,20 +15,27 @@ namespace MQTTnet.AspNetCore
{ {
public sealed class MqttHostedServer : MqttServer, IHostedService public sealed class MqttHostedServer : MqttServer, IHostedService
{ {
public MqttHostedServer(MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) readonly MqttFactory _mqttFactory;
: base(options, adapters, logger)
public MqttHostedServer(MqttFactory mqttFactory, MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) : base(
options,
adapters,
logger)
{ {
_mqttFactory = mqttFactory ?? throw new ArgumentNullException(nameof(mqttFactory));
} }
public Task StartAsync(CancellationToken cancellationToken) public async Task StartAsync(CancellationToken cancellationToken)
{ {
_ = StartAsync(); // The yield makes sure that the hosted service is considered up and running.
return Task.CompletedTask; await Task.Yield();
_ = StartAsync();
} }
public Task StopAsync(CancellationToken cancellationToken) public Task StopAsync(CancellationToken cancellationToken)
{ {
return StopAsync(); return StopAsync(_mqttFactory.CreateMqttServerStopOptionsBuilder().Build());
} }
} }
} }
@@ -4,7 +4,7 @@
using System; using System;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting; using Microsoft.Extensions.DependencyInjection.Extensions;
using MQTTnet.Adapter; using MQTTnet.Adapter;
using MQTTnet.Diagnostics; using MQTTnet.Diagnostics;
using MQTTnet.Implementations; using MQTTnet.Implementations;
@@ -14,6 +14,91 @@ namespace MQTTnet.AspNetCore
{ {
public static class ServiceCollectionExtensions public static class ServiceCollectionExtensions
{ {
public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, MqttServerOptions options)
{
if (services == null)
{
throw new ArgumentNullException(nameof(services));
}
if (options == null)
{
throw new ArgumentNullException(nameof(options));
}
services.AddSingleton(options);
services.AddHostedMqttServer();
return services;
}
public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, Action<MqttServerOptionsBuilder> configure)
{
if (services == null)
{
throw new ArgumentNullException(nameof(services));
}
if (configure == null)
{
throw new ArgumentNullException(nameof(configure));
}
var serverOptionsBuilder = new MqttServerOptionsBuilder();
configure.Invoke(serverOptionsBuilder);
var options = serverOptionsBuilder.Build();
return AddHostedMqttServer(services, options);
}
public static void AddHostedMqttServer(this IServiceCollection services)
{
// The user may have these services already registered.
services.TryAddSingleton<IMqttNetLogger>(MqttNetNullLogger.Instance);
services.TryAddSingleton(new MqttFactory());
services.AddSingleton<MqttHostedServer>();
services.AddHostedService<MqttHostedServer>();
}
public static IServiceCollection AddHostedMqttServerWithServices(this IServiceCollection services, Action<AspNetMqttServerOptionsBuilder> configure)
{
if (services == null)
{
throw new ArgumentNullException(nameof(services));
}
services.AddSingleton(
s =>
{
var builder = new AspNetMqttServerOptionsBuilder(s);
configure(builder);
return builder.Build();
});
services.AddHostedMqttServer();
return services;
}
public static IServiceCollection AddMqttConnectionHandler(this IServiceCollection services)
{
services.AddSingleton<MqttConnectionHandler>();
services.AddSingleton<IMqttServerAdapter>(s => s.GetService<MqttConnectionHandler>());
return services;
}
public static void AddMqttLogger(this IServiceCollection services, IMqttNetLogger logger)
{
if (services == null)
{
throw new ArgumentNullException(nameof(services));
}
services.AddSingleton(logger);
}
public static IServiceCollection AddMqttServer(this IServiceCollection serviceCollection, Action<MqttServerOptionsBuilder> configure = null) public static IServiceCollection AddMqttServer(this IServiceCollection serviceCollection, Action<MqttServerOptionsBuilder> configure = null)
{ {
if (serviceCollection is null) if (serviceCollection is null)
@@ -27,53 +112,10 @@ namespace MQTTnet.AspNetCore
return serviceCollection; return serviceCollection;
} }
public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, MqttServerOptions options) public static IServiceCollection AddMqttTcpServerAdapter(this IServiceCollection services)
{ {
if (options == null) throw new ArgumentNullException(nameof(options)); services.AddSingleton<MqttTcpServerAdapter>();
services.AddSingleton<IMqttServerAdapter>(s => s.GetService<MqttTcpServerAdapter>());
services.AddSingleton(options);
services.AddHostedMqttServer();
return services;
}
public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, Action<MqttServerOptionsBuilder> configure = null)
{
services.AddSingleton(s =>
{
var serverOptionsBuilder = new MqttServerOptionsBuilder();
configure?.Invoke(serverOptionsBuilder);
return serverOptionsBuilder.Build();
});
services.AddHostedMqttServer();
return services;
}
public static IServiceCollection AddHostedMqttServerWithServices(this IServiceCollection services, Action<AspNetMqttServerOptionsBuilder> configure)
{
services.AddSingleton(s =>
{
var builder = new AspNetMqttServerOptionsBuilder(s);
configure(builder);
return builder.Build();
});
services.AddHostedMqttServer();
return services;
}
static IServiceCollection AddHostedMqttServer(this IServiceCollection services)
{
var logger = new MqttNetEventLogger();
services.AddSingleton<IMqttNetLogger>(logger);
services.AddSingleton<MqttHostedServer>();
services.AddSingleton<IHostedService>(s => s.GetService<MqttHostedServer>());
services.AddSingleton<MqttServer>(s => s.GetService<MqttHostedServer>());
return services; return services;
} }
@@ -85,21 +127,5 @@ namespace MQTTnet.AspNetCore
return services; return services;
} }
public static IServiceCollection AddMqttTcpServerAdapter(this IServiceCollection services)
{
services.AddSingleton<MqttTcpServerAdapter>();
services.AddSingleton<IMqttServerAdapter>(s => s.GetService<MqttTcpServerAdapter>());
return services;
}
public static IServiceCollection AddMqttConnectionHandler(this IServiceCollection services)
{
services.AddSingleton<MqttConnectionHandler>();
services.AddSingleton<IMqttServerAdapter>(s => s.GetService<MqttConnectionHandler>());
return services;
}
} }
} }
@@ -13,6 +13,7 @@ using MQTTnet.Exceptions;
using MQTTnet.LowLevelClient; using MQTTnet.LowLevelClient;
using MQTTnet.Packets; using MQTTnet.Packets;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using MQTTnet.Server;
namespace MQTTnet.Tests.Clients.LowLevelMqttClient namespace MQTTnet.Tests.Clients.LowLevelMqttClient
{ {
@@ -17,6 +17,7 @@ using MQTTnet.Formatter;
using MQTTnet.Internal; using MQTTnet.Internal;
using MQTTnet.Packets; using MQTTnet.Packets;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using MQTTnet.Server;
using MQTTnet.Tests.Mockups; using MQTTnet.Tests.Mockups;
// ReSharper disable InconsistentNaming // ReSharper disable InconsistentNaming
@@ -10,6 +10,7 @@ using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Internal; using MQTTnet.Internal;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using MQTTnet.Server;
namespace MQTTnet.Tests.MQTTv5 namespace MQTTnet.Tests.MQTTv5
{ {
+4 -4
View File
@@ -303,7 +303,7 @@ namespace MQTTnet.Tests.Server
var server = await testEnvironment.StartServer(); var server = await testEnvironment.StartServer();
server.InterceptingPublishAsync += e => server.InterceptingPublishAsync += e =>
{ {
e.ApplicationMessage.Payload = Encoding.ASCII.GetBytes("extended"); e.ApplicationMessage.PayloadSegment = new ArraySegment<byte>(Encoding.ASCII.GetBytes("extended"));
return CompletedTask.Instance; return CompletedTask.Instance;
}; };
@@ -314,7 +314,7 @@ namespace MQTTnet.Tests.Server
var isIntercepted = false; var isIntercepted = false;
c2.ApplicationMessageReceivedAsync += e => c2.ApplicationMessageReceivedAsync += e =>
{ {
isIntercepted = string.Compare("extended", Encoding.UTF8.GetString(e.ApplicationMessage.Payload), StringComparison.Ordinal) == 0; isIntercepted = string.Compare("extended", Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment.ToArray()), StringComparison.Ordinal) == 0;
return CompletedTask.Instance; return CompletedTask.Instance;
}; };
@@ -425,7 +425,7 @@ namespace MQTTnet.Tests.Server
new MqttApplicationMessage new MqttApplicationMessage
{ {
Topic = "/test/1", Topic = "/test/1",
Payload = Encoding.UTF8.GetBytes("true"), PayloadSegment = new ArraySegment<byte>(Encoding.UTF8.GetBytes("true")),
QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce
}) })
{ {
@@ -780,7 +780,7 @@ namespace MQTTnet.Tests.Server
var client1 = await testEnvironment.ConnectClient(); var client1 = await testEnvironment.ConnectClient();
client1.ApplicationMessageReceivedAsync += e => client1.ApplicationMessageReceivedAsync += e =>
{ {
receivedBody = e.ApplicationMessage.Payload; receivedBody = e.ApplicationMessage.PayloadSegment.ToArray();
return CompletedTask.Instance; return CompletedTask.Instance;
}; };
@@ -8,6 +8,7 @@ using MQTTnet.Client;
using MQTTnet.Formatter; using MQTTnet.Formatter;
using MQTTnet.Internal; using MQTTnet.Internal;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using MQTTnet.Server;
namespace MQTTnet.Tests.Server namespace MQTTnet.Tests.Server
{ {
+1 -1
View File
@@ -283,7 +283,7 @@ namespace MQTTnet.Tests.Server
server.InterceptingPublishAsync += e => server.InterceptingPublishAsync += e =>
{ {
e.ApplicationMessage.Payload = Encoding.UTF8.GetBytes(e.SessionItems["default_payload"] as string ?? string.Empty); e.ApplicationMessage.PayloadSegment = new ArraySegment<byte>(Encoding.UTF8.GetBytes(e.SessionItems["default_payload"] as string ?? string.Empty));
return CompletedTask.Instance; return CompletedTask.Instance;
}; };
+9 -11
View File
@@ -101,7 +101,7 @@ namespace MQTTnet.Tests.Server
new MqttApplicationMessage new MqttApplicationMessage
{ {
Topic = "TestTopic1", Topic = "TestTopic1",
Payload = new byte[] { 1, 2, 3, 4 } PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 })
}); });
await testEnvironment.Server.InjectApplicationMessage( await testEnvironment.Server.InjectApplicationMessage(
@@ -109,7 +109,7 @@ namespace MQTTnet.Tests.Server
new MqttApplicationMessage new MqttApplicationMessage
{ {
Topic = "TestTopic1", Topic = "TestTopic1",
Payload = new byte[] { 1, 2, 3, 4 } PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 })
})); }));
certificateProvider.CurrentCertificate = CreateCertificate(secondOid); certificateProvider.CurrentCertificate = CreateCertificate(secondOid);
@@ -137,7 +137,7 @@ namespace MQTTnet.Tests.Server
new MqttApplicationMessage new MqttApplicationMessage
{ {
Topic = "TestTopic2", Topic = "TestTopic2",
Payload = new byte[] { 1, 2, 3, 4 } PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 })
}); });
await testEnvironment.Server.InjectApplicationMessage( await testEnvironment.Server.InjectApplicationMessage(
@@ -145,7 +145,7 @@ namespace MQTTnet.Tests.Server
new MqttApplicationMessage new MqttApplicationMessage
{ {
Topic = "TestTopic2", Topic = "TestTopic2",
Payload = new byte[] { 1, 2, 3, 4 } PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 })
})); }));
// Ensure first client still works // Ensure first client still works
@@ -153,7 +153,7 @@ namespace MQTTnet.Tests.Server
new MqttApplicationMessage new MqttApplicationMessage
{ {
Topic = "TestTopic1", Topic = "TestTopic1",
Payload = new byte[] { 1, 2, 3, 4 } PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 })
}); });
await testEnvironment.Server.InjectApplicationMessage( await testEnvironment.Server.InjectApplicationMessage(
@@ -161,7 +161,7 @@ namespace MQTTnet.Tests.Server
new MqttApplicationMessage new MqttApplicationMessage
{ {
Topic = "TestTopic1", Topic = "TestTopic1",
Payload = new byte[] { 1, 2, 3, 4 } PayloadSegment = new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 })
})); }));
await Task.Delay(1000); await Task.Delay(1000);
@@ -178,12 +178,10 @@ namespace MQTTnet.Tests.Server
var clientOptionsBuilder = testEnvironment.Factory.CreateClientOptionsBuilder(); var clientOptionsBuilder = testEnvironment.Factory.CreateClientOptionsBuilder();
clientOptionsBuilder.WithClientId(Guid.NewGuid().ToString()) clientOptionsBuilder.WithClientId(Guid.NewGuid().ToString())
.WithTcpServer("localhost", 8883) .WithTcpServer("localhost", 8883)
.WithTls( .WithTlsOptions(
tls => o =>
{ {
tls.UseTls = true; o.WithSslProtocols(SslProtocols.Tls12).WithCertificateValidationHandler(certValidator);
tls.SslProtocol = SslProtocols.Tls12;
tls.CertificateValidationHandler = certValidator;
}); });
var clientOptions = clientOptionsBuilder.Build(); var clientOptions = clientOptionsBuilder.Build();
+15 -14
View File
@@ -309,6 +309,7 @@ namespace MQTTnet.Client
AuthenticationMethod = Options.AuthenticationMethod, AuthenticationMethod = Options.AuthenticationMethod,
AuthenticationData = data.AuthenticationData, AuthenticationData = data.AuthenticationData,
ReasonString = data.ReasonString, ReasonString = data.ReasonString,
ReasonCode = data.ReasonCode,
UserProperties = data.UserProperties UserProperties = data.UserProperties
}; };
@@ -716,7 +717,7 @@ namespace MQTTnet.Client
} }
catch (Exception exception) catch (Exception exception)
{ {
_logger.Error(exception, "Error while handling application message."); _logger.Error(exception, "Error while handling application message");
} }
} }
} }
@@ -835,8 +836,7 @@ namespace MQTTnet.Client
_logger.Error(exception, "Error while receiving packets"); _logger.Error(exception, "Error while receiving packets");
} }
// The packet dispatcher is set to null when the client is being disposed so it may // The packet dispatcher is set to null when the client is being disposed so it may already being gone!
// already being gone!
_packetDispatcher?.FailAll(exception); _packetDispatcher?.FailAll(exception);
await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false); await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false);
@@ -865,7 +865,7 @@ namespace MQTTnet.Client
} }
catch (Exception exception) catch (Exception exception)
{ {
_logger.Warning(exception, "Error when sending request packet ({0}).", requestPacket.GetType().Name); _logger.Warning(exception, "Error when sending {0} request packet", requestPacket.GetType().Name);
packetAwaitable.Fail(exception); packetAwaitable.Fail(exception);
} }
@@ -877,7 +877,7 @@ namespace MQTTnet.Client
{ {
if (exception is MqttCommunicationTimedOutException) if (exception is MqttCommunicationTimedOutException)
{ {
_logger.Warning("Timeout while waiting for response packet ({0}).", typeof(TResponsePacket).Name); _logger.Warning("Timeout while waiting for {0} response packet", typeof(TResponsePacket).Name);
} }
throw; throw;
@@ -943,7 +943,7 @@ namespace MQTTnet.Client
} }
catch (Exception exception) catch (Exception exception)
{ {
_logger.Warning(exception, "Error while initiating disconnect."); _logger.Warning(exception, "Error while initiating disconnect");
} }
} }
} }
@@ -978,7 +978,7 @@ namespace MQTTnet.Client
} }
else if (packet is MqttPingReqPacket) else if (packet is MqttPingReqPacket)
{ {
throw new MqttProtocolViolationException("The PINGREQ Packet is sent from a Client to the Server only."); throw new MqttProtocolViolationException("The PINGREQ Packet is sent from a client to the server only.");
} }
else else
{ {
@@ -1000,14 +1000,15 @@ namespace MQTTnet.Client
} }
else if (exception is MqttCommunicationException) else if (exception is MqttCommunicationException)
{ {
_logger.Warning(exception, "Communication error while receiving packets."); _logger.Warning(exception, "Communication error while receiving packets");
} }
else else
{ {
_logger.Error(exception, $"Error while processing received packet ({packet.GetType().Name})."); _logger.Error(exception, "Error while processing received {0} packet", packet.GetType().Name);
} }
_packetDispatcher.FailAll(exception); // The packet dispatcher may already be gone due to disconnect etc!
_packetDispatcher?.FailAll(exception);
await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false); await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false);
} }
@@ -1017,7 +1018,7 @@ namespace MQTTnet.Client
{ {
try try
{ {
_logger.Verbose("Start sending keep alive packets."); _logger.Verbose("Start sending keep alive packets");
var keepAlivePeriod = Options.KeepAlivePeriod; var keepAlivePeriod = Options.KeepAlivePeriod;
@@ -1055,18 +1056,18 @@ namespace MQTTnet.Client
} }
else if (exception is MqttCommunicationException) else if (exception is MqttCommunicationException)
{ {
_logger.Warning(exception, "Communication error while sending/receiving keep alive packets."); _logger.Warning(exception, "Communication error while sending/receiving keep alive packets");
} }
else else
{ {
_logger.Error(exception, "Error exception while sending/receiving keep alive packets."); _logger.Error(exception, "Error exception while sending/receiving keep alive packets");
} }
await DisconnectInternal(_keepAlivePacketsSenderTask, exception, null).ConfigureAwait(false); await DisconnectInternal(_keepAlivePacketsSenderTask, exception, null).ConfigureAwait(false);
} }
finally finally
{ {
_logger.Verbose("Stopped sending keep alive packets."); _logger.Verbose("Stopped sending keep alive packets");
} }
} }
} }
@@ -5,45 +5,107 @@
using MQTTnet.Client; using MQTTnet.Client;
using MQTTnet.Packets; using MQTTnet.Packets;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using MQTTnet.Server;
using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Formatter namespace MQTTnet.Formatter
{ {
public sealed class MqttDisconnectPacketFactory public sealed class MqttDisconnectPacketFactory
{ {
readonly MqttDisconnectPacket _normalDisconnection = new MqttDisconnectPacket static readonly MqttDisconnectPacket DefaultNormalDisconnection = new MqttDisconnectPacket
{ {
ReasonCode = MqttDisconnectReasonCode.NormalDisconnection ReasonCode = MqttDisconnectReasonCode.NormalDisconnection,
UserProperties = null,
ReasonString = null,
ServerReference = null,
SessionExpiryInterval = 0
};
static readonly MqttDisconnectPacket DefaultServerShuttingDown = new MqttDisconnectPacket
{
ReasonCode = MqttDisconnectReasonCode.ServerShuttingDown,
UserProperties = null,
ReasonString = null,
ServerReference = null,
SessionExpiryInterval = 0
};
static readonly MqttDisconnectPacket DefaultUnspecifiedError = new MqttDisconnectPacket
{
ReasonCode = MqttDisconnectReasonCode.UnspecifiedError,
UserProperties = null,
ReasonString = null,
ServerReference = null,
SessionExpiryInterval = 0
}; };
public MqttDisconnectPacket Create(MqttDisconnectReasonCode reasonCode) public MqttDisconnectPacket Create(MqttDisconnectReasonCode reasonCode)
{ {
if (reasonCode == MqttDisconnectReasonCode.NormalDisconnection) if (reasonCode == MqttDisconnectReasonCode.NormalDisconnection)
{ {
return _normalDisconnection; return DefaultNormalDisconnection;
}
if (reasonCode == MqttDisconnectReasonCode.ServerShuttingDown)
{
return DefaultServerShuttingDown;
}
if (reasonCode == MqttDisconnectReasonCode.UnspecifiedError)
{
return DefaultUnspecifiedError;
} }
return new MqttDisconnectPacket return new MqttDisconnectPacket
{ {
ReasonCode = reasonCode ReasonCode = reasonCode,
UserProperties = null,
ReasonString = null,
ServerReference = null,
SessionExpiryInterval = 0
};
}
public MqttDisconnectPacket Create(MqttServerStopOptions serverStopOptions)
{
if (serverStopOptions == null)
{
return DefaultServerShuttingDown;
}
return Create(serverStopOptions.DefaultClientDisconnectOptions);
}
public MqttDisconnectPacket Create(MqttServerClientDisconnectOptions clientDisconnectOptions)
{
if (clientDisconnectOptions == null)
{
return DefaultNormalDisconnection;
}
return new MqttDisconnectPacket
{
ReasonCode = clientDisconnectOptions.ReasonCode,
UserProperties = clientDisconnectOptions.UserProperties,
ReasonString = clientDisconnectOptions.ReasonString,
ServerReference = clientDisconnectOptions.ServerReference,
SessionExpiryInterval = 0 // TODO: Not yet supported!
}; };
} }
public MqttDisconnectPacket Create(MqttClientDisconnectOptions clientDisconnectOptions) public MqttDisconnectPacket Create(MqttClientDisconnectOptions clientDisconnectOptions)
{ {
var packet = new MqttDisconnectPacket();
if (clientDisconnectOptions == null) if (clientDisconnectOptions == null)
{ {
packet.ReasonCode = MqttDisconnectReasonCode.NormalDisconnection; return DefaultNormalDisconnection;
}
else
{
packet.ReasonCode = (MqttDisconnectReasonCode)clientDisconnectOptions.Reason;
packet.UserProperties = clientDisconnectOptions.UserProperties;
packet.SessionExpiryInterval = clientDisconnectOptions.SessionExpiryInterval;
} }
return packet; return new MqttDisconnectPacket
{
ReasonCode = (MqttDisconnectReasonCode)clientDisconnectOptions.Reason,
UserProperties = clientDisconnectOptions.UserProperties,
SessionExpiryInterval = clientDisconnectOptions.SessionExpiryInterval
};
} }
} }
} }
@@ -17,4 +17,5 @@
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cmanagement/@EntryIndexedValue">True</s:Boolean> <s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cmanagement/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Coptions/@EntryIndexedValue">True</s:Boolean> <s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Coptions/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cstatus/@EntryIndexedValue">True</s:Boolean> <s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cstatus/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cstopping/@EntryIndexedValue">True</s:Boolean>
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cstorage/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary> <s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=server_005Cstorage/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>
+11
View File
@@ -11,6 +11,7 @@ using MQTTnet.Diagnostics;
using MQTTnet.Implementations; using MQTTnet.Implementations;
using MQTTnet.LowLevelClient; using MQTTnet.LowLevelClient;
using MQTTnet.Server; using MQTTnet.Server;
using MQTTnet.Server.Disconnecting;
using MqttClient = MQTTnet.Client.MqttClient; using MqttClient = MQTTnet.Client.MqttClient;
namespace MQTTnet namespace MQTTnet
@@ -175,6 +176,16 @@ namespace MQTTnet
return new MqttServer(options, serverAdapters, DefaultLogger); return new MqttServer(options, serverAdapters, DefaultLogger);
} }
public MqttServerClientDisconnectOptionsBuilder CreateMqttServerClientDisconnectOptionsBuilder()
{
return new MqttServerClientDisconnectOptionsBuilder();
}
public MqttServerStopOptionsBuilder CreateMqttServerStopOptionsBuilder()
{
return new MqttServerStopOptionsBuilder();
}
public MqttServerOptionsBuilder CreateServerOptionsBuilder() public MqttServerOptionsBuilder CreateServerOptionsBuilder()
{ {
return new MqttServerOptionsBuilder(); return new MqttServerOptionsBuilder();
@@ -0,0 +1,33 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Collections.Generic;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Server.Disconnecting
{
public sealed class MqttServerClientDisconnectOptions
{
public MqttDisconnectReasonCode ReasonCode { get; set; } = MqttDisconnectReasonCode.NormalDisconnection;
/// <summary>
/// The reason string is sent to every client via a DISCONNECT packet.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public string ReasonString { get; set; }
/// <summary>
/// The server reference is sent to every client via a DISCONNECT packet.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public string ServerReference { get; set; }
/// <summary>
/// These user properties are sent to every client via a DISCONNECT packet.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public List<MqttUserProperty> UserProperties { get; set; }
}
}
@@ -0,0 +1,55 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Collections.Generic;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Server.Disconnecting
{
public sealed class MqttServerClientDisconnectOptionsBuilder
{
readonly MqttServerClientDisconnectOptions _options = new MqttServerClientDisconnectOptions();
public MqttServerClientDisconnectOptions Build()
{
return _options;
}
public MqttServerClientDisconnectOptionsBuilder WithReasonCode(MqttDisconnectReasonCode value)
{
_options.ReasonCode = value;
return this;
}
public MqttServerClientDisconnectOptionsBuilder WithReasonString(string value)
{
_options.ReasonString = value;
return this;
}
public MqttServerClientDisconnectOptionsBuilder WithServerReference(string value)
{
_options.ServerReference = value;
return this;
}
public MqttServerClientDisconnectOptionsBuilder WithUserProperties(List<MqttUserProperty> value)
{
_options.UserProperties = value;
return this;
}
public MqttServerClientDisconnectOptionsBuilder WithUserProperty(string name, string value)
{
if (_options.UserProperties == null)
{
_options.UserProperties = new List<MqttUserProperty>();
}
_options.UserProperties.Add(new MqttUserProperty(name, value));
return this;
}
}
}
+22 -15
View File
@@ -4,6 +4,7 @@
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Adapter; using MQTTnet.Adapter;
@@ -14,6 +15,7 @@ using MQTTnet.Formatter;
using MQTTnet.Internal; using MQTTnet.Internal;
using MQTTnet.Packets; using MQTTnet.Packets;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Server namespace MQTTnet.Server
{ {
@@ -96,7 +98,8 @@ namespace MQTTnet.Server
var cancellationToken = _cancellationToken.Token; var cancellationToken = _cancellationToken.Token;
IsRunning = true; IsRunning = true;
_ = Task.Factory.StartNew(() => SendPacketsLoop(cancellationToken), cancellationToken, TaskCreationOptions.PreferFairness, TaskScheduler.Default).ConfigureAwait(false); _ = Task.Factory.StartNew(() => SendPacketsLoop(cancellationToken), cancellationToken, TaskCreationOptions.PreferFairness, TaskScheduler.Default)
.ConfigureAwait(false);
await ReceivePackagesLoop(cancellationToken).ConfigureAwait(false); await ReceivePackagesLoop(cancellationToken).ConfigureAwait(false);
} }
@@ -141,7 +144,7 @@ namespace MQTTnet.Server
Statistics.HandleSentPacket(packet); Statistics.HandleSentPacket(packet);
} }
public async Task StopAsync(MqttDisconnectReasonCode reason) public async Task StopAsync(MqttServerClientDisconnectOptions disconnectOptions)
{ {
IsRunning = false; IsRunning = false;
@@ -150,16 +153,20 @@ namespace MQTTnet.Server
// Sending DISCONNECT packets from the server to the client is only supported when using MQTTv5+. // Sending DISCONNECT packets from the server to the client is only supported when using MQTTv5+.
if (ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500) if (ChannelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500)
{ {
// The Client or Server MAY send a DISCONNECT packet before closing the Network Connection. // From RFC: The Client or Server MAY send a DISCONNECT packet before closing the Network Connection.
// This library does not sent a DISCONNECT packet for a normal disconnection. Maybe adding // This library does not sent a DISCONNECT packet for a normal disconnection.
// a configuration option is requested in the future. // TODO: Maybe adding a configuration option is requested in the future.
if (reason != MqttDisconnectReasonCode.NormalDisconnection) if (disconnectOptions != null)
{ {
// Is is very important to send the DISCONNECT packet here BEFORE cancelling the if (disconnectOptions.ReasonCode != MqttDisconnectReasonCode.NormalDisconnection || disconnectOptions.UserProperties?.Any() == true ||
// token because the entire connection is closed (disposed) as soon as the cancellation !string.IsNullOrEmpty(disconnectOptions.ReasonString) || !string.IsNullOrEmpty(disconnectOptions.ServerReference))
// token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive {
// at the client! // Is is very important to send the DISCONNECT packet here BEFORE cancelling the
await TrySendDisconnectPacket(reason).ConfigureAwait(false); // token because the entire connection is closed (disposed) as soon as the cancellation
// token is cancelled. To there is no chance that the DISCONNECT packet will ever arrive
// at the client!
await TrySendDisconnectPacket(disconnectOptions).ConfigureAwait(false);
}
} }
} }
} }
@@ -219,7 +226,7 @@ namespace MQTTnet.Server
if (dispatchApplicationMessageResult.CloseConnection) if (dispatchApplicationMessageResult.CloseConnection)
{ {
await StopAsync(MqttDisconnectReasonCode.UnspecifiedError); await StopAsync(new MqttServerClientDisconnectOptions { ReasonCode = MqttDisconnectReasonCode.UnspecifiedError });
return; return;
} }
@@ -550,14 +557,14 @@ namespace MQTTnet.Server
_cancellationToken?.TryCancel(); _cancellationToken?.TryCancel();
} }
async Task TrySendDisconnectPacket(MqttDisconnectReasonCode reasonCode) async Task TrySendDisconnectPacket(MqttServerClientDisconnectOptions options)
{ {
try try
{ {
// This also indicates that it was tried at least! // This also indicates that it was tried at least!
_disconnectPacketSent = true; _disconnectPacketSent = true;
var disconnectPacket = MqttPacketFactories.Disconnect.Create(reasonCode); var disconnectPacket = MqttPacketFactories.Disconnect.Create(options);
using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout)) using (var timeout = new CancellationTokenSource(_serverOptions.DefaultCommunicationTimeout))
{ {
@@ -566,7 +573,7 @@ namespace MQTTnet.Server
} }
catch (Exception exception) catch (Exception exception)
{ {
_logger.Warning(exception, "Client '{0}': Error while sending DISCONNECT packet (ReasonCode = {1})", Id, reasonCode); _logger.Warning(exception, "Client '{0}': Error while sending DISCONNECT packet (ReasonCode = {1})", Id, options.ReasonCode);
} }
} }
} }
@@ -16,6 +16,7 @@ using MQTTnet.Formatter;
using MQTTnet.Internal; using MQTTnet.Internal;
using MQTTnet.Packets; using MQTTnet.Packets;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Server namespace MQTTnet.Server
{ {
@@ -32,11 +33,11 @@ namespace MQTTnet.Server
readonly MqttRetainedMessagesManager _retainedMessagesManager; readonly MqttRetainedMessagesManager _retainedMessagesManager;
readonly IMqttNetLogger _rootLogger; readonly IMqttNetLogger _rootLogger;
readonly ReaderWriterLockSlim _sessionsManagementLock = new ReaderWriterLockSlim();
// The _sessions dictionary contains all session, the _subscriberSessions hash set contains subscriber sessions only. // The _sessions dictionary contains all session, the _subscriberSessions hash set contains subscriber sessions only.
// See the MqttSubscription object for a detailed explanation. // See the MqttSubscription object for a detailed explanation.
readonly MqttSessionsStorage _sessionsStorage = new MqttSessionsStorage(); readonly MqttSessionsStorage _sessionsStorage = new MqttSessionsStorage();
readonly ReaderWriterLockSlim _sessionsManagementLock = new ReaderWriterLockSlim();
readonly HashSet<MqttSession> _subscriberSessions = new HashSet<MqttSession>(); readonly HashSet<MqttSession> _subscriberSessions = new HashSet<MqttSession>();
public MqttClientSessionsManager( public MqttClientSessionsManager(
@@ -58,8 +59,13 @@ namespace MQTTnet.Server
_eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer)); _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));
} }
public async Task CloseAllConnections() public async Task CloseAllConnections(MqttServerClientDisconnectOptions options)
{ {
if (options == null)
{
throw new ArgumentNullException(nameof(options));
}
List<MqttClient> connections; List<MqttClient> connections;
lock (_clients) lock (_clients)
{ {
@@ -69,7 +75,7 @@ namespace MQTTnet.Server
foreach (var connection in connections) foreach (var connection in connections)
{ {
await connection.StopAsync(MqttDisconnectReasonCode.NormalDisconnection).ConfigureAwait(false); await connection.StopAsync(options).ConfigureAwait(false);
} }
} }
@@ -101,7 +107,7 @@ namespace MQTTnet.Server
{ {
if (connection != null) if (connection != null)
{ {
await connection.StopAsync(MqttDisconnectReasonCode.NormalDisconnection).ConfigureAwait(false); await connection.StopAsync(new MqttServerClientDisconnectOptions { ReasonCode = MqttDisconnectReasonCode.NormalDisconnection }).ConfigureAwait(false);
} }
} }
catch (Exception exception) catch (Exception exception)
@@ -366,7 +372,8 @@ namespace MQTTnet.Server
if (_eventContainer.ClientConnectedEvent.HasHandlers) if (_eventContainer.ClientConnectedEvent.HasHandlers)
{ {
var eventArgs = new ClientConnectedEventArgs(connectPacket, var eventArgs = new ClientConnectedEventArgs(
connectPacket,
channelAdapter.PacketFormatterAdapter.ProtocolVersion, channelAdapter.PacketFormatterAdapter.ProtocolVersion,
channelAdapter.Endpoint, channelAdapter.Endpoint,
client.Session.Items); client.Session.Items);
@@ -631,7 +638,8 @@ namespace MQTTnet.Server
if (oldClient != null) if (oldClient != null)
{ {
await oldClient.StopAsync(MqttDisconnectReasonCode.SessionTakenOver).ConfigureAwait(false); // TODO: Consider event here for session takeover to allow manipulation of user properties etc.
await oldClient.StopAsync(new MqttServerClientDisconnectOptions { ReasonCode = MqttDisconnectReasonCode.SessionTakenOver }).ConfigureAwait(false);
if (_eventContainer.ClientDisconnectedEvent.HasHandlers) if (_eventContainer.ClientDisconnectedEvent.HasHandlers)
{ {
@@ -92,7 +92,7 @@ namespace MQTTnet.Server
} }
} }
private struct Statistics struct Statistics
{ {
public long _receivedPacketsCount; public long _receivedPacketsCount;
public long _sentPacketsCount; public long _sentPacketsCount;
@@ -8,11 +8,20 @@ using System.Threading.Tasks;
using MQTTnet.Diagnostics; using MQTTnet.Diagnostics;
using MQTTnet.Internal; using MQTTnet.Internal;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Server namespace MQTTnet.Server
{ {
public sealed class MqttServerKeepAliveMonitor public sealed class MqttServerKeepAliveMonitor
{ {
static readonly MqttServerClientDisconnectOptions DefaultDisconnectOptions = new MqttServerClientDisconnectOptions
{
ReasonCode = MqttDisconnectReasonCode.KeepAliveTimeout,
UserProperties = null,
ReasonString = null,
ServerReference = null
};
readonly MqttNetSourceLogger _logger; readonly MqttNetSourceLogger _logger;
readonly MqttServerOptions _options; readonly MqttServerOptions _options;
readonly MqttClientSessionsManager _sessionsManager; readonly MqttClientSessionsManager _sessionsManager;
@@ -43,7 +52,7 @@ namespace MQTTnet.Server
{ {
try try
{ {
_logger.Info("Starting keep alive monitor."); _logger.Info("Starting keep alive monitor");
while (!cancellationToken.IsCancellationRequested) while (!cancellationToken.IsCancellationRequested)
{ {
@@ -56,11 +65,11 @@ namespace MQTTnet.Server
} }
catch (Exception exception) catch (Exception exception)
{ {
_logger.Error(exception, "Unhandled exception while checking keep alive timeouts."); _logger.Error(exception, "Unhandled exception while checking keep alive timeouts");
} }
finally finally
{ {
_logger.Verbose("Stopped checking keep alive timeout."); _logger.Verbose("Stopped checking keep alive timeout");
} }
} }
@@ -117,18 +126,18 @@ namespace MQTTnet.Server
return; return;
} }
_logger.Warning("Client '{0}': Did not receive any packet or keep alive signal.", connection.Id); _logger.Warning("Client '{0}': Did not receive any packet or keep alive signal", connection.Id);
// Execute the disconnection in background so that the keep alive monitor can continue // Execute the disconnection in background so that the keep alive monitor can continue
// with checking other connections. // with checking other connections.
// We do not need to wait for the task so no await is needed. // We do not need to wait for the task so no await is needed.
// Also the internal state of the connection must be swapped to "Finalizing" because the // Also the internal state of the connection must be swapped to "Finalizing" because the
// next iteration of the keep alive timer happens. // next iteration of the keep alive timer happens.
var _ = connection.StopAsync(MqttDisconnectReasonCode.KeepAliveTimeout); _ = connection.StopAsync(DefaultDisconnectOptions);
} }
catch (Exception exception) catch (Exception exception)
{ {
_logger.Error(exception, "Client {0}: Unhandled exception while checking keep alive timeouts.", connection.Id); _logger.Error(exception, "Client {0}: Unhandled exception while checking keep alive timeouts", connection.Id);
} }
} }
+43 -15
View File
@@ -14,6 +14,7 @@ using MQTTnet.Diagnostics;
using MQTTnet.Internal; using MQTTnet.Internal;
using MQTTnet.Packets; using MQTTnet.Packets;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Server namespace MQTTnet.Server
{ {
@@ -29,6 +30,7 @@ namespace MQTTnet.Server
readonly IMqttNetLogger _rootLogger; readonly IMqttNetLogger _rootLogger;
CancellationTokenSource _cancellationTokenSource; CancellationTokenSource _cancellationTokenSource;
bool _isStopping;
public MqttServer(MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) public MqttServer(MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
{ {
@@ -169,6 +171,13 @@ namespace MQTTnet.Server
remove => _eventContainer.ValidatingConnectionEvent.RemoveHandler(value); remove => _eventContainer.ValidatingConnectionEvent.RemoveHandler(value);
} }
/// <summary>
/// Gets or sets whether the server will accept new connections.
/// If not, the server will close the connection without any notification (DISCONNECT packet).
/// This feature can be used when the server is shutting down.
/// </summary>
public bool AcceptNewConnections { get; set; } = true;
public bool IsStarted => _cancellationTokenSource != null; public bool IsStarted => _cancellationTokenSource != null;
/// <summary> /// <summary>
@@ -184,16 +193,21 @@ namespace MQTTnet.Server
return _retainedMessagesManager?.ClearMessages() ?? CompletedTask.Instance; return _retainedMessagesManager?.ClearMessages() ?? CompletedTask.Instance;
} }
public Task DisconnectClientAsync(string id, MqttDisconnectReasonCode reasonCode) public Task DisconnectClientAsync(string id, MqttServerClientDisconnectOptions options)
{ {
if (id == null) if (id == null)
{ {
throw new ArgumentNullException(nameof(id)); throw new ArgumentNullException(nameof(id));
} }
if (options == null)
{
throw new ArgumentNullException(nameof(options));
}
ThrowIfNotStarted(); ThrowIfNotStarted();
return _clientSessionsManager.GetClient(id).StopAsync(reasonCode); return _clientSessionsManager.GetClient(id).StopAsync(options);
} }
public Task<IList<MqttClientStatus>> GetClientsAsync() public Task<IList<MqttClientStatus>> GetClientsAsync()
@@ -203,13 +217,6 @@ namespace MQTTnet.Server
return _clientSessionsManager.GetClientsStatus(); return _clientSessionsManager.GetClientsStatus();
} }
public Task<IList<MqttApplicationMessage>> GetRetainedMessagesAsync()
{
ThrowIfNotStarted();
return _retainedMessagesManager.GetMessages();
}
public Task<MqttApplicationMessage> GetRetainedMessageAsync(string topic) public Task<MqttApplicationMessage> GetRetainedMessageAsync(string topic)
{ {
if (topic == null) if (topic == null)
@@ -222,6 +229,13 @@ namespace MQTTnet.Server
return _retainedMessagesManager.GetMessage(topic); return _retainedMessagesManager.GetMessage(topic);
} }
public Task<IList<MqttApplicationMessage>> GetRetainedMessagesAsync()
{
ThrowIfNotStarted();
return _retainedMessagesManager.GetMessages();
}
public Task<IList<MqttSessionStatus>> GetSessionsAsync() public Task<IList<MqttSessionStatus>> GetSessionsAsync()
{ {
ThrowIfNotStarted(); ThrowIfNotStarted();
@@ -247,7 +261,7 @@ namespace MQTTnet.Server
if (string.IsNullOrEmpty(injectedApplicationMessage.ApplicationMessage.Topic)) if (string.IsNullOrEmpty(injectedApplicationMessage.ApplicationMessage.Topic))
{ {
throw new NotSupportedException("Injected application messages must contain a topic. Topic alias is not supported."); throw new NotSupportedException("Injected application messages must contain a topic (topic alias is not supported)");
} }
var sessionItems = injectedApplicationMessage.CustomSessionItems ?? ServerSessionItems; var sessionItems = injectedApplicationMessage.CustomSessionItems ?? ServerSessionItems;
@@ -263,6 +277,8 @@ namespace MQTTnet.Server
{ {
ThrowIfStarted(); ThrowIfStarted();
_isStopping = false;
_cancellationTokenSource = new CancellationTokenSource(); _cancellationTokenSource = new CancellationTokenSource();
var cancellationToken = _cancellationTokenSource.Token; var cancellationToken = _cancellationTokenSource.Token;
@@ -278,11 +294,16 @@ namespace MQTTnet.Server
await _eventContainer.StartedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false); await _eventContainer.StartedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
_logger.Info("Started."); _logger.Info("Started");
} }
public async Task StopAsync() public async Task StopAsync(MqttServerStopOptions options)
{ {
if (options == null)
{
throw new ArgumentNullException(nameof(options));
}
try try
{ {
if (_cancellationTokenSource == null) if (_cancellationTokenSource == null)
@@ -290,9 +311,11 @@ namespace MQTTnet.Server
return; return;
} }
_isStopping = true;
_cancellationTokenSource.Cancel(false); _cancellationTokenSource.Cancel(false);
await _clientSessionsManager.CloseAllConnections().ConfigureAwait(false); await _clientSessionsManager.CloseAllConnections(options.DefaultClientDisconnectOptions).ConfigureAwait(false);
foreach (var adapter in _adapters) foreach (var adapter in _adapters)
{ {
@@ -308,7 +331,7 @@ namespace MQTTnet.Server
await _eventContainer.StoppedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false); await _eventContainer.StoppedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
_logger.Info("Stopped."); _logger.Info("Stopped");
} }
public Task SubscribeAsync(string clientId, ICollection<MqttTopicFilter> topicFilters) public Task SubscribeAsync(string clientId, ICollection<MqttTopicFilter> topicFilters)
@@ -369,7 +392,7 @@ namespace MQTTnet.Server
{ {
if (disposing) if (disposing)
{ {
StopAsync().GetAwaiter().GetResult(); StopAsync(new MqttServerStopOptions()).GetAwaiter().GetResult();
foreach (var adapter in _adapters) foreach (var adapter in _adapters)
{ {
@@ -382,6 +405,11 @@ namespace MQTTnet.Server
Task OnHandleClient(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken) Task OnHandleClient(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
{ {
if (_isStopping || !AcceptNewConnections)
{
return CompletedTask.Instance;
}
return _clientSessionsManager.HandleClientConnectionAsync(channelAdapter, cancellationToken); return _clientSessionsManager.HandleClientConnectionAsync(channelAdapter, cancellationToken);
} }
@@ -8,11 +8,22 @@ using System.Threading.Tasks;
using MQTTnet.Internal; using MQTTnet.Internal;
using MQTTnet.Packets; using MQTTnet.Packets;
using MQTTnet.Protocol; using MQTTnet.Protocol;
using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Server namespace MQTTnet.Server
{ {
public static class MqttServerExtensions public static class MqttServerExtensions
{ {
public static Task DisconnectClientAsync(this MqttServer server, string id, MqttDisconnectReasonCode reasonCode = MqttDisconnectReasonCode.NormalDisconnection)
{
if (server == null)
{
throw new ArgumentNullException(nameof(server));
}
return server.DisconnectClientAsync(id, new MqttServerClientDisconnectOptions { ReasonCode = reasonCode });
}
public static Task InjectApplicationMessage( public static Task InjectApplicationMessage(
this MqttServer server, this MqttServer server,
string topic, string topic,
@@ -47,6 +58,16 @@ namespace MQTTnet.Server
})); }));
} }
public static Task StopAsync(this MqttServer server)
{
if (server == null)
{
throw new ArgumentNullException(nameof(server));
}
return server.StopAsync(new MqttServerStopOptions());
}
public static Task SubscribeAsync(this MqttServer server, string clientId, params MqttTopicFilter[] topicFilters) public static Task SubscribeAsync(this MqttServer server, string clientId, params MqttTopicFilter[] topicFilters)
{ {
if (server == null) if (server == null)
@@ -5,7 +5,7 @@
using System; using System;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Formatter; using MQTTnet.Formatter;
using MQTTnet.Protocol; using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Server namespace MQTTnet.Server
{ {
@@ -18,41 +18,46 @@ namespace MQTTnet.Server
_client = client ?? throw new ArgumentNullException(nameof(client)); _client = client ?? throw new ArgumentNullException(nameof(client));
} }
/// <summary> public long BytesReceived => _client.ChannelAdapter.BytesReceived;
/// Gets or sets the client identifier.
/// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. public long BytesSent => _client.ChannelAdapter.BytesSent;
/// </summary>
public string Id => _client.Id; public DateTime ConnectedTimestamp => _client.Statistics.ConnectedTimestamp;
public string Endpoint => _client.Endpoint; public string Endpoint => _client.Endpoint;
public MqttProtocolVersion ProtocolVersion => _client.ChannelAdapter.PacketFormatterAdapter.ProtocolVersion; /// <summary>
/// Gets or sets the client identifier.
/// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
/// </summary>
public string Id => _client.Id;
public DateTime ConnectedTimestamp => _client.Statistics.ConnectedTimestamp; public DateTime LastNonKeepAlivePacketReceivedTimestamp => _client.Statistics.LastNonKeepAlivePacketReceivedTimestamp;
public DateTime LastPacketReceivedTimestamp => _client.Statistics.LastPacketReceivedTimestamp; public DateTime LastPacketReceivedTimestamp => _client.Statistics.LastPacketReceivedTimestamp;
public DateTime LastPacketSentTimestamp => _client.Statistics.LastPacketSentTimestamp; public DateTime LastPacketSentTimestamp => _client.Statistics.LastPacketSentTimestamp;
public DateTime LastNonKeepAlivePacketReceivedTimestamp => _client.Statistics.LastNonKeepAlivePacketReceivedTimestamp; public MqttProtocolVersion ProtocolVersion => _client.ChannelAdapter.PacketFormatterAdapter.ProtocolVersion;
public long ReceivedApplicationMessagesCount => _client.Statistics.ReceivedApplicationMessagesCount; public long ReceivedApplicationMessagesCount => _client.Statistics.ReceivedApplicationMessagesCount;
public long SentApplicationMessagesCount => _client.Statistics.SentApplicationMessagesCount;
public long ReceivedPacketsCount => _client.Statistics.ReceivedPacketsCount; public long ReceivedPacketsCount => _client.Statistics.ReceivedPacketsCount;
public long SentApplicationMessagesCount => _client.Statistics.SentApplicationMessagesCount;
public long SentPacketsCount => _client.Statistics.SentPacketsCount; public long SentPacketsCount => _client.Statistics.SentPacketsCount;
public MqttSessionStatus Session { get; set; } public MqttSessionStatus Session { get; set; }
public long BytesSent => _client.ChannelAdapter.BytesSent; public Task DisconnectAsync(MqttServerClientDisconnectOptions options)
public long BytesReceived => _client.ChannelAdapter.BytesReceived;
public Task DisconnectAsync()
{ {
return _client.StopAsync(MqttDisconnectReasonCode.NormalDisconnection); if (options == null)
{
throw new ArgumentNullException(nameof(options));
}
return _client.StopAsync(options);
} }
public void ResetStatistics() public void ResetStatistics()
@@ -0,0 +1,32 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using System.Threading.Tasks;
using MQTTnet.Protocol;
using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Server
{
public static class MqttClientStatusExtensions
{
static readonly MqttServerClientDisconnectOptions DefaultDisconnectOptions = new MqttServerClientDisconnectOptions
{
ReasonCode = MqttDisconnectReasonCode.NormalDisconnection,
ReasonString = null,
UserProperties = null,
ServerReference = null
};
public static Task DisconnectAsync(this MqttClientStatus clientStatus)
{
if (clientStatus == null)
{
throw new ArgumentNullException(nameof(clientStatus));
}
return clientStatus.DisconnectAsync(DefaultDisconnectOptions);
}
}
}
@@ -0,0 +1,24 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using MQTTnet.Protocol;
using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Server
{
public sealed class MqttServerStopOptions
{
/// <summary>
/// These disconnect options are sent to every connected client via a DISCONNECT packet.
/// <remarks>MQTT 5.0.0+ feature.</remarks>
/// </summary>
public MqttServerClientDisconnectOptions DefaultClientDisconnectOptions { get; set; } = new MqttServerClientDisconnectOptions
{
ReasonCode = MqttDisconnectReasonCode.ServerShuttingDown,
UserProperties = null,
ReasonString = null,
ServerReference = null
};
}
}
@@ -0,0 +1,39 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System;
using MQTTnet.Server.Disconnecting;
namespace MQTTnet.Server
{
public sealed class MqttServerStopOptionsBuilder
{
readonly MqttServerStopOptions _options = new MqttServerStopOptions();
public MqttServerStopOptionsBuilder WithDefaultClientDisconnectOptions(MqttServerClientDisconnectOptions value)
{
_options.DefaultClientDisconnectOptions = value;
return this;
}
public MqttServerStopOptionsBuilder WithDefaultClientDisconnectOptions(Action<MqttServerClientDisconnectOptionsBuilder> builder)
{
if (builder == null)
{
throw new ArgumentNullException(nameof(builder));
}
var optionsBuilder = new MqttServerClientDisconnectOptionsBuilder();
builder.Invoke(optionsBuilder);
_options.DefaultClientDisconnectOptions = optionsBuilder.Build();
return this;
}
public MqttServerStopOptions Build()
{
return _options;
}
}
}