Merge branch 'feature/server-project-bootstrap' into develop

This commit is contained in:
Christian Kratky
2019-03-10 19:57:32 +01:00
127 changed files with 2952 additions and 490 deletions
@@ -10,6 +10,8 @@ namespace MQTTnet.Adapter
{
string Endpoint { get; }
bool IsSecureConnection { get; }
MqttPacketFormatterAdapter PacketFormatterAdapter { get; }
event EventHandler ReadingPacketStarted;
+1 -1
View File
@@ -6,7 +6,7 @@ namespace MQTTnet.Adapter
{
public interface IMqttServerAdapter : IDisposable
{
event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
Action<MqttServerAdapterClientAcceptedEventArgs> ClientAcceptedHandler { get; set; }
Task StartAsync(IMqttServerOptions options);
Task StopAsync();
@@ -42,6 +42,8 @@ namespace MQTTnet.Adapter
public string Endpoint => _channel.Endpoint;
public bool IsSecureConnection => _channel.IsSecureConnection;
public MqttPacketFormatterAdapter PacketFormatterAdapter { get; }
public event EventHandler ReadingPacketStarted;
+1
View File
@@ -7,6 +7,7 @@ namespace MQTTnet.Channel
public interface IMqttChannel : IDisposable
{
string Endpoint { get; }
bool IsSecureConnection { get; }
Task ConnectAsync(CancellationToken cancellationToken);
Task DisconnectAsync(CancellationToken cancellationToken);
+2 -6
View File
@@ -15,13 +15,9 @@ namespace MQTTnet.Client
IMqttClientOptions Options { get; }
IMqttClientConnectedHandler ConnectedHandler { get; set; }
[Obsolete("Use ConnectedHandler instead.")]
event EventHandler<MqttClientConnectedEventArgs> Connected;
IMqttClientDisconnectedHandler DisconnectedHandler { get; set; }
[Obsolete("Use DisconnectedHandler instead.")]
event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
Task<MqttClientAuthenticateResult> ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken);
Task DisconnectAsync(MqttClientDisconnectOptions options, CancellationToken cancellationToken);
+22 -32
View File
@@ -46,15 +46,9 @@ namespace MQTTnet.Client
public IMqttClientConnectedHandler ConnectedHandler { get; set; }
public event EventHandler<MqttClientConnectedEventArgs> Connected;
public IMqttClientDisconnectedHandler DisconnectedHandler { get; set; }
public event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
public IMqttApplicationMessageHandler ReceivedApplicationMessageHandler { get; set; }
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get; set; }
public bool IsConnected { get; private set; }
@@ -106,9 +100,6 @@ namespace MQTTnet.Client
await connectedHandler.HandleConnectedAsync(new MqttClientConnectedEventArgs(authenticateResult)).ConfigureAwait(false);
}
// TODO: Remove!
Connected?.Invoke(this, new MqttClientConnectedEventArgs(authenticateResult));
return authenticateResult;
}
catch (Exception exception)
@@ -276,8 +267,6 @@ namespace MQTTnet.Client
{
await disconnectedHandler.HandleDisconnectedAsync(new MqttClientDisconnectedEventArgs(clientWasConnected, exception)).ConfigureAwait(false);
}
Disconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(clientWasConnected, exception));
}
}
@@ -521,25 +510,27 @@ namespace MQTTnet.Client
}
else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
{
await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
await SendAsync(new MqttPubAckPacket
if (await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false))
{
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = MqttPubAckReasonCode.Success
}, cancellationToken).ConfigureAwait(false);
await SendAsync(new MqttPubAckPacket
{
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = MqttPubAckReasonCode.Success
}, cancellationToken).ConfigureAwait(false);
}
}
else if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
{
await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false);
var pubRecPacket = new MqttPubRecPacket
if (await HandleReceivedApplicationMessageAsync(publishPacket).ConfigureAwait(false))
{
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = MqttPubRecReasonCode.Success
};
var pubRecPacket = new MqttPubRecPacket
{
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = MqttPubRecReasonCode.Success
};
await SendAsync(pubRecPacket, cancellationToken).ConfigureAwait(false);
await SendAsync(pubRecPacket, cancellationToken).ConfigureAwait(false);
}
}
else
{
@@ -583,20 +574,19 @@ namespace MQTTnet.Client
return _adapter.PacketFormatterAdapter.DataConverter.CreatePublishResult(pubRecPacket, pubCompPacket);
}
private Task HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket)
private async Task<bool> HandleReceivedApplicationMessageAsync(MqttPublishPacket publishPacket)
{
var applicationMessage = _adapter.PacketFormatterAdapter.DataConverter.CreateApplicationMessage(publishPacket);
ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage));
var handler = ReceivedApplicationMessageHandler;
var handler = ApplicationMessageReceivedHandler;
if (handler != null)
{
return handler.HandleApplicationMessageAsync(
new MqttApplicationMessageHandlerContext(Options.ClientId, applicationMessage));
var eventArgs = new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage);
await handler.HandleApplicationMessageReceivedAsync(eventArgs).ConfigureAwait(false);
return !eventArgs.ProcessingFailed;
}
return Task.FromResult(0);
return true;
}
private static async Task WaitForTaskAsync(Task task, Task sender)
@@ -63,35 +63,35 @@ namespace MQTTnet.Client
return client;
}
public static IMqttClient UseReceivedApplicationMessageHandler(this IMqttClient client, Func<MqttApplicationMessageHandlerContext, Task> handler)
public static IMqttClient UseApplicationMessageReceivedHandler(this IMqttClient client, Func<MqttApplicationMessageReceivedEventArgs, Task> handler)
{
if (handler == null)
{
client.ReceivedApplicationMessageHandler = null;
client.ApplicationMessageReceivedHandler = null;
return client;
}
client.ReceivedApplicationMessageHandler = new MqttApplicationMessageHandlerDelegate(handler);
client.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(handler);
return client;
}
public static IMqttClient UseReceivedApplicationMessageHandler(this IMqttClient client, Action<MqttApplicationMessageHandlerContext> handler)
public static IMqttClient UseApplicationMessageReceivedHandler(this IMqttClient client, Action<MqttApplicationMessageReceivedEventArgs> handler)
{
if (handler == null)
{
client.ReceivedApplicationMessageHandler = null;
client.ApplicationMessageReceivedHandler = null;
return client;
}
client.ReceivedApplicationMessageHandler = new MqttApplicationMessageHandlerDelegate(handler);
client.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(handler);
return client;
}
public static IMqttClient UseReceivedApplicationMessageHandler(this IMqttClient client, IMqttApplicationMessageHandler handler)
public static IMqttClient UseApplicationMessageReceivedHandler(this IMqttClient client, IMqttApplicationMessageReceivedHandler handler)
{
client.ReceivedApplicationMessageHandler = handler;
client.ApplicationMessageReceivedHandler = handler;
return client;
}
@@ -1,9 +0,0 @@
using System.Threading.Tasks;
namespace MQTTnet.Client.Receiving
{
public interface IMqttApplicationMessageHandler
{
Task HandleApplicationMessageAsync(MqttApplicationMessageHandlerContext context);
}
}
@@ -0,0 +1,9 @@
using System.Threading.Tasks;
namespace MQTTnet.Client.Receiving
{
public interface IMqttApplicationMessageReceivedHandler
{
Task HandleApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs);
}
}
@@ -1,15 +0,0 @@
namespace MQTTnet.Client.Receiving
{
public class MqttApplicationMessageHandlerContext
{
public MqttApplicationMessageHandlerContext(string senderClientId, MqttApplicationMessage applicationMessage)
{
SenderClientId = senderClientId;
ApplicationMessage = applicationMessage;
}
public string SenderClientId { get; }
public MqttApplicationMessage ApplicationMessage { get; }
}
}
@@ -1,31 +0,0 @@
using System;
using System.Threading.Tasks;
namespace MQTTnet.Client.Receiving
{
public class MqttApplicationMessageHandlerDelegate : IMqttApplicationMessageHandler
{
private readonly Func<MqttApplicationMessageHandlerContext, Task> _handler;
public MqttApplicationMessageHandlerDelegate(Action<MqttApplicationMessageHandlerContext> handler)
{
if (handler == null) throw new ArgumentNullException(nameof(handler));
_handler = context =>
{
handler(context);
return Task.FromResult(0);
};
}
public MqttApplicationMessageHandlerDelegate(Func<MqttApplicationMessageHandlerContext, Task> handler)
{
_handler = handler ?? throw new ArgumentNullException(nameof(handler));
}
public Task HandleApplicationMessageAsync(MqttApplicationMessageHandlerContext context)
{
return _handler(context);
}
}
}
@@ -0,0 +1,31 @@
using System;
using System.Threading.Tasks;
namespace MQTTnet.Client.Receiving
{
public class MqttApplicationMessageReceivedHandlerDelegate : IMqttApplicationMessageReceivedHandler
{
private readonly Func<MqttApplicationMessageReceivedEventArgs, Task> _handler;
public MqttApplicationMessageReceivedHandlerDelegate(Action<MqttApplicationMessageReceivedEventArgs> handler)
{
if (handler == null) throw new ArgumentNullException(nameof(handler));
_handler = context =>
{
handler(context);
return Task.FromResult(0);
};
}
public MqttApplicationMessageReceivedHandlerDelegate(Func<MqttApplicationMessageReceivedEventArgs, Task> handler)
{
_handler = handler ?? throw new ArgumentNullException(nameof(handler));
}
public Task HandleApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs context)
{
return _handler(context);
}
}
}
@@ -1,13 +1,9 @@
using System;
using MQTTnet.Client.Receiving;
using MQTTnet.Client.Receiving;
namespace MQTTnet
{
public interface IApplicationMessageReceiver
{
IMqttApplicationMessageHandler ReceivedApplicationMessageHandler { get; set; }
[Obsolete("Use _ReceivedApplicationMessageHandler_ instead.")]
event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get; set; }
}
}
+11
View File
@@ -0,0 +1,11 @@
using MQTTnet.Client;
using MQTTnet.Diagnostics;
using MQTTnet.Server;
namespace MQTTnet
{
public interface IMqttFactory : IMqttClientFactory, IMqttServerFactory
{
IMqttNetLogger DefaultLogger { get; }
}
}
@@ -1,6 +1,5 @@
using System;
using MQTTnet.Adapter;
using MQTTnet.Client;
using MQTTnet.Client.Options;
using MQTTnet.Diagnostics;
using MQTTnet.Formatter;
@@ -36,6 +36,8 @@ namespace MQTTnet.Implementations
_bufferSize = serverOptions.DefaultEndpointOptions.BufferSize;
CreateStreams();
IsSecureConnection = socket.Information.ProtectionLevel >= SocketProtectionLevel.Tls12;
}
public static Func<MqttClientTcpOptions, IEnumerable<ChainValidationResult>> CustomIgnorableServerCertificateErrorsResolver { get; set; }
@@ -53,6 +55,8 @@ namespace MQTTnet.Implementations
}
}
public bool IsSecureConnection { get; }
public async Task ConnectAsync(CancellationToken cancellationToken)
{
if (_socket == null)
@@ -27,6 +27,8 @@ namespace MQTTnet.Implementations
{
_clientOptions = clientOptions ?? throw new ArgumentNullException(nameof(clientOptions));
_options = (MqttClientTcpOptions)clientOptions.ChannelOptions;
IsSecureConnection = clientOptions.ChannelOptions?.TlsOptions?.UseTls == true;
}
/// <summary>
@@ -37,6 +39,8 @@ namespace MQTTnet.Implementations
{
_socket = socket ?? throw new ArgumentNullException(nameof(socket));
IsSecureConnection = sslStream != null;
CreateStream(sslStream);
}
@@ -45,6 +49,8 @@ namespace MQTTnet.Implementations
public string Endpoint => _socket?.RemoteEndPoint?.ToString();
public bool IsSecureConnection { get; }
public async Task ConnectAsync(CancellationToken cancellationToken)
{
if (_socket == null)
@@ -23,7 +23,7 @@ namespace MQTTnet.Implementations
_logger = logger.CreateChildLogger(nameof(MqttTcpServerAdapter));
}
public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
public Action<MqttServerAdapterClientAcceptedEventArgs> ClientAcceptedHandler { get; set; }
public async Task StartAsync(IMqttServerOptions options)
{
@@ -73,7 +73,7 @@ namespace MQTTnet.Implementations
try
{
var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(args.Socket, _options), new MqttPacketFormatterAdapter(), _logger);
ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
ClientAcceptedHandler?.Invoke(new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
}
catch (Exception exception)
{
@@ -26,7 +26,7 @@ namespace MQTTnet.Implementations
_logger = logger.CreateChildLogger(nameof(MqttTcpServerAdapter));
}
public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
public Action<MqttServerAdapterClientAcceptedEventArgs> ClientAcceptedHandler { get; set; }
public Task StartAsync(IMqttServerOptions options)
{
@@ -87,9 +87,11 @@ namespace MQTTnet.Implementations
options,
tlsCertificate,
_cancellationTokenSource.Token,
_logger);
_logger)
{
ClientAcceptedHandler = OnClientAccepted
};
listenerV4.ClientAccepted += OnClientAccepted;
listenerV4.Start();
_listeners.Add(listenerV4);
}
@@ -101,17 +103,19 @@ namespace MQTTnet.Implementations
options,
tlsCertificate,
_cancellationTokenSource.Token,
_logger);
_logger)
{
ClientAcceptedHandler = OnClientAccepted
};
listenerV6.ClientAccepted += OnClientAccepted;
listenerV6.Start();
_listeners.Add(listenerV6);
}
}
private void OnClientAccepted(object sender, MqttServerAdapterClientAcceptedEventArgs e)
private void OnClientAccepted(MqttServerAdapterClientAcceptedEventArgs eventArgs)
{
ClientAccepted?.Invoke(this, e);
ClientAcceptedHandler?.Invoke(eventArgs);
}
}
}
@@ -42,7 +42,7 @@ namespace MQTTnet.Implementations
}
}
public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
public Action<MqttServerAdapterClientAcceptedEventArgs> ClientAcceptedHandler;
public void Start()
{
@@ -88,7 +88,7 @@ namespace MQTTnet.Implementations
_addressFamily == AddressFamily.InterNetwork ? "ipv4" : "ipv6");
var clientAdapter = new MqttChannelAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketFormatterAdapter(), _logger);
ClientAccepted?.Invoke(this, new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
ClientAcceptedHandler?.Invoke(new MqttServerAdapterClientAcceptedEventArgs(clientAdapter));
}
catch (ObjectDisposedException)
{
@@ -1,4 +1,5 @@
using System;
using System.Globalization;
using System.Net;
using System.Net.WebSockets;
using System.Security.Cryptography.X509Certificates;
@@ -21,14 +22,18 @@ namespace MQTTnet.Implementations
_options = options ?? throw new ArgumentNullException(nameof(options));
}
public MqttWebSocketChannel(WebSocket webSocket, string endpoint)
public MqttWebSocketChannel(WebSocket webSocket, string endpoint, bool isSecureConnection)
{
_webSocket = webSocket ?? throw new ArgumentNullException(nameof(webSocket));
Endpoint = endpoint;
IsSecureConnection = isSecureConnection;
}
public string Endpoint { get; }
public bool IsSecureConnection { get; private set; }
public async Task ConnectAsync(CancellationToken cancellationToken)
{
var uri = _options.Uri;
@@ -83,6 +88,8 @@ namespace MQTTnet.Implementations
await clientWebSocket.ConnectAsync(new Uri(uri), cancellationToken).ConfigureAwait(false);
_webSocket = clientWebSocket;
IsSecureConnection = uri.StartsWith("wss://", StringComparison.OrdinalIgnoreCase);
}
public async Task DisconnectAsync(CancellationToken cancellationToken)
@@ -16,6 +16,8 @@ namespace MQTTnet.Internal
public string Endpoint { get; } = "<Test channel>";
public bool IsSecureConnection { get; } = false;
public Task ConnectAsync(CancellationToken cancellationToken)
{
return Task.FromResult(0);
@@ -13,5 +13,7 @@ namespace MQTTnet
public string ClientId { get; }
public MqttApplicationMessage ApplicationMessage { get; }
public bool ProcessingFailed { get; set; }
}
}
+22 -5
View File
@@ -8,11 +8,22 @@ using MQTTnet.Server;
namespace MQTTnet
{
public class MqttFactory : IMqttClientFactory, IMqttServerFactory
public class MqttFactory : IMqttFactory
{
public MqttFactory() : this(new MqttNetLogger())
{
}
public MqttFactory(IMqttNetLogger logger)
{
DefaultLogger = logger ?? throw new ArgumentNullException(nameof(logger));
}
public IMqttNetLogger DefaultLogger { get; }
public IMqttClient CreateMqttClient()
{
return CreateMqttClient(new MqttNetLogger());
return CreateMqttClient(DefaultLogger);
}
public IMqttClient CreateMqttClient(IMqttNetLogger logger)
@@ -26,7 +37,7 @@ namespace MQTTnet
{
if (adapterFactory == null) throw new ArgumentNullException(nameof(adapterFactory));
return new MqttClient(adapterFactory, new MqttNetLogger());
return new MqttClient(adapterFactory, DefaultLogger);
}
public IMqttClient CreateMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory adapterFactory)
@@ -39,8 +50,7 @@ namespace MQTTnet
public IMqttServer CreateMqttServer()
{
var logger = new MqttNetLogger();
return CreateMqttServer(logger);
return CreateMqttServer(DefaultLogger);
}
public IMqttServer CreateMqttServer(IMqttNetLogger logger)
@@ -57,5 +67,12 @@ namespace MQTTnet
return new MqttServer(adapters, logger.CreateChildLogger());
}
public IMqttServer CreateMqttServer(IEnumerable<IMqttServerAdapter> adapters)
{
if (adapters == null) throw new ArgumentNullException(nameof(adapters));
return new MqttServer(adapters, DefaultLogger.CreateChildLogger());
}
}
}
+7 -1
View File
@@ -27,7 +27,13 @@
public override string ToString()
{
return string.Concat("Connect: [ProtocolLevel=", ProtocolLevel, "] [ClientId=", ClientId, "] [Username=", Username, "] [Password=", Password, "] [KeepAlivePeriod=", KeepAlivePeriod, "] [CleanSession=", CleanSession, "]");
var password = Password;
if (!string.IsNullOrEmpty(password))
{
password = "****";
}
return string.Concat("Connect: [ProtocolLevel=", ProtocolLevel, "] [ClientId=", ClientId, "] [Username=", Username, "] [Password=", password, "] [KeepAlivePeriod=", KeepAlivePeriod, "] [CleanSession=", CleanSession, "]");
}
}
}
+7 -8
View File
@@ -1,5 +1,4 @@
using System;
using System.Collections.Generic;
using System.Collections.Generic;
using System.Threading.Tasks;
using MQTTnet.Server.Status;
@@ -7,13 +6,13 @@ namespace MQTTnet.Server
{
public interface IMqttServer : IApplicationMessageReceiver, IApplicationMessagePublisher
{
event EventHandler Started;
event EventHandler Stopped;
IMqttServerStartedHandler StartedHandler { get; set; }
IMqttServerStoppedHandler StoppedHandler { get; set; }
event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
event EventHandler<MqttClientSubscribedTopicEventArgs> ClientSubscribedTopic;
event EventHandler<MqttClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopic;
IMqttServerClientConnectedHandler ClientConnectedHandler { get; set; }
IMqttServerClientDisconnectedHandler ClientDisconnectedHandler { get; set; }
IMqttServerClientSubscribedTopicHandler ClientSubscribedTopicHandler { get; set; }
IMqttServerClientUnsubscribedTopicHandler ClientUnsubscribedTopicHandler { get; set; }
IMqttServerOptions Options { get; }
@@ -0,0 +1,9 @@
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public interface IMqttServerClientConnectedHandler
{
Task HandleClientConnectedAsync(MqttServerClientConnectedEventArgs eventArgs);
}
}
@@ -0,0 +1,9 @@
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public interface IMqttServerClientDisconnectedHandler
{
Task HandleClientDisconnectedAsync(MqttServerClientDisconnectedEventArgs eventArgs);
}
}
@@ -0,0 +1,9 @@
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public interface IMqttServerClientSubscribedTopicHandler
{
Task HandleClientSubscribedTopicAsync(MqttServerClientSubscribedTopicEventArgs eventArgs);
}
}
@@ -0,0 +1,9 @@
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public interface IMqttServerClientUnsubscribedTopicHandler
{
Task HandleClientUnsubscribedTopicAsync(MqttServerClientUnsubscribedTopicEventArgs eventArgs);
}
}
@@ -10,6 +10,8 @@ namespace MQTTnet.Server
IMqttServer CreateMqttServer(IMqttNetLogger logger);
IMqttServer CreateMqttServer(IEnumerable<IMqttServerAdapter> adapters);
IMqttServer CreateMqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger);
}
}
@@ -0,0 +1,10 @@
using System;
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public interface IMqttServerStartedHandler
{
Task HandleServerStartedAsync(EventArgs eventArgs);
}
}
@@ -0,0 +1,10 @@
using System;
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public interface IMqttServerStoppedHandler
{
Task HandleServerStoppedAsync(EventArgs eventArgs);
}
}
+1 -57
View File
@@ -14,62 +14,6 @@ using MQTTnet.Server.Status;
namespace MQTTnet.Server
{
public class MqttClientSession
{
private readonly DateTime _createdTimestamp = DateTime.UtcNow;
public MqttClientSession(string clientId, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions serverOptions)
{
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
SubscriptionsManager = new MqttClientSubscriptionsManager(clientId, eventDispatcher, serverOptions);
ApplicationMessagesQueue = new MqttClientSessionApplicationMessagesQueue(serverOptions);
}
public string ClientId { get; }
public bool IsCleanSession { get; set; } = true;
public MqttApplicationMessage WillMessage { get; set; }
public MqttClientSubscriptionsManager SubscriptionsManager { get; }
public MqttClientSessionApplicationMessagesQueue ApplicationMessagesQueue { get; }
public void EnqueueApplicationMessage(MqttApplicationMessage applicationMessage, string senderClientId, bool isRetainedApplicationMessage)
{
var checkSubscriptionsResult = SubscriptionsManager.CheckSubscriptions(applicationMessage.Topic, applicationMessage.QualityOfServiceLevel);
if (!checkSubscriptionsResult.IsSubscribed)
{
return;
}
ApplicationMessagesQueue.Enqueue(applicationMessage, senderClientId, checkSubscriptionsResult.QualityOfServiceLevel, isRetainedApplicationMessage);
}
public async Task SubscribeAsync(ICollection<TopicFilter> topicFilters, MqttRetainedMessagesManager retainedMessagesManager)
{
await SubscriptionsManager.SubscribeAsync(topicFilters).ConfigureAwait(false);
var matchingRetainedMessages = await retainedMessagesManager.GetSubscribedMessagesAsync(topicFilters).ConfigureAwait(false);
foreach (var matchingRetainedMessage in matchingRetainedMessages)
{
EnqueueApplicationMessage(matchingRetainedMessage, null, true);
}
}
public Task UnsubscribeAsync(IEnumerable<string> topicFilters)
{
return SubscriptionsManager.UnsubscribeAsync(topicFilters);
}
public void FillStatus(MqttSessionStatus status)
{
status.ClientId = ClientId;
status.CreatedTimestamp = _createdTimestamp;
status.PendingApplicationMessagesCount = ApplicationMessagesQueue.Count;
}
}
public class MqttClientConnection : IMqttClientSession, IDisposable
{
private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
@@ -425,7 +369,7 @@ namespace MQTTnet.Server
//// }
////});
await Task.FromResult(0);
//await Task.FromResult(0);
}
private async Task SendPendingPacketsAsync()
@@ -0,0 +1,63 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using MQTTnet.Server.Status;
namespace MQTTnet.Server
{
public class MqttClientSession
{
private readonly DateTime _createdTimestamp = DateTime.UtcNow;
public MqttClientSession(string clientId, MqttServerEventDispatcher eventDispatcher, IMqttServerOptions serverOptions)
{
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
SubscriptionsManager = new MqttClientSubscriptionsManager(clientId, eventDispatcher, serverOptions);
ApplicationMessagesQueue = new MqttClientSessionApplicationMessagesQueue(serverOptions);
}
public string ClientId { get; }
public bool IsCleanSession { get; set; } = true;
public MqttApplicationMessage WillMessage { get; set; }
public MqttClientSubscriptionsManager SubscriptionsManager { get; }
public MqttClientSessionApplicationMessagesQueue ApplicationMessagesQueue { get; }
public void EnqueueApplicationMessage(MqttApplicationMessage applicationMessage, string senderClientId, bool isRetainedApplicationMessage)
{
var checkSubscriptionsResult = SubscriptionsManager.CheckSubscriptions(applicationMessage.Topic, applicationMessage.QualityOfServiceLevel);
if (!checkSubscriptionsResult.IsSubscribed)
{
return;
}
ApplicationMessagesQueue.Enqueue(applicationMessage, senderClientId, checkSubscriptionsResult.QualityOfServiceLevel, isRetainedApplicationMessage);
}
public async Task SubscribeAsync(ICollection<TopicFilter> topicFilters, MqttRetainedMessagesManager retainedMessagesManager)
{
await SubscriptionsManager.SubscribeAsync(topicFilters).ConfigureAwait(false);
var matchingRetainedMessages = await retainedMessagesManager.GetSubscribedMessagesAsync(topicFilters).ConfigureAwait(false);
foreach (var matchingRetainedMessage in matchingRetainedMessages)
{
EnqueueApplicationMessage(matchingRetainedMessage, null, true);
}
}
public Task UnsubscribeAsync(IEnumerable<string> topicFilters)
{
return SubscriptionsManager.UnsubscribeAsync(topicFilters);
}
public void FillStatus(MqttSessionStatus status)
{
status.ClientId = ClientId;
status.CreatedTimestamp = _createdTimestamp;
status.PendingApplicationMessagesCount = ApplicationMessagesQueue.Count;
}
}
}
@@ -15,6 +15,7 @@ namespace MQTTnet.Server
{
private readonly BlockingCollection<MqttEnqueuedApplicationMessage> _messageQueue = new BlockingCollection<MqttEnqueuedApplicationMessage>();
private readonly SemaphoreSlim _createConnectionGate = new SemaphoreSlim(1, 1);
private readonly ConcurrentDictionary<string, MqttClientConnection> _connections = new ConcurrentDictionary<string, MqttClientConnection>();
private readonly ConcurrentDictionary<string, MqttClientSession> _sessions = new ConcurrentDictionary<string, MqttClientSession>();
@@ -49,8 +50,6 @@ namespace MQTTnet.Server
public async Task StopAsync()
{
//using (await _sessionsLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
foreach (var connection in _connections.Values)
{
await connection.StopAsync().ConfigureAwait(false);
@@ -133,7 +132,7 @@ namespace MQTTnet.Server
{
if (_connections.TryGetValue(clientId, out var connection))
{
await connection.StopAsync();
await connection.StopAsync().ConfigureAwait(false);
}
if (_sessions.TryRemove(clientId, out var session))
@@ -196,7 +195,7 @@ namespace MQTTnet.Server
applicationMessage = interceptorContext.ApplicationMessage;
}
_eventDispatcher.OnApplicationMessageReceived(sender?.ClientId, applicationMessage);
await _eventDispatcher.HandleApplicationMessageReceivedAsync(sender?.ClientId, applicationMessage).ConfigureAwait(false);
if (applicationMessage.Retain)
{
@@ -258,7 +257,7 @@ namespace MQTTnet.Server
var connection = await CreateConnectionAsync(channelAdapter, connectPacket).ConfigureAwait(false);
_eventDispatcher.OnClientConnected(clientId);
await _eventDispatcher.HandleClientConnectedAsync(clientId).ConfigureAwait(false);
disconnectType = await connection.RunAsync().ConfigureAwait(false);
}
@@ -288,7 +287,7 @@ namespace MQTTnet.Server
await TryCleanupChannelAsync(channelAdapter).ConfigureAwait(false);
_eventDispatcher.OnClientDisconnected(clientId, disconnectType);
await _eventDispatcher.HandleClientDisconnectedAsync(clientId, disconnectType).ConfigureAwait(false);
}
}
@@ -299,7 +298,8 @@ namespace MQTTnet.Server
connectPacket.Username,
connectPacket.Password,
connectPacket.WillMessage,
clientAdapter.Endpoint);
clientAdapter.Endpoint,
clientAdapter.IsSecureConnection);
var connectionValidator = _options.ConnectionValidator;
@@ -313,8 +313,6 @@ namespace MQTTnet.Server
return context;
}
private readonly SemaphoreSlim _createConnectionGate = new SemaphoreSlim(1, 1);
private async Task<MqttClientConnection> CreateConnectionAsync(IMqttChannelAdapter channelAdapter, MqttConnectPacket connectPacket)
{
await _createConnectionGate.WaitAsync(_cancellationToken).ConfigureAwait(false);
@@ -61,7 +61,7 @@ namespace MQTTnet.Server
_subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
}
_eventDispatcher.OnClientSubscribedTopic(_clientId, topicFilter);
await _eventDispatcher.HandleClientSubscribedTopicAsync(_clientId, topicFilter).ConfigureAwait(false);
}
}
@@ -85,12 +85,12 @@ namespace MQTTnet.Server
_subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
}
_eventDispatcher.OnClientSubscribedTopic(_clientId, topicFilter);
await _eventDispatcher.HandleClientSubscribedTopicAsync(_clientId, topicFilter).ConfigureAwait(false);
}
}
}
public Task<MqttUnsubAckPacket> UnsubscribeAsync(MqttUnsubscribePacket unsubscribePacket)
public async Task<MqttUnsubAckPacket> UnsubscribeAsync(MqttUnsubscribePacket unsubscribePacket)
{
if (unsubscribePacket == null) throw new ArgumentNullException(nameof(unsubscribePacket));
@@ -111,12 +111,15 @@ namespace MQTTnet.Server
{
unsubAckPacket.ReasonCodes.Add(MqttUnsubscribeReasonCode.NoSubscriptionExisted);
}
_eventDispatcher.OnClientUnsubscribedTopic(_clientId, topicFilter);
}
}
return Task.FromResult(unsubAckPacket);
foreach (var topicFilter in unsubscribePacket.TopicFilters)
{
await _eventDispatcher.HandleClientUnsubscribedTopicAsync(_clientId, topicFilter).ConfigureAwait(false);
}
return unsubAckPacket;
}
public Task UnsubscribeAsync(IEnumerable<string> topicFilters)
@@ -4,13 +4,14 @@ namespace MQTTnet.Server
{
public class MqttConnectionValidatorContext
{
public MqttConnectionValidatorContext(string clientId, string username, string password, MqttApplicationMessage willMessage, string endpoint)
public MqttConnectionValidatorContext(string clientId, string username, string password, MqttApplicationMessage willMessage, string endpoint, bool isSecureConnection)
{
ClientId = clientId;
Username = username;
Password = password;
WillMessage = willMessage;
Endpoint = endpoint;
IsSecureConnection = isSecureConnection;
}
public string ClientId { get; }
@@ -23,6 +24,8 @@ namespace MQTTnet.Server
public string Endpoint { get; }
public bool IsSecureConnection { get; }
public MqttConnectReturnCode ReturnCode { get; set; } = MqttConnectReturnCode.ConnectionAccepted;
}
}
+46 -30
View File
@@ -28,35 +28,41 @@ namespace MQTTnet.Server
if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.CreateChildLogger(nameof(MqttServer));
_eventDispatcher.ClientConnected += (s, e) => ClientConnected?.Invoke(s, e);
_eventDispatcher.ClientDisconnected += (s, e) => ClientDisconnected?.Invoke(s, e);
_eventDispatcher.ClientSubscribedTopic += (s, e) => ClientSubscribedTopic?.Invoke(s, e);
_eventDispatcher.ClientUnsubscribedTopic += (s, e) => ClientUnsubscribedTopic?.Invoke(s, e);
_eventDispatcher.ApplicationMessageReceived += async (s, e) =>
{
// TODO: Migrate EventDispatcher to proper handlers and no events anymore.
ApplicationMessageReceived?.Invoke(s, e);
var handler = ReceivedApplicationMessageHandler;
if (handler != null)
{
await handler.HandleApplicationMessageAsync(
new MqttApplicationMessageHandlerContext(e.ClientId, e.ApplicationMessage)).ConfigureAwait(false);
}
};
}
public event EventHandler Started;
public event EventHandler Stopped;
public IMqttServerStartedHandler StartedHandler { get; set; }
public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
public event EventHandler<MqttClientSubscribedTopicEventArgs> ClientSubscribedTopic;
public event EventHandler<MqttClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopic;
public IMqttServerStoppedHandler StoppedHandler { get; set; }
public IMqttApplicationMessageHandler ReceivedApplicationMessageHandler { get; set; }
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public IMqttServerClientConnectedHandler ClientConnectedHandler
{
get => _eventDispatcher.ClientConnectedHandler;
set => _eventDispatcher.ClientConnectedHandler = value;
}
public IMqttServerClientDisconnectedHandler ClientDisconnectedHandler
{
get => _eventDispatcher.ClientDisconnectedHandler;
set => _eventDispatcher.ClientDisconnectedHandler = value;
}
public IMqttServerClientSubscribedTopicHandler ClientSubscribedTopicHandler
{
get => _eventDispatcher.ClientSubscribedTopicHandler;
set => _eventDispatcher.ClientSubscribedTopicHandler = value;
}
public IMqttServerClientUnsubscribedTopicHandler ClientUnsubscribedTopicHandler
{
get => _eventDispatcher.ClientUnsubscribedTopicHandler;
set => _eventDispatcher.ClientUnsubscribedTopicHandler = value;
}
public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler
{
get => _eventDispatcher.ApplicationMessageReceivedHandler;
set => _eventDispatcher.ApplicationMessageReceivedHandler = value;
}
public IMqttServerOptions Options { get; private set; }
@@ -118,12 +124,17 @@ namespace MQTTnet.Server
foreach (var adapter in _adapters)
{
adapter.ClientAccepted += OnClientAccepted;
adapter.ClientAcceptedHandler = OnClientAccepted;
await adapter.StartAsync(Options).ConfigureAwait(false);
}
_logger.Info("Started.");
Started?.Invoke(this, EventArgs.Empty);
var startedHandler = StartedHandler;
if (startedHandler != null)
{
await startedHandler.HandleServerStartedAsync(EventArgs.Empty).ConfigureAwait(false);
}
}
public async Task StopAsync()
@@ -141,12 +152,11 @@ namespace MQTTnet.Server
foreach (var adapter in _adapters)
{
adapter.ClientAccepted -= OnClientAccepted;
adapter.ClientAcceptedHandler = null;
await adapter.StopAsync().ConfigureAwait(false);
}
_logger.Info("Stopped.");
Stopped?.Invoke(this, EventArgs.Empty);
}
finally
{
@@ -158,6 +168,12 @@ namespace MQTTnet.Server
_clientSessionsManager?.Dispose();
_clientSessionsManager = null;
}
var stoppedHandler = StoppedHandler;
if (stoppedHandler != null)
{
await stoppedHandler.HandleServerStoppedAsync(EventArgs.Empty).ConfigureAwait(false);
}
}
public Task ClearRetainedMessagesAsync()
@@ -165,7 +181,7 @@ namespace MQTTnet.Server
return _retainedMessagesManager?.ClearMessagesAsync();
}
private void OnClientAccepted(object sender, MqttServerAdapterClientAcceptedEventArgs eventArgs)
private void OnClientAccepted(MqttServerAdapterClientAcceptedEventArgs eventArgs)
{
eventArgs.SessionTask = _clientSessionsManager.HandleConnectionAsync(eventArgs.Client);
}
@@ -2,9 +2,9 @@
namespace MQTTnet.Server
{
public class MqttClientConnectedEventArgs : EventArgs
public class MqttServerClientConnectedEventArgs : EventArgs
{
public MqttClientConnectedEventArgs(string clientId)
public MqttServerClientConnectedEventArgs(string clientId)
{
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
}
@@ -0,0 +1,31 @@
using System;
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public class MqttServerClientConnectedHandlerDelegate : IMqttServerClientConnectedHandler
{
private readonly Func<MqttServerClientConnectedEventArgs, Task> _handler;
public MqttServerClientConnectedHandlerDelegate(Action<MqttServerClientConnectedEventArgs> handler)
{
if (handler == null) throw new ArgumentNullException(nameof(handler));
_handler = eventArgs =>
{
handler(eventArgs);
return Task.FromResult(0);
};
}
public MqttServerClientConnectedHandlerDelegate(Func<MqttServerClientConnectedEventArgs, Task> handler)
{
_handler = handler ?? throw new ArgumentNullException(nameof(handler));
}
public Task HandleClientConnectedAsync(MqttServerClientConnectedEventArgs eventArgs)
{
return _handler(eventArgs);
}
}
}
@@ -2,9 +2,9 @@
namespace MQTTnet.Server
{
public class MqttClientDisconnectedEventArgs : EventArgs
public class MqttServerClientDisconnectedEventArgs : EventArgs
{
public MqttClientDisconnectedEventArgs(string clientId, MqttClientDisconnectType disconnectType)
public MqttServerClientDisconnectedEventArgs(string clientId, MqttClientDisconnectType disconnectType)
{
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
DisconnectType = disconnectType;
@@ -0,0 +1,31 @@
using System;
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public class MqttServerClientDisconnectedHandlerDelegate : IMqttServerClientDisconnectedHandler
{
private readonly Func<MqttServerClientDisconnectedEventArgs, Task> _handler;
public MqttServerClientDisconnectedHandlerDelegate(Action<MqttServerClientDisconnectedEventArgs> handler)
{
if (handler == null) throw new ArgumentNullException(nameof(handler));
_handler = eventArgs =>
{
handler(eventArgs);
return Task.FromResult(0);
};
}
public MqttServerClientDisconnectedHandlerDelegate(Func<MqttServerClientDisconnectedEventArgs, Task> handler)
{
_handler = handler ?? throw new ArgumentNullException(nameof(handler));
}
public Task HandleClientDisconnectedAsync(MqttServerClientDisconnectedEventArgs eventArgs)
{
return _handler(eventArgs);
}
}
}
@@ -2,9 +2,9 @@
namespace MQTTnet.Server
{
public class MqttClientSubscribedTopicEventArgs : EventArgs
public class MqttServerClientSubscribedTopicEventArgs : EventArgs
{
public MqttClientSubscribedTopicEventArgs(string clientId, TopicFilter topicFilter)
public MqttServerClientSubscribedTopicEventArgs(string clientId, TopicFilter topicFilter)
{
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter));
@@ -0,0 +1,31 @@
using System;
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public class MqttServerClientSubscribedHandlerDelegate : IMqttServerClientSubscribedTopicHandler
{
private readonly Func<MqttServerClientSubscribedTopicEventArgs, Task> _handler;
public MqttServerClientSubscribedHandlerDelegate(Action<MqttServerClientSubscribedTopicEventArgs> handler)
{
if (handler == null) throw new ArgumentNullException(nameof(handler));
_handler = eventArgs =>
{
handler(eventArgs);
return Task.FromResult(0);
};
}
public MqttServerClientSubscribedHandlerDelegate(Func<MqttServerClientSubscribedTopicEventArgs, Task> handler)
{
_handler = handler ?? throw new ArgumentNullException(nameof(handler));
}
public Task HandleClientSubscribedTopicAsync(MqttServerClientSubscribedTopicEventArgs eventArgs)
{
return _handler(eventArgs);
}
}
}
@@ -2,9 +2,9 @@
namespace MQTTnet.Server
{
public class MqttClientUnsubscribedTopicEventArgs : EventArgs
public class MqttServerClientUnsubscribedTopicEventArgs : EventArgs
{
public MqttClientUnsubscribedTopicEventArgs(string clientId, string topicFilter)
public MqttServerClientUnsubscribedTopicEventArgs(string clientId, string topicFilter)
{
ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId));
TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter));
@@ -0,0 +1,31 @@
using System;
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public class MqttServerClientUnsubscribedTopicHandlerDelegate : IMqttServerClientUnsubscribedTopicHandler
{
private readonly Func<MqttServerClientUnsubscribedTopicEventArgs, Task> _handler;
public MqttServerClientUnsubscribedTopicHandlerDelegate(Action<MqttServerClientUnsubscribedTopicEventArgs> handler)
{
if (handler == null) throw new ArgumentNullException(nameof(handler));
_handler = eventArgs =>
{
handler(eventArgs);
return Task.FromResult(0);
};
}
public MqttServerClientUnsubscribedTopicHandlerDelegate(Func<MqttServerClientUnsubscribedTopicEventArgs, Task> handler)
{
_handler = handler ?? throw new ArgumentNullException(nameof(handler));
}
public Task HandleClientUnsubscribedTopicAsync(MqttServerClientUnsubscribedTopicEventArgs eventArgs)
{
return _handler(eventArgs);
}
}
}
@@ -1,42 +1,73 @@
using System;
using System.Threading.Tasks;
using MQTTnet.Client.Receiving;
namespace MQTTnet.Server
{
public class MqttServerEventDispatcher
{
public event EventHandler<MqttClientSubscribedTopicEventArgs> ClientSubscribedTopic;
public IMqttServerClientConnectedHandler ClientConnectedHandler { get; set; }
public event EventHandler<MqttClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopic;
public IMqttServerClientDisconnectedHandler ClientDisconnectedHandler { get; set; }
public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
public IMqttServerClientSubscribedTopicHandler ClientSubscribedTopicHandler { get; set; }
public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
public IMqttServerClientUnsubscribedTopicHandler ClientUnsubscribedTopicHandler { get; set; }
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public IMqttApplicationMessageReceivedHandler ApplicationMessageReceivedHandler { get; set; }
public void OnClientSubscribedTopic(string clientId, TopicFilter topicFilter)
public Task HandleClientConnectedAsync(string clientId)
{
ClientSubscribedTopic?.Invoke(this, new MqttClientSubscribedTopicEventArgs(clientId, topicFilter));
var handler = ClientConnectedHandler;
if (handler == null)
{
return Task.FromResult(0);
}
return handler.HandleClientConnectedAsync(new MqttServerClientConnectedEventArgs(clientId));
}
public void OnClientUnsubscribedTopic(string clientId, string topicFilter)
public Task HandleClientDisconnectedAsync(string clientId, MqttClientDisconnectType disconnectType)
{
ClientUnsubscribedTopic?.Invoke(this, new MqttClientUnsubscribedTopicEventArgs(clientId, topicFilter));
var handler = ClientDisconnectedHandler;
if (handler == null)
{
return Task.FromResult(0);
}
return handler.HandleClientDisconnectedAsync(new MqttServerClientDisconnectedEventArgs(clientId, disconnectType));
}
public void OnClientDisconnected(string clientId, MqttClientDisconnectType disconnectType)
public Task HandleClientSubscribedTopicAsync(string clientId, TopicFilter topicFilter)
{
ClientDisconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(clientId, disconnectType));
var handler = ClientSubscribedTopicHandler;
if (handler == null)
{
return Task.FromResult(0);
}
return handler.HandleClientSubscribedTopicAsync(new MqttServerClientSubscribedTopicEventArgs(clientId, topicFilter));
}
public void OnApplicationMessageReceived(string senderClientId, MqttApplicationMessage applicationMessage)
public Task HandleClientUnsubscribedTopicAsync(string clientId, string topicFilter)
{
ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage));
var handler = ClientUnsubscribedTopicHandler;
if (handler == null)
{
return Task.FromResult(0);
}
return handler.HandleClientUnsubscribedTopicAsync(new MqttServerClientUnsubscribedTopicEventArgs(clientId, topicFilter));
}
public void OnClientConnected(string clientId)
public Task HandleApplicationMessageReceivedAsync(string senderClientId, MqttApplicationMessage applicationMessage)
{
ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientId));
var handler = ApplicationMessageReceivedHandler;
if (handler == null)
{
return Task.FromResult(0);
}
return handler.HandleApplicationMessageReceivedAsync(new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage));
}
}
}
@@ -75,6 +75,12 @@ namespace MQTTnet.Server
return this;
}
public MqttServerOptionsBuilder WithEncryptedEndpointBoundIPV6Address(IPAddress value)
{
_options.TlsEndpointOptions.BoundInterNetworkV6Address = value;
return this;
}
public MqttServerOptionsBuilder WithEncryptionCertificate(byte[] value)
{
_options.TlsEndpointOptions.Certificate = value;
@@ -0,0 +1,31 @@
using System;
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public class MqttServerStartedHandlerDelegate : IMqttServerStartedHandler
{
private readonly Func<EventArgs, Task> _handler;
public MqttServerStartedHandlerDelegate(Action<EventArgs> handler)
{
if (handler == null) throw new ArgumentNullException(nameof(handler));
_handler = eventArgs =>
{
handler(eventArgs);
return Task.FromResult(0);
};
}
public MqttServerStartedHandlerDelegate(Func<EventArgs, Task> handler)
{
_handler = handler ?? throw new ArgumentNullException(nameof(handler));
}
public Task HandleServerStartedAsync(EventArgs eventArgs)
{
return _handler(eventArgs);
}
}
}
@@ -0,0 +1,31 @@
using System;
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public class MqttServerStoppedHandlerDelegate : IMqttServerStoppedHandler
{
private readonly Func<EventArgs, Task> _handler;
public MqttServerStoppedHandlerDelegate(Action<EventArgs> handler)
{
if (handler == null) throw new ArgumentNullException(nameof(handler));
_handler = eventArgs =>
{
handler(eventArgs);
return Task.FromResult(0);
};
}
public MqttServerStoppedHandlerDelegate(Func<EventArgs, Task> handler)
{
_handler = handler ?? throw new ArgumentNullException(nameof(handler));
}
public Task HandleServerStoppedAsync(EventArgs eventArgs)
{
return _handler(eventArgs);
}
}
}