Add interface for server options. Add context class for connection validator (hide internal packets).

This commit is contained in:
Christian Kratky
2017-11-25 20:12:50 +01:00
parent f9d67e2cd2
commit ee6ec14bf8
20 changed files with 74 additions and 50 deletions
+3 -3
View File
@@ -2,7 +2,7 @@
<package > <package >
<metadata> <metadata>
<id>MQTTnet.AspNetCore</id> <id>MQTTnet.AspNetCore</id>
<version>2.5.3</version> <version>2.6.0</version>
<authors>Christian Kratky</authors> <authors>Christian Kratky</authors>
<owners>Christian Kratky</owners> <owners>Christian Kratky</owners>
<licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl> <licenseUrl>https://github.com/chkr1011/MQTTnet/blob/master/LICENSE</licenseUrl>
@@ -10,13 +10,13 @@
<iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl> <iconUrl>https://raw.githubusercontent.com/chkr1011/MQTTnet/master/Images/Logo_128x128.png</iconUrl>
<requireLicenseAcceptance>false</requireLicenseAcceptance> <requireLicenseAcceptance>false</requireLicenseAcceptance>
<description>This is a support library to integrate MQTTnet into AspNetCore.</description> <description>This is a support library to integrate MQTTnet into AspNetCore.</description>
<releaseNotes>* Updated to MQTTnet 2.5.3. <releaseNotes>* Updated to MQTTnet 2.6.0.
</releaseNotes> </releaseNotes>
<copyright>Copyright Christian Kratky 2016-2017</copyright> <copyright>Copyright Christian Kratky 2016-2017</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>
<dependencies> <dependencies>
<group targetFramework="netstandard2.0"> <group targetFramework="netstandard2.0">
<dependency id="MQTTnet" version="2.5.3" /> <dependency id="MQTTnet" version="2.6.0" />
</group> </group>
</dependencies> </dependencies>
</metadata> </metadata>
@@ -12,7 +12,7 @@ namespace MQTTnet.AspNetCore
{ {
public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted; public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
public Task StartAsync(MqttServerOptions options) public Task StartAsync(IMqttServerOptions options)
{ {
return Task.CompletedTask; return Task.CompletedTask;
} }
@@ -9,7 +9,7 @@ namespace MQTTnet.AspNetCore
{ {
public static class ServiceCollectionExtensions public static class ServiceCollectionExtensions
{ {
public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, MqttServerOptions options) public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, IMqttServerOptions options)
{ {
if (options == null) throw new ArgumentNullException(nameof(options)); if (options == null) throw new ArgumentNullException(nameof(options));
@@ -8,7 +8,7 @@ namespace MQTTnet.Adapter
{ {
event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted; event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
Task StartAsync(MqttServerOptions options); Task StartAsync(IMqttServerOptions options);
Task StopAsync(); Task StopAsync();
} }
} }
@@ -21,7 +21,7 @@ namespace MQTTnet.Implementations
public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted; public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
public async Task StartAsync(MqttServerOptions options) public async Task StartAsync(IMqttServerOptions options)
{ {
if (options == null) throw new ArgumentNullException(nameof(options)); if (options == null) throw new ArgumentNullException(nameof(options));
@@ -30,7 +30,7 @@ namespace MQTTnet.Implementations
public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted; public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
public Task StartAsync(MqttServerOptions options) public Task StartAsync(IMqttServerOptions options)
{ {
if (_cancellationTokenSource != null) throw new InvalidOperationException("Server is already started."); if (_cancellationTokenSource != null) throw new InvalidOperationException("Server is already started.");
@@ -12,7 +12,7 @@ namespace MQTTnet.Server
Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync(); Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync();
Task StartAsync(MqttServerOptions options); Task StartAsync(IMqttServerOptions options);
Task StopAsync(); Task StopAsync();
} }
} }
@@ -0,0 +1,16 @@
using System;
namespace MQTTnet.Server
{
public interface IMqttServerOptions
{
Action<MqttApplicationMessageInterceptorContext> ApplicationMessageInterceptor { get; }
int ConnectionBacklog { get; }
Action<MqttConnectionValidatorContext> ConnectionValidator { get; }
TimeSpan DefaultCommunicationTimeout { get; }
MqttServerDefaultEndpointOptions DefaultEndpointOptions { get; }
IMqttServerStorage Storage { get; }
Action<MqttSubscriptionInterceptorContext> SubscriptionInterceptor { get; }
MqttServerTlsEndpointOptions TlsEndpointOptions { get; }
}
}
@@ -13,11 +13,11 @@ namespace MQTTnet.Server
public sealed class MqttClientPendingMessagesQueue public sealed class MqttClientPendingMessagesQueue
{ {
private readonly BlockingCollection<MqttPublishPacket> _pendingPublishPackets = new BlockingCollection<MqttPublishPacket>(); private readonly BlockingCollection<MqttPublishPacket> _pendingPublishPackets = new BlockingCollection<MqttPublishPacket>();
private readonly MqttServerOptions _options; private readonly IMqttServerOptions _options;
private readonly MqttClientSession _session; private readonly MqttClientSession _session;
private readonly IMqttNetLogger _logger; private readonly IMqttNetLogger _logger;
public MqttClientPendingMessagesQueue(MqttServerOptions options, MqttClientSession session, IMqttNetLogger logger) public MqttClientPendingMessagesQueue(IMqttServerOptions options, MqttClientSession session, IMqttNetLogger logger)
{ {
_logger = logger ?? throw new ArgumentNullException(nameof(logger)); _logger = logger ?? throw new ArgumentNullException(nameof(logger));
_session = session ?? throw new ArgumentNullException(nameof(session)); _session = session ?? throw new ArgumentNullException(nameof(session));
@@ -19,7 +19,7 @@ namespace MQTTnet.Server
private readonly MqttClientSubscriptionsManager _subscriptionsManager; private readonly MqttClientSubscriptionsManager _subscriptionsManager;
private readonly MqttClientSessionsManager _sessionsManager; private readonly MqttClientSessionsManager _sessionsManager;
private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue; private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue;
private readonly MqttServerOptions _options; private readonly IMqttServerOptions _options;
private readonly IMqttNetLogger _logger; private readonly IMqttNetLogger _logger;
private IMqttChannelAdapter _adapter; private IMqttChannelAdapter _adapter;
@@ -28,7 +28,7 @@ namespace MQTTnet.Server
public MqttClientSession( public MqttClientSession(
string clientId, string clientId,
MqttServerOptions options, IMqttServerOptions options,
MqttClientSessionsManager sessionsManager, MqttClientSessionsManager sessionsManager,
IMqttNetLogger logger) IMqttNetLogger logger)
{ {
@@ -18,11 +18,11 @@ namespace MQTTnet.Server
private readonly Dictionary<string, MqttClientSession> _sessions = new Dictionary<string, MqttClientSession>(); private readonly Dictionary<string, MqttClientSession> _sessions = new Dictionary<string, MqttClientSession>();
private readonly SemaphoreSlim _sessionsSemaphore = new SemaphoreSlim(1, 1); private readonly SemaphoreSlim _sessionsSemaphore = new SemaphoreSlim(1, 1);
private readonly MqttServerOptions _options; private readonly IMqttServerOptions _options;
private readonly MqttRetainedMessagesManager _retainedMessagesManager; private readonly MqttRetainedMessagesManager _retainedMessagesManager;
private readonly IMqttNetLogger _logger; private readonly IMqttNetLogger _logger;
public MqttClientSessionsManager(MqttServerOptions options, MqttRetainedMessagesManager retainedMessagesManager, IMqttNetLogger logger) public MqttClientSessionsManager(IMqttServerOptions options, MqttRetainedMessagesManager retainedMessagesManager, IMqttNetLogger logger)
{ {
_logger = logger ?? throw new ArgumentNullException(nameof(logger)); _logger = logger ?? throw new ArgumentNullException(nameof(logger));
_options = options ?? throw new ArgumentNullException(nameof(options)); _options = options ?? throw new ArgumentNullException(nameof(options));
@@ -174,12 +174,19 @@ namespace MQTTnet.Server
private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket) private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket)
{ {
if (_options.ConnectionValidator != null) if (_options.ConnectionValidator == null)
{ {
return _options.ConnectionValidator(connectPacket); return MqttConnectReturnCode.ConnectionAccepted;
} }
var context = new MqttConnectionValidatorContext(
connectPacket.ClientId,
connectPacket.Username,
connectPacket.Password,
connectPacket.WillMessage);
return MqttConnectReturnCode.ConnectionAccepted; _options.ConnectionValidator(context);
return context.ReturnCode;
} }
private async Task<GetOrCreateClientSessionResult> GetOrCreateClientSessionAsync(MqttConnectPacket connectPacket) private async Task<GetOrCreateClientSessionResult> GetOrCreateClientSessionAsync(MqttConnectPacket connectPacket)
@@ -8,9 +8,9 @@ namespace MQTTnet.Server
public sealed class MqttClientSubscriptionsManager public sealed class MqttClientSubscriptionsManager
{ {
private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>(); private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
private readonly MqttServerOptions _options; private readonly IMqttServerOptions _options;
public MqttClientSubscriptionsManager(MqttServerOptions options) public MqttClientSubscriptionsManager(IMqttServerOptions options)
{ {
_options = options ?? throw new ArgumentNullException(nameof(options)); _options = options ?? throw new ArgumentNullException(nameof(options));
} }
@@ -13,9 +13,9 @@ namespace MQTTnet.Server
private readonly Dictionary<string, MqttApplicationMessage> _retainedMessages = new Dictionary<string, MqttApplicationMessage>(); private readonly Dictionary<string, MqttApplicationMessage> _retainedMessages = new Dictionary<string, MqttApplicationMessage>();
private readonly SemaphoreSlim _gate = new SemaphoreSlim(1, 1); private readonly SemaphoreSlim _gate = new SemaphoreSlim(1, 1);
private readonly IMqttNetLogger _logger; private readonly IMqttNetLogger _logger;
private readonly MqttServerOptions _options; private readonly IMqttServerOptions _options;
public MqttRetainedMessagesManager(MqttServerOptions options, IMqttNetLogger logger) public MqttRetainedMessagesManager(IMqttServerOptions options, IMqttNetLogger logger)
{ {
_logger = logger ?? throw new ArgumentNullException(nameof(logger)); _logger = logger ?? throw new ArgumentNullException(nameof(logger));
_options = options ?? throw new ArgumentNullException(nameof(options)); _options = options ?? throw new ArgumentNullException(nameof(options));
@@ -16,7 +16,7 @@ namespace MQTTnet.Server
private MqttClientSessionsManager _clientSessionsManager; private MqttClientSessionsManager _clientSessionsManager;
private MqttRetainedMessagesManager _retainedMessagesManager; private MqttRetainedMessagesManager _retainedMessagesManager;
private CancellationTokenSource _cancellationTokenSource; private CancellationTokenSource _cancellationTokenSource;
private MqttServerOptions _options; private IMqttServerOptions _options;
public MqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger) public MqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
{ {
@@ -52,7 +52,7 @@ namespace MQTTnet.Server
} }
} }
public async Task StartAsync(MqttServerOptions options) public async Task StartAsync(IMqttServerOptions options)
{ {
_options = options ?? throw new ArgumentNullException(nameof(options)); _options = options ?? throw new ArgumentNullException(nameof(options));
@@ -1,10 +1,8 @@
using System; using System;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Server namespace MQTTnet.Server
{ {
public class MqttServerOptions public class MqttServerOptions : IMqttServerOptions
{ {
public MqttServerDefaultEndpointOptions DefaultEndpointOptions { get; } = new MqttServerDefaultEndpointOptions(); public MqttServerDefaultEndpointOptions DefaultEndpointOptions { get; } = new MqttServerDefaultEndpointOptions();
@@ -14,7 +12,7 @@ namespace MQTTnet.Server
public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(15); public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(15);
public Func<MqttConnectPacket, MqttConnectReturnCode> ConnectionValidator { get; set; } public Action<MqttConnectionValidatorContext> ConnectionValidator { get; set; }
public Action<MqttApplicationMessageInterceptorContext> ApplicationMessageInterceptor { get; set; } public Action<MqttApplicationMessageInterceptorContext> ApplicationMessageInterceptor { get; set; }
@@ -1,6 +1,4 @@
using System; using System;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Server namespace MQTTnet.Server
{ {
@@ -62,7 +60,7 @@ namespace MQTTnet.Server
return this; return this;
} }
public MqttServerOptionsBuilder WithConnectionValidator(Func<MqttConnectPacket, MqttConnectReturnCode> value) public MqttServerOptionsBuilder WithConnectionValidator(Action<MqttConnectionValidatorContext> value)
{ {
_options.ConnectionValidator = value; _options.ConnectionValidator = value;
return this; return this;
@@ -80,7 +78,7 @@ namespace MQTTnet.Server
return this; return this;
} }
public MqttServerOptions Build() public IMqttServerOptions Build()
{ {
return _options; return _options;
} }
@@ -4,7 +4,7 @@ namespace MQTTnet.Server
{ {
public static class MqttServerOptionsExtensions public static class MqttServerOptionsExtensions
{ {
public static int GetTlsEndpointPort(this MqttServerOptions options) public static int GetTlsEndpointPort(this IMqttServerOptions options)
{ {
if (options == null) throw new ArgumentNullException(nameof(options)); if (options == null) throw new ArgumentNullException(nameof(options));
@@ -16,7 +16,7 @@ namespace MQTTnet.Server
return options.TlsEndpointOptions.Port.Value; return options.TlsEndpointOptions.Port.Value;
} }
public static int GetDefaultEndpointPort(this MqttServerOptions options) public static int GetDefaultEndpointPort(this IMqttServerOptions options)
{ {
if (options == null) throw new ArgumentNullException(nameof(options)); if (options == null) throw new ArgumentNullException(nameof(options));
@@ -62,7 +62,7 @@ namespace MQTTnet.Core.Tests
ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(adapter)); ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(adapter));
} }
public Task StartAsync(MqttServerOptions options) public Task StartAsync(IMqttServerOptions options)
{ {
return Task.FromResult(0); return Task.FromResult(0);
} }
+1 -3
View File
@@ -23,11 +23,9 @@ namespace MQTTnet.TestApp.NetCore
{ {
if (p.Username != "USER" || p.Password != "PASS") if (p.Username != "USER" || p.Password != "PASS")
{ {
return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; p.ReturnCode = MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
} }
} }
return MqttConnectReturnCode.ConnectionAccepted;
}, },
Storage = new RetainedMessageHandler(), Storage = new RetainedMessageHandler(),
@@ -354,20 +354,23 @@ namespace MQTTnet.TestApp.UniversalWindows
{ {
if (c.ClientId.Length < 10) if (c.ClientId.Length < 10)
{ {
return MqttConnectReturnCode.ConnectionRefusedIdentifierRejected; c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedIdentifierRejected;
return;
} }
if (c.Username != "mySecretUser") if (c.Username != "mySecretUser")
{ {
return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
return;
} }
if (c.Password != "mySecretPassword") if (c.Password != "mySecretPassword")
{ {
return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
return;
} }
return MqttConnectReturnCode.ConnectionAccepted; c.ReturnCode = MqttConnectReturnCode.ConnectionAccepted;
}; };
var factory = new MqttFactory(); var factory = new MqttFactory();
@@ -410,14 +413,15 @@ namespace MQTTnet.TestApp.UniversalWindows
}; };
options.DefaultEndpointOptions.Port = 1884; options.DefaultEndpointOptions.Port = 1884;
options.ConnectionValidator = packet => options.ConnectionValidator = c =>
{ {
if (packet.ClientId != "Highlander") if (c.ClientId != "Highlander")
{ {
return MqttConnectReturnCode.ConnectionRefusedIdentifierRejected; c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedIdentifierRejected;
return;
} }
return MqttConnectReturnCode.ConnectionAccepted; c.ReturnCode = MqttConnectReturnCode.ConnectionAccepted;
}; };
var mqttServer = new MqttFactory().CreateMqttServer(); var mqttServer = new MqttFactory().CreateMqttServer();
@@ -432,20 +436,23 @@ namespace MQTTnet.TestApp.UniversalWindows
{ {
if (c.ClientId.Length < 10) if (c.ClientId.Length < 10)
{ {
return MqttConnectReturnCode.ConnectionRefusedIdentifierRejected; c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedIdentifierRejected;
return;
} }
if (c.Username != "mySecretUser") if (c.Username != "mySecretUser")
{ {
return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
return;
} }
if (c.Password != "mySecretPassword") if (c.Password != "mySecretPassword")
{ {
return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
return;
} }
return MqttConnectReturnCode.ConnectionAccepted; c.ReturnCode = MqttConnectReturnCode.ConnectionAccepted;
} }
}; };
} }