diff --git a/Source/MQTTnet/Formatter/MqttConnAckPacketFactory.cs b/Source/MQTTnet/Formatter/MqttConnAckPacketFactory.cs index 83d63977..043357fa 100644 --- a/Source/MQTTnet/Formatter/MqttConnAckPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttConnAckPacketFactory.cs @@ -6,7 +6,7 @@ namespace MQTTnet.Formatter { public sealed class MqttConnAckPacketFactory { - public MqttConnAckPacket Create(ValidatingMqttClientConnectionEventArgs clientConnectionValidationEventArgs) + public MqttConnAckPacket Create(ValidatingConnectionEventArgs clientConnectionValidationEventArgs) { if (clientConnectionValidationEventArgs == null) throw new ArgumentNullException(nameof(clientConnectionValidationEventArgs)); diff --git a/Source/MQTTnet/Formatter/MqttPubAckPacketFactory.cs b/Source/MQTTnet/Formatter/MqttPubAckPacketFactory.cs index 0ad95631..6d587017 100644 --- a/Source/MQTTnet/Formatter/MqttPubAckPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttPubAckPacketFactory.cs @@ -8,7 +8,7 @@ namespace MQTTnet.Formatter { public sealed class MqttPubAckPacketFactory { - public MqttPubAckPacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageResponse applicationMessageResponse) + public MqttPubAckPacket Create(MqttPublishPacket publishPacket, PublishResponse applicationMessageResponse) { if (applicationMessageResponse == null) throw new ArgumentNullException(nameof(applicationMessageResponse)); diff --git a/Source/MQTTnet/Formatter/MqttPubRecPacketFactory.cs b/Source/MQTTnet/Formatter/MqttPubRecPacketFactory.cs index 4f0c5f5c..90ab86b6 100644 --- a/Source/MQTTnet/Formatter/MqttPubRecPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttPubRecPacketFactory.cs @@ -18,7 +18,7 @@ namespace MQTTnet.Formatter return pubRecPacket; } - public MqttBasePacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageResponse applicationMessageResponse) + public MqttBasePacket Create(MqttPublishPacket publishPacket, PublishResponse applicationMessageResponse) { if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); diff --git a/Source/MQTTnet/Formatter/MqttSubAckPacketFactory.cs b/Source/MQTTnet/Formatter/MqttSubAckPacketFactory.cs index ae9a0c4f..d944c5cc 100644 --- a/Source/MQTTnet/Formatter/MqttSubAckPacketFactory.cs +++ b/Source/MQTTnet/Formatter/MqttSubAckPacketFactory.cs @@ -6,17 +6,17 @@ namespace MQTTnet.Formatter { public sealed class MqttSubAckPacketFactory { - public MqttSubAckPacket Create(MqttSubscribePacket subscribePacket, MqttSubscribeResult mqttSubscribeResult) + public MqttSubAckPacket Create(MqttSubscribePacket subscribePacket, SubscribeResult subscribeResult) { if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket)); - if (mqttSubscribeResult == null) throw new ArgumentNullException(nameof(mqttSubscribeResult)); + if (subscribeResult == null) throw new ArgumentNullException(nameof(subscribeResult)); var subAckPacket = new MqttSubAckPacket { PacketIdentifier = subscribePacket.PacketIdentifier }; - subAckPacket.ReasonCodes.AddRange(mqttSubscribeResult.ReasonCodes); + subAckPacket.ReasonCodes.AddRange(subscribeResult.ReasonCodes); return subAckPacket; } diff --git a/Source/MQTTnet/MqttFactory.cs b/Source/MQTTnet/MqttFactory.cs index a8ad266c..378a864e 100644 --- a/Source/MQTTnet/MqttFactory.cs +++ b/Source/MQTTnet/MqttFactory.cs @@ -7,6 +7,7 @@ using System; using System.Collections.Generic; using System.Linq; using MQTTnet.Diagnostics; +using MqttClient = MQTTnet.Client.MqttClient; namespace MQTTnet { diff --git a/Source/MQTTnet/Server/Events/MqttApplicationMessageNotConsumedEventArgs.cs b/Source/MQTTnet/Server/Events/ApplicationMessageNotConsumedEventArgs.cs similarity index 85% rename from Source/MQTTnet/Server/Events/MqttApplicationMessageNotConsumedEventArgs.cs rename to Source/MQTTnet/Server/Events/ApplicationMessageNotConsumedEventArgs.cs index bbc09675..885d9ae1 100644 --- a/Source/MQTTnet/Server/Events/MqttApplicationMessageNotConsumedEventArgs.cs +++ b/Source/MQTTnet/Server/Events/ApplicationMessageNotConsumedEventArgs.cs @@ -2,7 +2,7 @@ namespace MQTTnet.Server { - public sealed class MqttApplicationMessageNotConsumedEventArgs : EventArgs + public sealed class ApplicationMessageNotConsumedEventArgs : EventArgs { /// /// Gets the application message which was not consumed by any client. diff --git a/Source/MQTTnet/Server/Events/MqttClientConnectedEventArgs.cs b/Source/MQTTnet/Server/Events/ClientConnectedEventArgs.cs similarity index 92% rename from Source/MQTTnet/Server/Events/MqttClientConnectedEventArgs.cs rename to Source/MQTTnet/Server/Events/ClientConnectedEventArgs.cs index 21e8df3e..6bb82f70 100644 --- a/Source/MQTTnet/Server/Events/MqttClientConnectedEventArgs.cs +++ b/Source/MQTTnet/Server/Events/ClientConnectedEventArgs.cs @@ -3,7 +3,7 @@ using MQTTnet.Formatter; namespace MQTTnet.Server { - public sealed class MqttServerClientConnectedEventArgs : EventArgs + public sealed class ClientConnectedEventArgs : EventArgs { /// /// Gets the client identifier of the connected client. diff --git a/Source/MQTTnet/Server/Events/MqttClientDisconnectedEventArgs.cs b/Source/MQTTnet/Server/Events/ClientDisconnectedEventArgs.cs similarity index 86% rename from Source/MQTTnet/Server/Events/MqttClientDisconnectedEventArgs.cs rename to Source/MQTTnet/Server/Events/ClientDisconnectedEventArgs.cs index 77ad0763..3edcd395 100644 --- a/Source/MQTTnet/Server/Events/MqttClientDisconnectedEventArgs.cs +++ b/Source/MQTTnet/Server/Events/ClientDisconnectedEventArgs.cs @@ -2,7 +2,7 @@ using System; namespace MQTTnet.Server { - public sealed class MqttServerClientDisconnectedEventArgs : EventArgs + public sealed class ClientDisconnectedEventArgs : EventArgs { /// /// Gets the client identifier. diff --git a/Source/MQTTnet/Server/Events/MqttClientSubscribedTopicEventArgs.cs b/Source/MQTTnet/Server/Events/ClientSubscribedTopicEventArgs.cs similarity index 87% rename from Source/MQTTnet/Server/Events/MqttClientSubscribedTopicEventArgs.cs rename to Source/MQTTnet/Server/Events/ClientSubscribedTopicEventArgs.cs index b94ed184..c87a9dcd 100644 --- a/Source/MQTTnet/Server/Events/MqttClientSubscribedTopicEventArgs.cs +++ b/Source/MQTTnet/Server/Events/ClientSubscribedTopicEventArgs.cs @@ -3,7 +3,7 @@ using MQTTnet.Packets; namespace MQTTnet.Server { - public sealed class MqttServerClientSubscribedTopicEventArgs : EventArgs + public sealed class ClientSubscribedTopicEventArgs : EventArgs { /// /// Gets the client identifier. diff --git a/Source/MQTTnet/Server/Events/MqttClientUnsubscribedTopicEventArgs.cs b/Source/MQTTnet/Server/Events/ClientUnsubscribedTopicEventArgs.cs similarity index 87% rename from Source/MQTTnet/Server/Events/MqttClientUnsubscribedTopicEventArgs.cs rename to Source/MQTTnet/Server/Events/ClientUnsubscribedTopicEventArgs.cs index 341e1f4b..b2e11c53 100644 --- a/Source/MQTTnet/Server/Events/MqttClientUnsubscribedTopicEventArgs.cs +++ b/Source/MQTTnet/Server/Events/ClientUnsubscribedTopicEventArgs.cs @@ -2,7 +2,7 @@ namespace MQTTnet.Server { - public sealed class MqttServerClientUnsubscribedTopicEventArgs : EventArgs + public sealed class ClientUnsubscribedTopicEventArgs : EventArgs { /// /// Gets the client identifier. diff --git a/Source/MQTTnet/Server/Events/InterceptingMqttClientPublishEventArgs.cs b/Source/MQTTnet/Server/Events/InterceptingPublishEventArgs.cs similarity index 87% rename from Source/MQTTnet/Server/Events/InterceptingMqttClientPublishEventArgs.cs rename to Source/MQTTnet/Server/Events/InterceptingPublishEventArgs.cs index ca67b04c..5054bf39 100644 --- a/Source/MQTTnet/Server/Events/InterceptingMqttClientPublishEventArgs.cs +++ b/Source/MQTTnet/Server/Events/InterceptingPublishEventArgs.cs @@ -4,7 +4,7 @@ using System.Threading; namespace MQTTnet.Server { - public sealed class InterceptingMqttClientPublishEventArgs : EventArgs + public sealed class InterceptingPublishEventArgs : EventArgs { /// /// Gets the client identifier. @@ -22,7 +22,7 @@ namespace MQTTnet.Server /// /// Gets the response which will be sent to the client via the PUBACK etc. packets. /// - public MqttApplicationMessageResponse Response { get; } = new MqttApplicationMessageResponse(); + public PublishResponse Response { get; } = new PublishResponse(); /// /// Gets or sets whether the publish should be processed internally. diff --git a/Source/MQTTnet/Server/Events/InterceptingMqttClientSubscriptionEventArgs.cs b/Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs similarity index 92% rename from Source/MQTTnet/Server/Events/InterceptingMqttClientSubscriptionEventArgs.cs rename to Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs index 17c85d34..724f2c7e 100644 --- a/Source/MQTTnet/Server/Events/InterceptingMqttClientSubscriptionEventArgs.cs +++ b/Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs @@ -5,7 +5,7 @@ using MQTTnet.Packets; namespace MQTTnet.Server { - public sealed class InterceptingMqttClientSubscriptionEventArgs : EventArgs + public sealed class InterceptingSubscriptionEventArgs : EventArgs { /// /// Gets the client identifier. @@ -44,7 +44,7 @@ namespace MQTTnet.Server /// /// Gets the response which will be sent to the client via the SUBACK packet. /// - public MqttSubscribeResponse Response { get; } = new MqttSubscribeResponse(); + public SubscribeResponse Response { get; } = new SubscribeResponse(); /// /// Gets the cancellation token which can indicate that the client connection gets down. diff --git a/Source/MQTTnet/Server/Events/InterceptingMqttClientUnsubscriptionEventArgs.cs b/Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs similarity index 91% rename from Source/MQTTnet/Server/Events/InterceptingMqttClientUnsubscriptionEventArgs.cs rename to Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs index 0092abb7..8f90df43 100644 --- a/Source/MQTTnet/Server/Events/InterceptingMqttClientUnsubscriptionEventArgs.cs +++ b/Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs @@ -4,7 +4,7 @@ using System.Threading; namespace MQTTnet.Server { - public sealed class InterceptingMqttClientUnsubscriptionEventArgs : EventArgs + public sealed class InterceptingUnsubscriptionEventArgs : EventArgs { /// /// Gets the client identifier. @@ -27,7 +27,7 @@ namespace MQTTnet.Server /// /// Gets the response which will be sent to the client via the UNSUBACK pocket. /// - public MqttUnsubscribeResponse Response { get; } = new MqttUnsubscribeResponse(); + public UnsubscribeResponse Response { get; } = new UnsubscribeResponse(); /// /// Gets or sets whether the broker should remove an internal subscription for the client. diff --git a/Source/MQTTnet/Server/Events/LoadingRetainedMessagesEventArgs.cs b/Source/MQTTnet/Server/Events/LoadingRetainedMessagesEventArgs.cs new file mode 100644 index 00000000..d557f08e --- /dev/null +++ b/Source/MQTTnet/Server/Events/LoadingRetainedMessagesEventArgs.cs @@ -0,0 +1,10 @@ +using System; +using System.Collections.Generic; + +namespace MQTTnet.Server +{ + public sealed class LoadingRetainedMessagesEventArgs : EventArgs + { + public List StoredRetainedMessages { get; set; } = new List(); + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Events/PreparingMqttClientSessionEventArgs.cs b/Source/MQTTnet/Server/Events/PreparingSessionEventArgs.cs similarity index 62% rename from Source/MQTTnet/Server/Events/PreparingMqttClientSessionEventArgs.cs rename to Source/MQTTnet/Server/Events/PreparingSessionEventArgs.cs index 14968984..befa334b 100644 --- a/Source/MQTTnet/Server/Events/PreparingMqttClientSessionEventArgs.cs +++ b/Source/MQTTnet/Server/Events/PreparingSessionEventArgs.cs @@ -4,9 +4,9 @@ using MQTTnet.Packets; namespace MQTTnet.Server { - public sealed class PreparingMqttClientSessionEventArgs : EventArgs + public sealed class PreparingSessionEventArgs : EventArgs { - public string ClientId { get; internal set; } + public string Id { get; internal set; } // TODO: Allow adding of packets to the queue etc. @@ -25,5 +25,21 @@ namespace MQTTnet.Server public List Subscriptions { get; } = new List(); public List PublishPackets { get; } = new List(); + + IDictionary Items { get; set; } + + /// + /// Gets the last will message. + /// In MQTT, you use the last will message feature to notify other clients about an ungracefully disconnected client. + /// + MqttApplicationMessage WillMessage { get; set; } + + /// + /// Gets the will delay interval. + /// This is the time between the client disconnect and the time the will message will be sent. + /// + uint? WillDelayInterval { get; set; } + + DateTime? SessionExpiryTimestamp { get; set; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Events/RetainedMessageChangedEventArgs.cs b/Source/MQTTnet/Server/Events/RetainedMessageChangedEventArgs.cs new file mode 100644 index 00000000..507f84de --- /dev/null +++ b/Source/MQTTnet/Server/Events/RetainedMessageChangedEventArgs.cs @@ -0,0 +1,14 @@ +using System; +using System.Collections.Generic; + +namespace MQTTnet.Server +{ + public sealed class RetainedMessageChangedEventArgs : EventArgs + { + public string ClientId { get; internal set; } + + public MqttApplicationMessage ChangedRetainedMessage { get; internal set; } + + public List StoredRetainedMessages { get; internal set; } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Events/ValidatingMqttClientConnectionEventArgs.cs b/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs similarity index 97% rename from Source/MQTTnet/Server/Events/ValidatingMqttClientConnectionEventArgs.cs rename to Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs index f412db7a..757b6eb2 100644 --- a/Source/MQTTnet/Server/Events/ValidatingMqttClientConnectionEventArgs.cs +++ b/Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs @@ -10,12 +10,12 @@ using MQTTnet.Protocol; namespace MQTTnet.Server { - public sealed class ValidatingMqttClientConnectionEventArgs : EventArgs + public sealed class ValidatingConnectionEventArgs : EventArgs { readonly MqttConnectPacket _connectPacket; readonly IMqttChannelAdapter _clientAdapter; - public ValidatingMqttClientConnectionEventArgs(MqttConnectPacket connectPacket, IMqttChannelAdapter clientAdapter) + public ValidatingConnectionEventArgs(MqttConnectPacket connectPacket, IMqttChannelAdapter clientAdapter) { _connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket)); _clientAdapter = clientAdapter ?? throw new ArgumentNullException(nameof(clientAdapter)); diff --git a/Source/MQTTnet/Server/Internal/MqttApplicationMessageInterceptorInvoker.cs b/Source/MQTTnet/Server/Internal/MqttApplicationMessageInterceptorInvoker.cs index ffff15a0..dad6ad84 100644 --- a/Source/MQTTnet/Server/Internal/MqttApplicationMessageInterceptorInvoker.cs +++ b/Source/MQTTnet/Server/Internal/MqttApplicationMessageInterceptorInvoker.cs @@ -20,7 +20,7 @@ namespace MQTTnet.Server public MqttApplicationMessage ApplicationMessage { get; private set; } - public MqttApplicationMessageResponse Response { get; private set; } = new MqttApplicationMessageResponse(); + public PublishResponse Response { get; private set; } = new PublishResponse(); public bool ProcessPublish { get; private set; } = true; @@ -29,13 +29,13 @@ namespace MQTTnet.Server public async Task Invoke(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken) { // Reset to defaults. - Response = new MqttApplicationMessageResponse(); + Response = new PublishResponse(); ProcessPublish = true; CloseConnection = false; ApplicationMessage = applicationMessage; // Intercept. - var eventArgs = new InterceptingMqttClientPublishEventArgs + var eventArgs = new InterceptingPublishEventArgs { ClientId = _clientId, ApplicationMessage = applicationMessage, @@ -53,7 +53,7 @@ namespace MQTTnet.Server eventArgs.ProcessPublish = false; } - await _eventContainer.InterceptingClientPublishEvent.InvokeAsync(eventArgs).ConfigureAwait(false); + await _eventContainer.InterceptingPublishEvent.InvokeAsync(eventArgs).ConfigureAwait(false); // Expose results. ProcessPublish = eventArgs.ProcessPublish && eventArgs.ApplicationMessage != null; diff --git a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs b/Source/MQTTnet/Server/Internal/MqttClient.cs similarity index 98% rename from Source/MQTTnet/Server/Internal/MqttClientConnection.cs rename to Source/MQTTnet/Server/Internal/MqttClient.cs index 203b0a25..6f5d5f90 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/Internal/MqttClient.cs @@ -14,7 +14,7 @@ using MQTTnet.Protocol; namespace MQTTnet.Server { - public sealed class MqttClientConnection : IDisposable + public sealed class MqttClient : IDisposable { readonly Dictionary _topicAlias = new Dictionary(); readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher(); @@ -32,10 +32,10 @@ namespace MQTTnet.Server CancellationTokenSource _cancellationToken; - public MqttClientConnection( + public MqttClient( MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, - MqttClientSession session, + MqttSession session, MqttServerOptions serverOptions, MqttServerEventContainer eventContainer, MqttClientSessionsManager sessionsManager, @@ -54,7 +54,7 @@ namespace MQTTnet.Server _applicationMessageInterceptorInvoker = new MqttApplicationMessageInterceptorInvoker(eventContainer, Id, Session.Items); if (logger == null) throw new ArgumentNullException(nameof(logger)); - _logger = logger.WithSource(nameof(MqttClientConnection)); + _logger = logger.WithSource(nameof(MqttClient)); } public string Id => _connectPacket.ClientId; @@ -63,13 +63,13 @@ namespace MQTTnet.Server public IMqttChannelAdapter ChannelAdapter { get; } - public MqttClientConnectionStatistics Statistics { get; } = new MqttClientConnectionStatistics(); + public MqttClientStatistics Statistics { get; } = new MqttClientStatistics(); public bool IsRunning { get; private set; } public ushort KeepAlivePeriod => _connectPacket.KeepAlivePeriod; - public MqttClientSession Session { get; } + public MqttSession Session { get; } public bool IsTakenOver { get; set; } diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 13a8b3cb..c722a45f 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -20,14 +20,14 @@ namespace MQTTnet.Server readonly MqttPacketFactories _packetFactories = new MqttPacketFactories(); readonly AsyncLock _createConnectionSyncRoot = new AsyncLock(); - readonly Dictionary _clientConnections = new Dictionary(4096); - readonly Dictionary _clientSessions = new Dictionary(4096); + readonly Dictionary _clients = new Dictionary(4096); + readonly Dictionary _sessions = new Dictionary(4096); readonly IDictionary _serverSessionItems = new ConcurrentDictionary(); readonly MqttConnAckPacketFactory _connAckPacketFactory = new MqttConnAckPacketFactory(); - readonly IMqttRetainedMessagesManager _retainedMessagesManager; + readonly MqttRetainedMessagesManager _retainedMessagesManager; readonly MqttServerEventContainer _eventContainer; readonly MqttServerOptions _options; readonly MqttNetSourceLogger _logger; @@ -35,7 +35,7 @@ namespace MQTTnet.Server public MqttClientSessionsManager( MqttServerOptions options, - IMqttRetainedMessagesManager retainedMessagesManager, + MqttRetainedMessagesManager retainedMessagesManager, MqttServerEventContainer eventContainer, IMqttNetLogger logger) { @@ -78,7 +78,7 @@ namespace MQTTnet.Server public async Task HandleClientConnectionAsync(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken) { - MqttClientConnection clientConnection = null; + MqttClient client = null; try { @@ -103,11 +103,11 @@ namespace MQTTnet.Server connAckPacket = _connAckPacketFactory.Create(connectionValidatorContext); // Pass connAckPacket so that IsSessionPresent flag can be set if the client session already exists - clientConnection = await CreateClientConnection(connectPacket, connAckPacket, channelAdapter, connectionValidatorContext.SessionItems).ConfigureAwait(false); + client = await CreateClientConnection(connectPacket, connAckPacket, channelAdapter, connectionValidatorContext.SessionItems).ConfigureAwait(false); - await clientConnection.SendPacketAsync(connAckPacket, cancellationToken).ConfigureAwait(false); + await client.SendPacketAsync(connAckPacket, cancellationToken).ConfigureAwait(false); - await _eventContainer.ClientConnectedEvent.InvokeAsync(() => new MqttServerClientConnectedEventArgs + await _eventContainer.ClientConnectedEvent.InvokeAsync(() => new ClientConnectedEventArgs { ClientId = connectPacket.ClientId, UserName = connectPacket.Username, @@ -115,7 +115,7 @@ namespace MQTTnet.Server Endpoint = channelAdapter.Endpoint }).ConfigureAwait(false); - await clientConnection.RunAsync().ConfigureAwait(false); + await client.RunAsync().ConfigureAwait(false); } catch (OperationCanceledException) { @@ -126,33 +126,33 @@ namespace MQTTnet.Server } finally { - if (clientConnection != null) + if (client != null) { - if (clientConnection.Id != null) + if (client.Id != null) { // in case it is a takeover _clientConnections already contains the new connection - if (!clientConnection.IsTakenOver) + if (!client.IsTakenOver) { - lock (_clientConnections) + lock (_clients) { - _clientConnections.Remove(clientConnection.Id); + _clients.Remove(client.Id); } if (!_options.EnablePersistentSessions) { - await DeleteSessionAsync(clientConnection.Id).ConfigureAwait(false); + await DeleteSessionAsync(client.Id).ConfigureAwait(false); } } } - var endpoint = clientConnection.Endpoint; + var endpoint = client.Endpoint; - if (clientConnection.Id != null && !clientConnection.IsTakenOver) + if (client.Id != null && !client.IsTakenOver) { - await _eventContainer.ClientDisconnectedEvent.InvokeAsync(() => new MqttServerClientDisconnectedEventArgs + await _eventContainer.ClientDisconnectedEvent.InvokeAsync(() => new ClientDisconnectedEventArgs { - ClientId = clientConnection.Id, - DisconnectType = clientConnection.IsCleanDisconnect ? MqttClientDisconnectType.Clean : MqttClientDisconnectType.NotClean, + ClientId = client.Id, + DisconnectType = client.IsCleanDisconnect ? MqttClientDisconnectType.Clean : MqttClientDisconnectType.NotClean, Endpoint = endpoint }).ConfigureAwait(false); } @@ -167,11 +167,11 @@ namespace MQTTnet.Server public async Task CloseAllConnectionsAsync() { - List connections; - lock (_clientConnections) + List connections; + lock (_clients) { - connections = _clientConnections.Values.ToList(); - _clientConnections.Clear(); + connections = _clients.Values.ToList(); + _clients.Clear(); } foreach (var connection in connections) @@ -180,11 +180,11 @@ namespace MQTTnet.Server } } - public List GetConnections() + public List GetConnections() { - lock (_clientConnections) + lock (_clients) { - return _clientConnections.Values.ToList(); + return _clients.Values.ToList(); } } @@ -192,9 +192,9 @@ namespace MQTTnet.Server { var result = new List(); - lock (_clientConnections) + lock (_clients) { - foreach (var connection in _clientConnections.Values) + foreach (var connection in _clients.Values) { var clientStatus = new MqttClientStatus(connection) { @@ -212,9 +212,9 @@ namespace MQTTnet.Server { var result = new List(); - lock (_clientSessions) + lock (_sessions) { - foreach (var session in _clientSessions.Values) + foreach (var session in _sessions.Values) { var sessionStatus = new MqttSessionStatus(session); result.Add(sessionStatus); @@ -263,18 +263,18 @@ namespace MQTTnet.Server public async Task DeleteSessionAsync(string clientId) { - MqttClientConnection connection; - MqttClientSession session; + MqttClient connection; + MqttSession session; - lock (_clientConnections) + lock (_clients) { - _clientConnections.TryGetValue(clientId, out connection); + _clients.TryGetValue(clientId, out connection); } - lock (_clientSessions) + lock (_sessions) { - _clientSessions.TryGetValue(clientId, out session); - _clientSessions.Remove(clientId); + _sessions.TryGetValue(clientId, out session); + _sessions.Remove(clientId); } try @@ -331,67 +331,16 @@ namespace MQTTnet.Server { try { - // MqttPendingApplicationMessage pendingApplicationMessage; - // try - // { - // pendingApplicationMessage = _messageQueue.Take(); - // } - // catch (ArgumentNullException) - // { - // return; - // } - // // catch (ThreadAbortException) - // // { - // // return; - // // } - // catch (ObjectDisposedException) - // { - // return; - // } - - //var clientConnection = pendingApplicationMessage.Sender; - //var senderClientId = clientConnection?.Id ?? _options.ClientId; - - // // TODO: Only use PUBLISH packets internally! - // var applicationMessage = new MqttApplicationMessageFactory().Create(pendingApplicationMessage.PublishPacket); - // - // var interceptor = _options.ApplicationMessageInterceptor; - // if (interceptor != null) - // { - // var interceptorContext = await InterceptApplicationMessageAsync(interceptor, clientConnection, applicationMessage).ConfigureAwait(false); - // if (interceptorContext != null) - // { - // if (interceptorContext.CloseConnection) - // { - // if (clientConnection != null) - // { - // await clientConnection.StopAsync(MqttClientDisconnectReason.NormalDisconnection).ConfigureAwait(false); - // } - // } - // - // if (interceptorContext.ApplicationMessage == null || !interceptorContext.ProcessPublish) - // { - // return; - // } - // - // applicationMessage = interceptorContext.ApplicationMessage; - // } - // } - - // await _eventContainer.ApplicationMessageReceivedEvent.InvokeAsync( - // () => new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage, null, null)) - // .ConfigureAwait(false); - if (applicationMessage.Retain) { - await _retainedMessagesManager.HandleMessageAsync(senderClientId, applicationMessage).ConfigureAwait(false); + await _retainedMessagesManager.UpdateMessage(senderClientId, applicationMessage).ConfigureAwait(false); } var deliveryCount = 0; - List sessions; - lock (_clientSessions) + List sessions; + lock (_sessions) { - sessions = _clientSessions.Values.ToList(); + sessions = _sessions.Values.ToList(); } foreach (var clientSession in sessions) @@ -434,7 +383,7 @@ namespace MQTTnet.Server if (deliveryCount == 0) { - await _eventContainer.ApplicationMessageNotConsumedEvent.InvokeAsync(() => new MqttApplicationMessageNotConsumedEventArgs + await _eventContainer.ApplicationMessageNotConsumedEvent.InvokeAsync(() => new ApplicationMessageNotConsumedEventArgs { ApplicationMessage = applicationMessage, SenderClientId = senderClientId @@ -447,14 +396,14 @@ namespace MQTTnet.Server } } - async Task ValidateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter) + async Task ValidateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter) { - var context = new ValidatingMqttClientConnectionEventArgs(connectPacket, channelAdapter) + var context = new ValidatingConnectionEventArgs(connectPacket, channelAdapter) { SessionItems = new ConcurrentDictionary() }; - await _eventContainer.ValidatingClientConnectionEvent.InvokeAsync(context).ConfigureAwait(false); + await _eventContainer.ValidatingConnectionEvent.InvokeAsync(context).ConfigureAwait(false); // Check the client ID and set a random one if supported. if (string.IsNullOrEmpty(connectPacket.ClientId) && channelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500) @@ -470,22 +419,21 @@ namespace MQTTnet.Server return context; } - async Task CreateClientConnection( + async Task CreateClientConnection( MqttConnectPacket connectPacket, MqttConnAckPacket connAckPacket, IMqttChannelAdapter channelAdapter, IDictionary sessionItems) { - MqttClientConnection connection; + MqttClient connection; using (await _createConnectionSyncRoot.WaitAsync(CancellationToken.None).ConfigureAwait(false)) { - MqttClientSession session; - lock (_clientSessions) + MqttSession session; + lock (_sessions) { - if (!_clientSessions.TryGetValue(connectPacket.ClientId, out session)) + if (!_sessions.TryGetValue(connectPacket.ClientId, out session)) { - _logger.Verbose("Created a new session for client '{0}'.", connectPacket.ClientId); session = CreateSession(connectPacket.ClientId, sessionItems); } else @@ -503,29 +451,38 @@ namespace MQTTnet.Server } } - _clientSessions[connectPacket.ClientId] = session; + _sessions[connectPacket.ClientId] = session; } - MqttClientConnection existingConnection; - - lock (_clientConnections) + if (!connAckPacket.IsSessionPresent) { - _clientConnections.TryGetValue(connectPacket.ClientId, out existingConnection); + var preparingSessionEventArgs = new PreparingSessionEventArgs(); + await _eventContainer.PreparingSessionEvent.InvokeAsync(preparingSessionEventArgs).ConfigureAwait(false); + + // TODO: Import subscriptions etc. + //session.SubscriptionsManager.Subscribe() + } + + MqttClient existing; + + lock (_clients) + { + _clients.TryGetValue(connectPacket.ClientId, out existing); connection = CreateConnection(connectPacket, channelAdapter, session); - _clientConnections[connectPacket.ClientId] = connection; + _clients[connectPacket.ClientId] = connection; } - if (existingConnection != null) + if (existing != null) { - existingConnection.IsTakenOver = true; - await existingConnection.StopAsync(MqttClientDisconnectReason.SessionTakenOver).ConfigureAwait(false); + existing.IsTakenOver = true; + await existing.StopAsync(MqttClientDisconnectReason.SessionTakenOver).ConfigureAwait(false); - await _eventContainer.ClientDisconnectedEvent.InvokeAsync(() => new MqttServerClientDisconnectedEventArgs + await _eventContainer.ClientDisconnectedEvent.InvokeAsync(() => new ClientDisconnectedEventArgs { - ClientId = existingConnection.Id, + ClientId = existing.Id, DisconnectType = MqttClientDisconnectType.Takeover, - Endpoint = existingConnection.Endpoint + Endpoint = existing.Endpoint }).ConfigureAwait(false); } } @@ -566,11 +523,11 @@ namespace MQTTnet.Server // return interceptorContext; // } - MqttClientSession GetClientSession(string clientId) + MqttSession GetClientSession(string clientId) { - lock (_clientSessions) + lock (_sessions) { - if (!_clientSessions.TryGetValue(clientId, out var session)) + if (!_sessions.TryGetValue(clientId, out var session)) { throw new InvalidOperationException($"Client session '{clientId}' is unknown."); } @@ -579,9 +536,9 @@ namespace MQTTnet.Server } } - MqttClientConnection CreateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, MqttClientSession session) + MqttClient CreateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, MqttSession session) { - return new MqttClientConnection( + return new MqttClient( connectPacket, channelAdapter, session, @@ -591,9 +548,11 @@ namespace MQTTnet.Server _rootLogger); } - MqttClientSession CreateSession(string clientId, IDictionary sessionItems) + MqttSession CreateSession(string clientId, IDictionary sessionItems) { - return new MqttClientSession( + _logger.Verbose("Created a new session for client '{0}'.", clientId); + + return new MqttSession( clientId, sessionItems, _options, @@ -606,12 +565,12 @@ namespace MQTTnet.Server { _createConnectionSyncRoot?.Dispose(); - foreach (var clientConnection in _clientConnections.Values) + foreach (var clientConnection in _clients.Values) { clientConnection.Dispose(); } - foreach (var session in _clientSessions.Values) + foreach (var session in _sessions.Values) { session.Dispose(); } diff --git a/Source/MQTTnet/Server/Internal/MqttClientConnectionStatistics.cs b/Source/MQTTnet/Server/Internal/MqttClientStatistics.cs similarity index 96% rename from Source/MQTTnet/Server/Internal/MqttClientConnectionStatistics.cs rename to Source/MQTTnet/Server/Internal/MqttClientStatistics.cs index 8491801d..e4e64b8e 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientConnectionStatistics.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientStatistics.cs @@ -4,7 +4,7 @@ using MQTTnet.Packets; namespace MQTTnet.Server { - public sealed class MqttClientConnectionStatistics + public sealed class MqttClientStatistics { // Start with 1 because the CONNACK packet is not counted here. long _receivedPacketsCount = 1; @@ -15,7 +15,7 @@ namespace MQTTnet.Server long _receivedApplicationMessagesCount; long _sentApplicationMessagesCount; - public MqttClientConnectionStatistics() + public MqttClientStatistics() { ConnectedTimestamp = DateTime.UtcNow; diff --git a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs index 013a5040..5b6e4179 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs @@ -16,29 +16,29 @@ namespace MQTTnet.Server readonly ReaderWriterLockSlim _subscriptionsLock = new ReaderWriterLockSlim(); readonly Dictionary _subscriptions = new Dictionary(4096); - readonly MqttClientSession _clientSession; + readonly MqttSession _session; readonly MqttServerOptions _options; readonly MqttServerEventContainer _eventContainer; - readonly IMqttRetainedMessagesManager _retainedMessagesManager; + readonly MqttRetainedMessagesManager _retainedMessagesManager; public MqttClientSubscriptionsManager( - MqttClientSession clientSession, + MqttSession session, MqttServerOptions serverOptions, MqttServerEventContainer eventContainer, - IMqttRetainedMessagesManager retainedMessagesManager) + MqttRetainedMessagesManager retainedMessagesManager) { - _clientSession = clientSession ?? throw new ArgumentNullException(nameof(clientSession)); + _session = session ?? throw new ArgumentNullException(nameof(session)); _options = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions)); _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer)); _retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager)); } - public async Task Subscribe(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) + public async Task Subscribe(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) { if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket)); - var retainedApplicationMessages = await _retainedMessagesManager.GetMessagesAsync().ConfigureAwait(false); - var result = new MqttSubscribeResult(); + var retainedApplicationMessages = await _retainedMessagesManager.GetMessages().ConfigureAwait(false); + var result = new SubscribeResult(); // The topic filters are order by its QoS so that the higher QoS will win over a // lower one. @@ -79,9 +79,9 @@ namespace MQTTnet.Server subscribePacket.Properties?.SubscriptionIdentifier ?? 0, interceptorContext.Response.ReasonCode); - await _eventContainer.ClientSubscribedTopicEvent.InvokeAsync(() => new MqttServerClientSubscribedTopicEventArgs + await _eventContainer.ClientSubscribedTopicEvent.InvokeAsync(() => new ClientSubscribedTopicEventArgs { - ClientId = _clientSession.Id, + ClientId = _session.Id, TopicFilter = finalTopicFilter }).ConfigureAwait(false); @@ -135,9 +135,9 @@ namespace MQTTnet.Server foreach (var topicFilter in unsubscribePacket.TopicFilters) { - await _eventContainer.ClientUnsubscribedTopicEvent.InvokeAsync(() => new MqttServerClientUnsubscribedTopicEventArgs + await _eventContainer.ClientUnsubscribedTopicEvent.InvokeAsync(() => new ClientUnsubscribedTopicEventArgs { - ClientId = _clientSession.Id, + ClientId = _session.Id, TopicFilter = topicFilter }).ConfigureAwait(false); } @@ -159,7 +159,7 @@ namespace MQTTnet.Server _subscriptionsLock.ExitReadLock(); } - var senderIsReceiver = string.Equals(senderClientId, _clientSession.Id); + var senderIsReceiver = string.Equals(senderClientId, _session.Id); var qosLevels = new HashSet(); var subscriptionIdentifiers = new HashSet(); @@ -262,7 +262,7 @@ namespace MQTTnet.Server } static void FilterRetainedApplicationMessages(IList retainedApplicationMessages, CreateSubscriptionResult createSubscriptionResult, - MqttSubscribeResult subscribeResult) + SubscribeResult subscribeResult) { for (var i = retainedApplicationMessages.Count - 1; i >= 0; i--) { @@ -294,14 +294,13 @@ namespace MQTTnet.Server var queuedApplicationMessage = new MqttQueuedApplicationMessage { ApplicationMessage = retainedApplicationMessage, - IsRetainedMessage = true, SubscriptionQualityOfServiceLevel = createSubscriptionResult.Subscription.GrantedQualityOfServiceLevel }; - if (createSubscriptionResult.Subscription.Identifier > 0) - { - queuedApplicationMessage.SubscriptionIdentifiers = new List {createSubscriptionResult.Subscription.Identifier}; - } + // if (createSubscriptionResult.Subscription.Identifier > 0) + // { + // queuedApplicationMessage.SubscriptionIdentifiers = new List {createSubscriptionResult.Subscription.Identifier}; + // } subscribeResult.RetainedApplicationMessages.Add(queuedApplicationMessage); @@ -309,14 +308,14 @@ namespace MQTTnet.Server } } - async Task InterceptSubscribe(MqttTopicFilter topicFilter, CancellationToken cancellationToken) + async Task InterceptSubscribe(MqttTopicFilter topicFilter, CancellationToken cancellationToken) { - var subscriptionReceivedEventArgs = new InterceptingMqttClientSubscriptionEventArgs + var subscriptionReceivedEventArgs = new InterceptingSubscriptionEventArgs { - ClientId = _clientSession.Id, + ClientId = _session.Id, TopicFilter = topicFilter, - SessionItems = _clientSession.Items, - Session = new MqttSessionStatus(_clientSession), + SessionItems = _session.Items, + Session = new MqttSessionStatus(_session), CancellationToken = cancellationToken }; @@ -339,19 +338,19 @@ namespace MQTTnet.Server } else { - await _eventContainer.InterceptingClientSubscriptionEvent.InvokeAsync(subscriptionReceivedEventArgs).ConfigureAwait(false); + await _eventContainer.InterceptingSubscriptionEvent.InvokeAsync(subscriptionReceivedEventArgs).ConfigureAwait(false); } return subscriptionReceivedEventArgs; } - async Task InterceptUnsubscribe(string topicFilter, MqttSubscription mqttSubscription, CancellationToken cancellationToken) + async Task InterceptUnsubscribe(string topicFilter, MqttSubscription mqttSubscription, CancellationToken cancellationToken) { - var clientUnsubscribingTopicEventArgs = new InterceptingMqttClientUnsubscriptionEventArgs + var clientUnsubscribingTopicEventArgs = new InterceptingUnsubscriptionEventArgs { - ClientId = _clientSession.Id, + ClientId = _session.Id, Topic = topicFilter, - SessionItems = _clientSession.Items, + SessionItems = _session.Items, CancellationToken = cancellationToken }; @@ -364,7 +363,7 @@ namespace MQTTnet.Server clientUnsubscribingTopicEventArgs.Response.ReasonCode = MqttUnsubscribeReasonCode.Success; } - await _eventContainer.InterceptingClientUnsubscriptionEvent.InvokeAsync(clientUnsubscribingTopicEventArgs).ConfigureAwait(false); + await _eventContainer.InterceptingUnsubscriptionEvent.InvokeAsync(clientUnsubscribingTopicEventArgs).ConfigureAwait(false); return clientUnsubscribingTopicEventArgs; } diff --git a/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs b/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs index 645b8f9a..6764ecb5 100644 --- a/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs @@ -9,45 +9,36 @@ using MQTTnet.Internal; namespace MQTTnet.Server { - public sealed class MqttRetainedMessagesManager : IMqttRetainedMessagesManager + public sealed class MqttRetainedMessagesManager { - readonly AsyncLock _storageAccessLock = new AsyncLock(); readonly Dictionary _messages = new Dictionary(4096); + readonly AsyncLock _storageAccessLock = new AsyncLock(); - MqttNetSourceLogger _logger; - MqttServerOptions _options; - - // TODO: Get rid of the logger here! - public Task Start(MqttServerOptions options, IMqttNetLogger logger) + readonly MqttServerEventContainer _eventContainer; + readonly MqttNetSourceLogger _logger; + + public MqttRetainedMessagesManager(MqttServerEventContainer eventContainer, IMqttNetLogger logger) { + _eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer)); + if (logger == null) throw new ArgumentNullException(nameof(logger)); _logger = logger.WithSource(nameof(MqttRetainedMessagesManager)); - - _options = options ?? throw new ArgumentNullException(nameof(options)); - return PlatformAbstractionLayer.CompletedTask; } - - public async Task LoadMessagesAsync() + + public async Task LoadMessages() { - if (_options.Storage == null) - { - return; - } - try { - var retainedMessages = await _options.Storage.LoadRetainedMessagesAsync().ConfigureAwait(false); - + var eventArgs = new LoadingRetainedMessagesEventArgs(); + await _eventContainer.LoadingRetainedMessagesEvent.InvokeAsync(eventArgs).ConfigureAwait(false); + lock (_messages) { _messages.Clear(); - if (retainedMessages != null) + foreach (var retainedMessage in eventArgs.StoredRetainedMessages ?? new List()) { - foreach (var retainedMessage in retainedMessages) - { - _messages[retainedMessage.Topic] = retainedMessage; - } + _messages[retainedMessage.Topic] = retainedMessage; } } } @@ -57,7 +48,7 @@ namespace MQTTnet.Server } } - public async Task HandleMessageAsync(string clientId, MqttApplicationMessage applicationMessage) + public async Task UpdateMessage(string clientId, MqttApplicationMessage applicationMessage) { if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); @@ -102,12 +93,16 @@ namespace MQTTnet.Server if (saveIsRequired) { - if (_options.Storage != null) + using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) { - using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) + var eventArgs = new RetainedMessageChangedEventArgs { - await _options.Storage.SaveRetainedMessagesAsync(messagesForSave).ConfigureAwait(false); - } + ClientId = clientId, + ChangedRetainedMessage = applicationMessage, + StoredRetainedMessages = messagesForSave + }; + + await _eventContainer.RetainedMessageChangedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); } } } @@ -117,7 +112,7 @@ namespace MQTTnet.Server } } - public Task> GetMessagesAsync() + public Task> GetMessages() { lock (_messages) { @@ -125,19 +120,16 @@ namespace MQTTnet.Server } } - public async Task ClearMessagesAsync() + public async Task ClearMessages() { lock (_messages) { _messages.Clear(); } - if (_options.Storage != null) + using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) { - using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) - { - await _options.Storage.SaveRetainedMessagesAsync(new List()).ConfigureAwait(false); - } + await _eventContainer.RetainedMessagesClearedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false); } } } diff --git a/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs b/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs index 8e137401..dbd2cae8 100644 --- a/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs +++ b/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs @@ -5,29 +5,31 @@ namespace MQTTnet.Server { public sealed class MqttServerEventContainer { - public AsyncEvent InterceptingClientSubscriptionEvent { get; } = new AsyncEvent(); + public AsyncEvent InterceptingSubscriptionEvent { get; } = new AsyncEvent(); - public AsyncEvent InterceptingClientUnsubscriptionEvent { get; } = new AsyncEvent(); + public AsyncEvent InterceptingUnsubscriptionEvent { get; } = new AsyncEvent(); - public AsyncEvent InterceptingClientPublishEvent { get; } = new AsyncEvent(); + public AsyncEvent InterceptingPublishEvent { get; } = new AsyncEvent(); - public AsyncEvent ValidatingClientConnectionEvent { get; } = new AsyncEvent(); + public AsyncEvent ValidatingConnectionEvent { get; } = new AsyncEvent(); - public AsyncEvent ClientConnectedEvent { get; } = new AsyncEvent(); + public AsyncEvent ClientConnectedEvent { get; } = new AsyncEvent(); - public AsyncEvent ClientDisconnectedEvent { get; } = new AsyncEvent(); + public AsyncEvent ClientDisconnectedEvent { get; } = new AsyncEvent(); - public AsyncEvent ClientSubscribedTopicEvent { get; } = new AsyncEvent(); + public AsyncEvent ClientSubscribedTopicEvent { get; } = new AsyncEvent(); - public AsyncEvent ClientUnsubscribedTopicEvent { get; } = new AsyncEvent(); + public AsyncEvent ClientUnsubscribedTopicEvent { get; } = new AsyncEvent(); - public AsyncEvent PreparingClientSessionEvent { get; } = new AsyncEvent(); + public AsyncEvent PreparingSessionEvent { get; } = new AsyncEvent(); - public AsyncEvent ApplicationMessageNotConsumedEvent { get; } = new AsyncEvent(); + public AsyncEvent ApplicationMessageNotConsumedEvent { get; } = new AsyncEvent(); - public AsyncEvent RetainedApplicationMessageChangedEvent { get; } = new AsyncEvent(); + public AsyncEvent RetainedMessageChangedEvent { get; } = new AsyncEvent(); - public AsyncEvent RetainedApplicationMessageClearedEvent { get; } = new AsyncEvent(); + public AsyncEvent LoadingRetainedMessagesEvent { get; } = new AsyncEvent(); + + public AsyncEvent RetainedMessagesClearedEvent { get; } = new AsyncEvent(); public AsyncEvent InterceptingInboundPacketEvent { get; } = new AsyncEvent(); diff --git a/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs b/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs index cd03c4b5..e164d429 100644 --- a/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs +++ b/Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs @@ -66,7 +66,7 @@ namespace MQTTnet.Server } } - void TryMaintainConnection(MqttClientConnection connection, DateTime now) + void TryMaintainConnection(MqttClient connection, DateTime now) { try { diff --git a/Source/MQTTnet/Server/Internal/MqttClientSession.cs b/Source/MQTTnet/Server/Internal/MqttSession.cs similarity index 96% rename from Source/MQTTnet/Server/Internal/MqttClientSession.cs rename to Source/MQTTnet/Server/Internal/MqttSession.cs index eb12600c..c82755f8 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSession.cs +++ b/Source/MQTTnet/Server/Internal/MqttSession.cs @@ -10,7 +10,7 @@ using MQTTnet.Protocol; namespace MQTTnet.Server { - public sealed class MqttClientSession : IDisposable + public sealed class MqttSession : IDisposable { readonly MqttPacketBus _packetBus = new MqttPacketBus(); @@ -19,11 +19,11 @@ namespace MQTTnet.Server readonly MqttServerOptions _serverOptions; readonly MqttClientSessionsManager _clientSessionsManager; - public MqttClientSession(string clientId, + public MqttSession(string clientId, IDictionary items, MqttServerOptions serverOptions, MqttServerEventContainer eventContainer, - IMqttRetainedMessagesManager retainedMessagesManager, + MqttRetainedMessagesManager retainedMessagesManager, MqttClientSessionsManager clientSessionsManager) { _serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions)); diff --git a/Source/MQTTnet/Server/Internal/MqttSubscribeResult.cs b/Source/MQTTnet/Server/Internal/SubscribeResult.cs similarity index 92% rename from Source/MQTTnet/Server/Internal/MqttSubscribeResult.cs rename to Source/MQTTnet/Server/Internal/SubscribeResult.cs index 76195220..fcd5e898 100644 --- a/Source/MQTTnet/Server/Internal/MqttSubscribeResult.cs +++ b/Source/MQTTnet/Server/Internal/SubscribeResult.cs @@ -3,7 +3,7 @@ using MQTTnet.Protocol; namespace MQTTnet.Server { - public sealed class MqttSubscribeResult + public sealed class SubscribeResult { public List ReturnCodes { get; } = new List(128); diff --git a/Source/MQTTnet/Server/MqttQueuedApplicationMessage.cs b/Source/MQTTnet/Server/MqttQueuedApplicationMessage.cs index b31ea848..90513799 100644 --- a/Source/MQTTnet/Server/MqttQueuedApplicationMessage.cs +++ b/Source/MQTTnet/Server/MqttQueuedApplicationMessage.cs @@ -6,14 +6,6 @@ namespace MQTTnet.Server public sealed class MqttQueuedApplicationMessage { public MqttApplicationMessage ApplicationMessage { get; set; } - - public string SenderClientId { get; set; } - - public bool IsRetainedMessage { get; set; } - - public List SubscriptionIdentifiers { get; set; } - - public bool IsDuplicate { get; set; } /// /// Gets or sets the subscription quality of service level. diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 341a2cef..c485a607 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -1,5 +1,4 @@ using MQTTnet.Adapter; -using MQTTnet.Exceptions; using MQTTnet.Protocol; using System; using System.Collections.Generic; @@ -24,7 +23,7 @@ namespace MQTTnet.Server readonly MqttNetSourceLogger _logger; MqttClientSessionsManager _clientSessionsManager; - IMqttRetainedMessagesManager _retainedMessagesManager; + MqttRetainedMessagesManager _retainedMessagesManager; MqttServerKeepAliveMonitor _keepAliveMonitor; CancellationTokenSource _cancellationTokenSource; @@ -52,19 +51,19 @@ namespace MQTTnet.Server remove => _eventContainer.StoppedEvent.RemoveHandler(value); } - public event Func ValidatingClientConnectionAsync + public event Func ValidatingConnectionAsync { - add => _eventContainer.ValidatingClientConnectionEvent.AddHandler(value); - remove => _eventContainer.ValidatingClientConnectionEvent.RemoveHandler(value); + add => _eventContainer.ValidatingConnectionEvent.AddHandler(value); + remove => _eventContainer.ValidatingConnectionEvent.RemoveHandler(value); } - public event Func ClientConnectedAsync + public event Func ClientConnectedAsync { add => _eventContainer.ClientConnectedEvent.AddHandler(value); remove => _eventContainer.ClientConnectedEvent.RemoveHandler(value); } - public event Func ClientDisconnectedAsync + public event Func ClientDisconnectedAsync { add => _eventContainer.ClientDisconnectedEvent.AddHandler(value); remove => _eventContainer.ClientDisconnectedEvent.RemoveHandler(value); @@ -82,63 +81,69 @@ namespace MQTTnet.Server remove => _eventContainer.InterceptingOutboundPacketEvent.RemoveHandler(value); } - public event Func InterceptingClientSubscriptionAsync + public event Func InterceptingSubscriptionAsync { - add => _eventContainer.InterceptingClientSubscriptionEvent.AddHandler(value); - remove => _eventContainer.InterceptingClientSubscriptionEvent.RemoveHandler(value); + add => _eventContainer.InterceptingSubscriptionEvent.AddHandler(value); + remove => _eventContainer.InterceptingSubscriptionEvent.RemoveHandler(value); } - public event Func InterceptingClientUnsubscriptionAsync + public event Func InterceptingUnsubscriptionAsync { - add => _eventContainer.InterceptingClientUnsubscriptionEvent.AddHandler(value); - remove => _eventContainer.InterceptingClientUnsubscriptionEvent.RemoveHandler(value); + add => _eventContainer.InterceptingUnsubscriptionEvent.AddHandler(value); + remove => _eventContainer.InterceptingUnsubscriptionEvent.RemoveHandler(value); } - public event Func InterceptingClientPublishAsync + public event Func InterceptingPublishAsync { - add => _eventContainer.InterceptingClientPublishEvent.AddHandler(value); - remove => _eventContainer.InterceptingClientPublishEvent.RemoveHandler(value); + add => _eventContainer.InterceptingPublishEvent.AddHandler(value); + remove => _eventContainer.InterceptingPublishEvent.RemoveHandler(value); } - public event Func ClientSubscribedTopicAsync + public event Func ClientSubscribedTopicAsync { add => _eventContainer.ClientSubscribedTopicEvent.AddHandler(value); remove => _eventContainer.ClientSubscribedTopicEvent.RemoveHandler(value); } - public event Func ClientUnsubscribedTopicAsync + public event Func ClientUnsubscribedTopicAsync { add => _eventContainer.ClientUnsubscribedTopicEvent.AddHandler(value); remove => _eventContainer.ClientUnsubscribedTopicEvent.RemoveHandler(value); } - public event Func PreparingClientSessionAsync + public event Func PreparingSessionAsync { - add => _eventContainer.PreparingClientSessionEvent.AddHandler(value); - remove => _eventContainer.PreparingClientSessionEvent.RemoveHandler(value); + add => _eventContainer.PreparingSessionEvent.AddHandler(value); + remove => _eventContainer.PreparingSessionEvent.RemoveHandler(value); } - public event Func ApplicationMessageNotConsumedAsync + public event Func ApplicationMessageNotConsumedAsync { add => _eventContainer.ApplicationMessageNotConsumedEvent.AddHandler(value); remove => _eventContainer.ApplicationMessageNotConsumedEvent.RemoveHandler(value); } - public event Func RetainedApplicationMessageChangedAsync + public event Func RetainedMessageChangedAsync { - add => _eventContainer.RetainedApplicationMessageChangedEvent.AddHandler(value); - remove => _eventContainer.RetainedApplicationMessageChangedEvent.RemoveHandler(value); + add => _eventContainer.RetainedMessageChangedEvent.AddHandler(value); + remove => _eventContainer.RetainedMessageChangedEvent.RemoveHandler(value); } - public event Func RetainedApplicationMessagesClearedAsync + public event Func LoadingRetainedMessageAsync { - add => _eventContainer.RetainedApplicationMessageClearedEvent.AddHandler(value); - remove => _eventContainer.RetainedApplicationMessageClearedEvent.RemoveHandler(value); + add => _eventContainer.LoadingRetainedMessagesEvent.AddHandler(value); + remove => _eventContainer.LoadingRetainedMessagesEvent.RemoveHandler(value); + } + + public event Func RetainedMessagesClearedAsync + { + add => _eventContainer.RetainedMessagesClearedEvent.AddHandler(value); + remove => _eventContainer.RetainedMessagesClearedEvent.RemoveHandler(value); } public bool IsStarted => _cancellationTokenSource != null; - public Task> GetClientStatusAsync() + public Task> GetClientsAsync() { ThrowIfDisposed(); ThrowIfNotStarted(); @@ -146,7 +151,7 @@ namespace MQTTnet.Server return _clientSessionsManager.GetClientStatusAsync(); } - public Task> GetSessionStatusAsync() + public Task> GetSessionsAsync() { ThrowIfDisposed(); ThrowIfNotStarted(); @@ -154,20 +159,44 @@ namespace MQTTnet.Server return _clientSessionsManager.GetSessionStatusAsync(); } - public Task> GetRetainedApplicationMessagesAsync() + /// + /// Gets all currently cached retained messages. + /// + /// A list of retained messages. + public Task> GetRetainedMessagesAsync() { ThrowIfDisposed(); ThrowIfNotStarted(); - return _retainedMessagesManager.GetMessagesAsync(); + return _retainedMessagesManager.GetMessages(); } - public Task ClearRetainedApplicationMessagesAsync() + /// + /// Deletes all retained messages. + /// + /// A task. + public Task DeleteRetainedMessagesAsync() { ThrowIfDisposed(); ThrowIfNotStarted(); - return _retainedMessagesManager?.ClearMessagesAsync() ?? PlatformAbstractionLayer.CompletedTask; + return _retainedMessagesManager?.ClearMessages() ?? PlatformAbstractionLayer.CompletedTask; + } + + /// + /// Updates the specified retained message. + /// Setting the payload to null or empty will delete the retained message. + /// + /// The retained message. + /// A task. + public Task UpdateRetainedMessageAsync(MqttApplicationMessage retainedMessage) + { + if (retainedMessage == null) throw new ArgumentNullException(nameof(retainedMessage)); + + ThrowIfDisposed(); + ThrowIfNotStarted(); + + return _retainedMessagesManager?.UpdateMessage(null, retainedMessage); } public Task SubscribeAsync(string clientId, ICollection topicFilters) @@ -220,10 +249,8 @@ namespace MQTTnet.Server _cancellationTokenSource = new CancellationTokenSource(); var cancellationToken = _cancellationTokenSource.Token; - _retainedMessagesManager = _options.RetainedMessagesManager ?? throw new MqttConfigurationException("options.RetainedMessagesManager should not be null."); - - await _retainedMessagesManager.Start(_options, _rootLogger).ConfigureAwait(false); - await _retainedMessagesManager.LoadMessagesAsync().ConfigureAwait(false); + _retainedMessagesManager = new MqttRetainedMessagesManager(_eventContainer, _rootLogger); + await _retainedMessagesManager.LoadMessages().ConfigureAwait(false); _clientSessionsManager = new MqttClientSessionsManager(_options, _retainedMessagesManager, _eventContainer, _rootLogger); diff --git a/Source/MQTTnet/Server/MqttServerExtensions.cs b/Source/MQTTnet/Server/MqttServerExtensions.cs index 0fcc01e3..d1debd06 100644 --- a/Source/MQTTnet/Server/MqttServerExtensions.cs +++ b/Source/MQTTnet/Server/MqttServerExtensions.cs @@ -16,16 +16,7 @@ namespace MQTTnet.Server return server.SubscribeAsync(clientId, topicFilters); } - - public static Task SubscribeAsync(this MqttServer server, string clientId, string topic, MqttQualityOfServiceLevel qualityOfServiceLevel) - { - if (server == null) throw new ArgumentNullException(nameof(server)); - if (clientId == null) throw new ArgumentNullException(nameof(clientId)); - if (topic == null) throw new ArgumentNullException(nameof(topic)); - - return server.SubscribeAsync(clientId, new MqttTopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build()); - } - + public static Task SubscribeAsync(this MqttServer server, string clientId, string topic) { if (server == null) throw new ArgumentNullException(nameof(server)); @@ -34,102 +25,5 @@ namespace MQTTnet.Server return server.SubscribeAsync(clientId, new MqttTopicFilterBuilder().WithTopic(topic).Build()); } - - public static Task UnsubscribeAsync(this MqttServer server, string clientId, params string[] topicFilters) - { - if (server == null) throw new ArgumentNullException(nameof(server)); - if (clientId == null) throw new ArgumentNullException(nameof(clientId)); - if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); - - return server.UnsubscribeAsync(clientId, topicFilters); - } - - // public static async Task PublishAsync(this IMqttServer server, IEnumerable applicationMessages) - // { - // if (server == null) throw new ArgumentNullException(nameof(server)); - // if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages)); - // - // foreach (var applicationMessage in applicationMessages) - // { - // await server.PublishAsync(applicationMessage).ConfigureAwait(false); - // } - // } - - // public static Task PublishAsync(this IMqttServer server, MqttApplicationMessage applicationMessage) - // { - // if (server == null) throw new ArgumentNullException(nameof(server)); - // if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - // - // return server.PublishAsync(applicationMessage, CancellationToken.None); - // } - // - // public static async Task PublishAsync(this IMqttServer server, params MqttApplicationMessage[] applicationMessages) - // { - // if (server == null) throw new ArgumentNullException(nameof(server)); - // if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages)); - // - // foreach (var applicationMessage in applicationMessages) - // { - // await server.PublishAsync(applicationMessage, CancellationToken.None).ConfigureAwait(false); - // } - // } - - // public static Task PublishAsync(this IMqttServer server, string topic) - // { - // if (server == null) throw new ArgumentNullException(nameof(server)); - // if (topic == null) throw new ArgumentNullException(nameof(topic)); - // - // return server.PublishAsync(builder => builder - // .WithTopic(topic)); - // } - - // public static Task PublishAsync(this IMqttServer server, string topic, string payload) - // { - // if (server == null) throw new ArgumentNullException(nameof(server)); - // if (topic == null) throw new ArgumentNullException(nameof(topic)); - // - // return server.PublishAsync(builder => builder - // .WithTopic(topic) - // .WithPayload(payload)); - // } - - // public static Task PublishAsync(this IMqttServer server, string topic, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel) - // { - // if (server == null) throw new ArgumentNullException(nameof(server)); - // if (topic == null) throw new ArgumentNullException(nameof(topic)); - // - // return server.PublishAsync(builder => builder - // .WithTopic(topic) - // .WithPayload(payload) - // .WithQualityOfServiceLevel(qualityOfServiceLevel)); - // } - - // public static Task PublishAsync(this IMqttServer server, string topic, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel, bool retain) - // { - // if (server == null) throw new ArgumentNullException(nameof(server)); - // if (topic == null) throw new ArgumentNullException(nameof(topic)); - // - // return server.PublishAsync(builder => builder - // .WithTopic(topic) - // .WithPayload(payload) - // .WithQualityOfServiceLevel(qualityOfServiceLevel) - // .WithRetainFlag(retain)); - // } - - // public static Task PublishAsync(this IMqttServer server, Func builder, CancellationToken cancellationToken) - // { - // if (server == null) throw new ArgumentNullException(nameof(server)); - // - // var message = builder(new MqttApplicationMessageBuilder()).Build(); - // return server.PublishAsync(message, cancellationToken); - // } - // - // public static Task PublishAsync(this IMqttServer server, Func builder) - // { - // if (server == null) throw new ArgumentNullException(nameof(server)); - // - // var message = builder(new MqttApplicationMessageBuilder()).Build(); - // return server.PublishAsync(message, CancellationToken.None); - // } } } diff --git a/Source/MQTTnet/Server/Options/MqttServerOptions.cs b/Source/MQTTnet/Server/Options/MqttServerOptions.cs index 38a781bc..58553fb9 100644 --- a/Source/MQTTnet/Server/Options/MqttServerOptions.cs +++ b/Source/MQTTnet/Server/Options/MqttServerOptions.cs @@ -17,9 +17,5 @@ namespace MQTTnet.Server public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(15); public TimeSpan KeepAliveMonitorInterval { get; set; } = TimeSpan.FromMilliseconds(500); - - public IMqttServerStorage Storage { get; set; } - - public IMqttRetainedMessagesManager RetainedMessagesManager { get; set; } = new MqttRetainedMessagesManager(); } } diff --git a/Source/MQTTnet/Server/Options/MqttServerOptionsBuilder.cs b/Source/MQTTnet/Server/Options/MqttServerOptionsBuilder.cs index 9aec4f89..c832ac90 100644 --- a/Source/MQTTnet/Server/Options/MqttServerOptionsBuilder.cs +++ b/Source/MQTTnet/Server/Options/MqttServerOptionsBuilder.cs @@ -140,18 +140,6 @@ namespace MQTTnet.Server return this; } #endif - - public MqttServerOptionsBuilder WithStorage(IMqttServerStorage value) - { - _options.Storage = value; - return this; - } - - public MqttServerOptionsBuilder WithRetainedMessagesManager(IMqttRetainedMessagesManager value) - { - _options.RetainedMessagesManager = value; - return this; - } // public MqttServerOptionsBuilder WithApplicationMessageInterceptor(IMqttServerApplicationMessageInterceptor value) // { diff --git a/Source/MQTTnet/Server/MqttPublishResponse.cs b/Source/MQTTnet/Server/PublishResponse.cs similarity index 87% rename from Source/MQTTnet/Server/MqttPublishResponse.cs rename to Source/MQTTnet/Server/PublishResponse.cs index 5647bbc7..bc94e34a 100644 --- a/Source/MQTTnet/Server/MqttPublishResponse.cs +++ b/Source/MQTTnet/Server/PublishResponse.cs @@ -3,7 +3,7 @@ using MQTTnet.Packets; namespace MQTTnet.Server { - public sealed class MqttApplicationMessageResponse + public sealed class PublishResponse { public MqttApplicationMessageResponseReasonCode ReasonCode { get; set; } = MqttApplicationMessageResponseReasonCode.Success; diff --git a/Source/MQTTnet/Server/Status/MqttClientStatus.cs b/Source/MQTTnet/Server/Status/MqttClientStatus.cs index fca72fc8..03c5a487 100644 --- a/Source/MQTTnet/Server/Status/MqttClientStatus.cs +++ b/Source/MQTTnet/Server/Status/MqttClientStatus.cs @@ -7,9 +7,9 @@ namespace MQTTnet.Server { public sealed class MqttClientStatus : IMqttClientStatus { - readonly MqttClientConnection _connection; + readonly MqttClient _connection; - public MqttClientStatus(MqttClientConnection connection) + public MqttClientStatus(MqttClient connection) { _connection = connection ?? throw new ArgumentNullException(nameof(connection)); } diff --git a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs index ac2e205d..c7a5d990 100644 --- a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs +++ b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs @@ -9,9 +9,9 @@ namespace MQTTnet.Server { public sealed class MqttSessionStatus : IMqttSessionStatus { - readonly MqttClientSession _session; + readonly MqttSession _session; - public MqttSessionStatus(MqttClientSession session) + public MqttSessionStatus(MqttSession session) { _session = session ?? throw new ArgumentNullException(nameof(session)); } diff --git a/Source/MQTTnet/Server/Storage/IMqttRetainedMessagesManager.cs b/Source/MQTTnet/Server/Storage/IMqttRetainedMessagesManager.cs deleted file mode 100644 index 7794aff8..00000000 --- a/Source/MQTTnet/Server/Storage/IMqttRetainedMessagesManager.cs +++ /dev/null @@ -1,19 +0,0 @@ -using System.Collections.Generic; -using System.Threading.Tasks; -using MQTTnet.Diagnostics; - -namespace MQTTnet.Server -{ - public interface IMqttRetainedMessagesManager - { - Task Start(MqttServerOptions options, IMqttNetLogger logger); - - Task LoadMessagesAsync(); - - Task ClearMessagesAsync(); - - Task HandleMessageAsync(string clientId, MqttApplicationMessage applicationMessage); - - Task> GetMessagesAsync(); - } -} diff --git a/Source/MQTTnet/Server/Storage/IMqttServerPersistedSession.cs b/Source/MQTTnet/Server/Storage/IMqttServerPersistedSession.cs deleted file mode 100644 index b68ef81a..00000000 --- a/Source/MQTTnet/Server/Storage/IMqttServerPersistedSession.cs +++ /dev/null @@ -1,35 +0,0 @@ -using System; -using System.Collections.Generic; -using MQTTnet.Packets; - -namespace MQTTnet.Server -{ - public interface IMqttServerPersistedSession - { - /// - /// Gets the client identifier. - /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. - /// - string ClientId { get; } - - IDictionary Items { get; } - - IList Subscriptions { get; } - - /// - /// Gets the last will message. - /// In MQTT, you use the last will message feature to notify other clients about an ungracefully disconnected client. - /// - MqttApplicationMessage WillMessage { get; } - - /// - /// Gets the will delay interval. - /// This is the time between the client disconnect and the time the will message will be sent. - /// - uint? WillDelayInterval { get; } - - DateTime? SessionExpiryTimestamp { get; } - - IList PendingApplicationMessages { get; } - } -} diff --git a/Source/MQTTnet/Server/Storage/IMqttServerPersistedSessionsStorage.cs b/Source/MQTTnet/Server/Storage/IMqttServerPersistedSessionsStorage.cs deleted file mode 100644 index fa2a4cb7..00000000 --- a/Source/MQTTnet/Server/Storage/IMqttServerPersistedSessionsStorage.cs +++ /dev/null @@ -1,11 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Threading.Tasks; - -namespace MQTTnet.Server -{ - public interface IMqttServerPersistedSessionsStorage - { - Task> LoadPersistedSessionsAsync(); - } -} diff --git a/Source/MQTTnet/Server/Storage/IMqttServerStorage.cs b/Source/MQTTnet/Server/Storage/IMqttServerStorage.cs deleted file mode 100644 index da6e6e28..00000000 --- a/Source/MQTTnet/Server/Storage/IMqttServerStorage.cs +++ /dev/null @@ -1,12 +0,0 @@ -using System.Collections.Generic; -using System.Threading.Tasks; - -namespace MQTTnet.Server -{ - public interface IMqttServerStorage - { - Task SaveRetainedMessagesAsync(IList messages); - - Task> LoadRetainedMessagesAsync(); - } -} diff --git a/Source/MQTTnet/Server/MqttSubscribeResponse.cs b/Source/MQTTnet/Server/SubscribeResponse.cs similarity index 92% rename from Source/MQTTnet/Server/MqttSubscribeResponse.cs rename to Source/MQTTnet/Server/SubscribeResponse.cs index d0b31010..6c3fcebb 100644 --- a/Source/MQTTnet/Server/MqttSubscribeResponse.cs +++ b/Source/MQTTnet/Server/SubscribeResponse.cs @@ -4,7 +4,7 @@ using MQTTnet.Protocol; namespace MQTTnet.Server { - public sealed class MqttSubscribeResponse + public sealed class SubscribeResponse { /// /// Gets or sets the reason code which is sent to the client. diff --git a/Source/MQTTnet/Server/MqttUnsubscribeResponse.cs b/Source/MQTTnet/Server/UnsubscribeResponse.cs similarity index 90% rename from Source/MQTTnet/Server/MqttUnsubscribeResponse.cs rename to Source/MQTTnet/Server/UnsubscribeResponse.cs index 6735cd6f..ac497f5e 100644 --- a/Source/MQTTnet/Server/MqttUnsubscribeResponse.cs +++ b/Source/MQTTnet/Server/UnsubscribeResponse.cs @@ -4,7 +4,7 @@ using MQTTnet.Protocol; namespace MQTTnet.Server { - public sealed class MqttUnsubscribeResponse + public sealed class UnsubscribeResponse { /// /// Gets or sets the reason code which is sent to the client. diff --git a/Tests/MQTTnet.Core.Tests/Client/ManagedMqttClient_Tests.cs b/Tests/MQTTnet.Core.Tests/Client/ManagedMqttClient_Tests.cs index 4c06a4ba..b653a4bc 100644 --- a/Tests/MQTTnet.Core.Tests/Client/ManagedMqttClient_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Client/ManagedMqttClient_Tests.cs @@ -110,7 +110,7 @@ namespace MQTTnet.Tests.Client await Task.Delay(500); - Assert.AreEqual(0, (await server.GetClientStatusAsync()).Count); + Assert.AreEqual(0, (await server.GetClientsAsync()).Count); } } diff --git a/Tests/MQTTnet.Core.Tests/Client/MqttClient_Tests.cs b/Tests/MQTTnet.Core.Tests/Client/MqttClient_Tests.cs index 2e81721c..77f183cf 100644 --- a/Tests/MQTTnet.Core.Tests/Client/MqttClient_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Client/MqttClient_Tests.cs @@ -27,7 +27,7 @@ namespace MQTTnet.Tests.Client var client = await testEnvironment.ConnectLowLevelClient(); var i = 0; - server.InterceptingClientPublishAsync += c => + server.InterceptingPublishAsync += c => { i++; return Task.CompletedTask; @@ -766,8 +766,8 @@ namespace MQTTnet.Tests.Client await Task.Delay(500); - var clientStatus = await testEnvironment.Server.GetClientStatusAsync(); - var sessionStatus = await testEnvironment.Server.GetSessionStatusAsync(); + var clientStatus = await testEnvironment.Server.GetClientsAsync(); + var sessionStatus = await testEnvironment.Server.GetSessionsAsync(); for (var i = 0; i < 98; i++) { diff --git a/Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs b/Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs index b815b47b..c9485900 100644 --- a/Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs @@ -1,17 +1,14 @@ using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; using MQTTnet.Formatter; -using MQTTnet.Tests.Mockups; using System.Threading; using System.Threading.Tasks; namespace MQTTnet.Tests.MQTTv5 { [TestClass] - public class Server_Tests : BaseTestClass + public sealed class Server_Tests : BaseTestClass { - public TestContext TestContext { get; set; } - [TestMethod] public async Task Will_Message_Send() { @@ -46,7 +43,7 @@ namespace MQTTnet.Tests.MQTTv5 [TestMethod] public async Task Validate_IsSessionPresent() { - using (var testEnvironment = new TestEnvironment(TestContext)) + using (var testEnvironment = CreateTestEnvironment()) { // Create server with persistent sessions enabled @@ -84,7 +81,6 @@ namespace MQTTnet.Tests.MQTTv5 .WithClientId(ClientId) .Build(); - var result = await client2.ConnectAsync(options).ConfigureAwait(false); await client2.DisconnectAsync(); diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs index d39c7819..c0a93548 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs @@ -181,7 +181,7 @@ namespace MQTTnet.Tests.Mockups Server = Factory.CreateMqttServer(options, ServerLogger); - Server.ValidatingClientConnectionAsync += e => + Server.ValidatingConnectionAsync += e => { if (TestContext != null) { diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestServerStorage.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestServerStorage.cs deleted file mode 100644 index 55ca6bb1..00000000 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestServerStorage.cs +++ /dev/null @@ -1,22 +0,0 @@ -using System.Collections.Generic; -using System.Threading.Tasks; -using MQTTnet.Server; - -namespace MQTTnet.Tests.Mockups -{ - public class TestServerStorage : IMqttServerStorage - { - public IList Messages = new List(); - - public Task SaveRetainedMessagesAsync(IList messages) - { - Messages = messages; - return Task.CompletedTask; - } - - public Task> LoadRetainedMessagesAsync() - { - return Task.FromResult(Messages); - } - } -} diff --git a/Tests/MQTTnet.Core.Tests/Server/Assigned_Client_ID_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Assigned_Client_ID_Tests.cs index 1e31a7de..f46cae78 100644 --- a/Tests/MQTTnet.Core.Tests/Server/Assigned_Client_ID_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/Assigned_Client_ID_Tests.cs @@ -34,7 +34,7 @@ namespace MQTTnet.Tests.Server var disconnectedMre = new ManualResetEventSlim(); var server = await testEnvironment.StartServer(); - server.ValidatingClientConnectionAsync += e => + server.ValidatingConnectionAsync += e => { if (string.IsNullOrEmpty(e.ClientId)) { diff --git a/Tests/MQTTnet.Core.Tests/Server/Events_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Events_Tests.cs index 95a917dc..07668a61 100644 --- a/Tests/MQTTnet.Core.Tests/Server/Events_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/Events_Tests.cs @@ -18,7 +18,7 @@ namespace MQTTnet.Tests.Server { var server = await testEnvironment.StartServer(); - MqttServerClientConnectedEventArgs eventArgs = null; + ClientConnectedEventArgs eventArgs = null; server.ClientConnectedAsync += e => { eventArgs = e; @@ -45,7 +45,7 @@ namespace MQTTnet.Tests.Server { var server = await testEnvironment.StartServer(); - MqttServerClientDisconnectedEventArgs eventArgs = null; + ClientDisconnectedEventArgs eventArgs = null; server.ClientDisconnectedAsync += e => { eventArgs = e; @@ -72,7 +72,7 @@ namespace MQTTnet.Tests.Server { var server = await testEnvironment.StartServer(); - MqttServerClientSubscribedTopicEventArgs eventArgs = null; + ClientSubscribedTopicEventArgs eventArgs = null; server.ClientSubscribedTopicAsync += e => { eventArgs = e; @@ -99,7 +99,7 @@ namespace MQTTnet.Tests.Server { var server = await testEnvironment.StartServer(); - MqttServerClientUnsubscribedTopicEventArgs eventArgs = null; + ClientUnsubscribedTopicEventArgs eventArgs = null; server.ClientUnsubscribedTopicAsync += e => { eventArgs = e; @@ -125,8 +125,8 @@ namespace MQTTnet.Tests.Server { var server = await testEnvironment.StartServer(); - InterceptingMqttClientPublishEventArgs eventArgs = null; - server.InterceptingClientPublishAsync += e => + InterceptingPublishEventArgs eventArgs = null; + server.InterceptingPublishAsync += e => { eventArgs = e; return Task.CompletedTask; diff --git a/Tests/MQTTnet.Core.Tests/Server/General.cs b/Tests/MQTTnet.Core.Tests/Server/General.cs index e2e4a5ad..f597b184 100644 --- a/Tests/MQTTnet.Core.Tests/Server/General.cs +++ b/Tests/MQTTnet.Core.Tests/Server/General.cs @@ -512,7 +512,7 @@ namespace MQTTnet.Tests.Server await Task.Delay(1000); - var retainedMessages = await server.GetRetainedApplicationMessagesAsync(); + var retainedMessages = await server.GetRetainedMessagesAsync(); Assert.AreEqual(ClientCount, retainedMessages.Count); @@ -530,7 +530,7 @@ namespace MQTTnet.Tests.Server { var server = await testEnvironment.StartServer(); - server.InterceptingClientPublishAsync += e => + server.InterceptingPublishAsync += e => { e.ApplicationMessage = new MqttApplicationMessage {Topic = "new_topic"}; @@ -557,19 +557,24 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task Persist_Retained_Message() { - var serverStorage = new TestServerStorage(); - using (var testEnvironment = CreateTestEnvironment()) { - await testEnvironment.StartServer(new MqttServerOptionsBuilder().WithStorage(serverStorage)); + List savedRetainedMessages = null; + + var s = await testEnvironment.StartServer(); + s.RetainedMessageChangedAsync += e => + { + savedRetainedMessages = e.StoredRetainedMessages; + return Task.CompletedTask; + }; var c1 = await testEnvironment.ConnectClient(); - + await c1.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("retained").WithPayload(new byte[3]).WithRetainFlag().Build()); await Task.Delay(500); - Assert.AreEqual(1, serverStorage.Messages.Count); + Assert.AreEqual(1, savedRetainedMessages?.Count); } } @@ -617,7 +622,7 @@ namespace MQTTnet.Tests.Server using (var testEnvironment = CreateTestEnvironment()) { var server = await testEnvironment.StartServer(); - server.InterceptingClientPublishAsync += e => + server.InterceptingPublishAsync += e => { e.ApplicationMessage.Payload = Encoding.ASCII.GetBytes("extended"); return Task.CompletedTask; @@ -698,7 +703,7 @@ namespace MQTTnet.Tests.Server var server = await testEnvironment.StartServer(); - server.ValidatingClientConnectionAsync += e => + server.ValidatingConnectionAsync += e => { e.ReasonCode = MqttConnectReasonCode.NotAuthorized; return Task.CompletedTask; @@ -711,7 +716,7 @@ namespace MQTTnet.Tests.Server Dictionary _connected; - void ConnectionValidationHandler(ValidatingMqttClientConnectionEventArgs eventArgs) + void ConnectionValidationHandler(ValidatingConnectionEventArgs eventArgs) { if (_connected.ContainsKey(eventArgs.ClientId)) { @@ -734,7 +739,7 @@ namespace MQTTnet.Tests.Server var server = await testEnvironment.StartServer(); - server.ValidatingClientConnectionAsync += e => + server.ValidatingConnectionAsync += e => { ConnectionValidationHandler(e); return Task.CompletedTask; @@ -923,12 +928,12 @@ namespace MQTTnet.Tests.Server var clientOptions = new MqttClientOptionsBuilder(); var c1 = await testEnvironment.ConnectClient(clientOptions); await Task.Delay(500); - Assert.AreEqual(1, (await server.GetClientStatusAsync()).Count); + Assert.AreEqual(1, (await server.GetClientsAsync()).Count); await c1.DisconnectAsync(); await Task.Delay(500); - Assert.AreEqual(0, (await server.GetClientStatusAsync()).Count); + Assert.AreEqual(0, (await server.GetClientsAsync()).Count); } } @@ -965,7 +970,7 @@ namespace MQTTnet.Tests.Server { var server = await testEnvironment.StartServer(); - server.InterceptingClientSubscriptionAsync += e => + server.InterceptingSubscriptionAsync += e => { // This should lead to no subscriptions for "n" at all. So also no sending of retained messages. if (e.TopicFilter.Topic == "n") @@ -1021,7 +1026,7 @@ namespace MQTTnet.Tests.Server await client1.DisconnectAsync(); await Task.Delay(500); - var clientStatus = await server.GetClientStatusAsync(); + var clientStatus = await server.GetClientsAsync(); Assert.AreEqual(0, clientStatus.Count); var client2 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithClientId("b")); @@ -1032,8 +1037,8 @@ namespace MQTTnet.Tests.Server await Task.Delay(500); - clientStatus = await server.GetClientStatusAsync(); - var sessionStatus = await server.GetSessionStatusAsync(); + clientStatus = await server.GetClientsAsync(); + var sessionStatus = await server.GetSessionsAsync(); Assert.AreEqual(0, clientStatus.Count); Assert.AreEqual(2, sessionStatus.Count); diff --git a/Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs index 05ba4f2a..570e4999 100644 --- a/Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/MqttSubscriptionsManager_Tests.cs @@ -2,6 +2,7 @@ using System.Threading; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Diagnostics; using MQTTnet.Packets; using MQTTnet.Protocol; using MQTTnet.Server; @@ -12,19 +13,37 @@ namespace MQTTnet.Tests.Server [TestClass] public sealed class MqttSubscriptionsManager_Tests : BaseTestClass { + MqttClientSubscriptionsManager _subscriptionsManager; + + [TestInitialize] + public void TestInitialize() + { + var logger = new TestLogger(); + var options = new MqttServerOptions(); + var retainedMessagesManager = new MqttRetainedMessagesManager(new MqttServerEventContainer(), logger); + var eventContainer = new MqttServerEventContainer(); + + var session = new MqttSession( + "", + new ConcurrentDictionary(), + options, + eventContainer, + retainedMessagesManager, + new MqttClientSessionsManager(options, retainedMessagesManager, eventContainer, logger)); + + _subscriptionsManager = new MqttClientSubscriptionsManager(session, new MqttServerOptions(), new MqttServerEventContainer(), + new MqttRetainedMessagesManager(new MqttServerEventContainer(), new MqttNetNullLogger())); + } + [TestMethod] public async Task MqttSubscriptionsManager_SubscribeSingleSuccess() { - var s = CreateSession(); - - var sm = new MqttClientSubscriptionsManager(s, new MqttServerOptions(), new MqttServerEventContainer(), new MqttRetainedMessagesManager()); - var sp = new MqttSubscribePacket(); sp.TopicFilters.Add(new MqttTopicFilterBuilder().WithTopic("A/B/C").Build()); - await sm.Subscribe(sp, CancellationToken.None); + await _subscriptionsManager.Subscribe(sp, CancellationToken.None); - var result = sm.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.AtMostOnce, ""); + var result = _subscriptionsManager.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.AtMostOnce, ""); Assert.IsTrue(result.IsSubscribed); Assert.AreEqual(result.QualityOfServiceLevel, MqttQualityOfServiceLevel.AtMostOnce); } @@ -32,16 +51,12 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task MqttSubscriptionsManager_SubscribeDifferentQoSSuccess() { - var s = CreateSession(); - - var sm = new MqttClientSubscriptionsManager(s, new MqttServerOptions(), new MqttServerEventContainer(), new MqttRetainedMessagesManager()); - var sp = new MqttSubscribePacket(); - sp.TopicFilters.Add(new MqttTopicFilter { Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce }); + sp.TopicFilters.Add(new MqttTopicFilter {Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce}); - await sm.Subscribe(sp, CancellationToken.None); + await _subscriptionsManager.Subscribe(sp, CancellationToken.None); - var result = sm.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.ExactlyOnce, ""); + var result = _subscriptionsManager.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.ExactlyOnce, ""); Assert.IsTrue(result.IsSubscribed); Assert.AreEqual(result.QualityOfServiceLevel, MqttQualityOfServiceLevel.AtMostOnce); } @@ -49,17 +64,13 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task MqttSubscriptionsManager_SubscribeTwoTimesSuccess() { - var s = CreateSession(); - - var sm = new MqttClientSubscriptionsManager(s, new MqttServerOptions(), new MqttServerEventContainer(), new MqttRetainedMessagesManager()); - var sp = new MqttSubscribePacket(); - sp.TopicFilters.Add(new MqttTopicFilter { Topic = "#", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce }); - sp.TopicFilters.Add(new MqttTopicFilter { Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce }); + sp.TopicFilters.Add(new MqttTopicFilter {Topic = "#", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce}); + sp.TopicFilters.Add(new MqttTopicFilter {Topic = "A/B/C", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce}); - await sm.Subscribe(sp, CancellationToken.None); + await _subscriptionsManager.Subscribe(sp, CancellationToken.None); - var result = sm.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.ExactlyOnce, ""); + var result = _subscriptionsManager.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.ExactlyOnce, ""); Assert.IsTrue(result.IsSubscribed); Assert.AreEqual(result.QualityOfServiceLevel, MqttQualityOfServiceLevel.AtLeastOnce); } @@ -67,53 +78,29 @@ namespace MQTTnet.Tests.Server [TestMethod] public async Task MqttSubscriptionsManager_SubscribeSingleNoSuccess() { - var s = CreateSession(); - - var sm = new MqttClientSubscriptionsManager(s, new MqttServerOptions(), new MqttServerEventContainer(), new MqttRetainedMessagesManager()); - var sp = new MqttSubscribePacket(); sp.TopicFilters.Add(new MqttTopicFilterBuilder().WithTopic("A/B/C").Build()); - await sm.Subscribe(sp, CancellationToken.None); + await _subscriptionsManager.Subscribe(sp, CancellationToken.None); - Assert.IsFalse(sm.CheckSubscriptions("A/B/X", MqttQualityOfServiceLevel.AtMostOnce, "").IsSubscribed); + Assert.IsFalse(_subscriptionsManager.CheckSubscriptions("A/B/X", MqttQualityOfServiceLevel.AtMostOnce, "").IsSubscribed); } [TestMethod] public async Task MqttSubscriptionsManager_SubscribeAndUnsubscribeSingle() { - var s = CreateSession(); - - var sm = new MqttClientSubscriptionsManager(s, new MqttServerOptions(), new MqttServerEventContainer(), new MqttRetainedMessagesManager()); - var sp = new MqttSubscribePacket(); sp.TopicFilters.Add(new MqttTopicFilterBuilder().WithTopic("A/B/C").Build()); - await sm.Subscribe(sp, CancellationToken.None); + await _subscriptionsManager.Subscribe(sp, CancellationToken.None); - Assert.IsTrue(sm.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.AtMostOnce, "").IsSubscribed); + Assert.IsTrue(_subscriptionsManager.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.AtMostOnce, "").IsSubscribed); var up = new MqttUnsubscribePacket(); up.TopicFilters.Add("A/B/C"); - await sm.Unsubscribe(up, CancellationToken.None); + await _subscriptionsManager.Unsubscribe(up, CancellationToken.None); - Assert.IsFalse(sm.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.AtMostOnce, "").IsSubscribed); - } - - MqttClientSession CreateSession() - { - var logger = new TestLogger(); - var options = new MqttServerOptions(); - var retainedMessagesManager = new MqttRetainedMessagesManager(); - var eventContainer = new MqttServerEventContainer(); - - return new MqttClientSession( - "", - new ConcurrentDictionary(), - options, - eventContainer, - retainedMessagesManager, - new MqttClientSessionsManager(options, retainedMessagesManager, eventContainer, logger)); + Assert.IsFalse(_subscriptionsManager.CheckSubscriptions("A/B/C", MqttQualityOfServiceLevel.AtMostOnce, "").IsSubscribed); } } -} +} \ No newline at end of file diff --git a/Tests/MQTTnet.Core.Tests/Server/Security_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Security_Tests.cs index ebe3c8b9..6e4ffa9b 100644 --- a/Tests/MQTTnet.Core.Tests/Server/Security_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/Security_Tests.cs @@ -61,7 +61,7 @@ namespace MQTTnet.Tests.Server var server = await testEnvironment.StartServer(); - server.ValidatingClientConnectionAsync += e => + server.ValidatingConnectionAsync += e => { if (e.Username != "UserName1") { diff --git a/Tests/MQTTnet.Core.Tests/Server/Server_Reference_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Server_Reference_Tests.cs index 7419242e..49a82f8b 100644 --- a/Tests/MQTTnet.Core.Tests/Server/Server_Reference_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/Server_Reference_Tests.cs @@ -19,7 +19,7 @@ namespace MQTTnet.Tests.Server var server = await testEnvironment.StartServer(); - server.ValidatingClientConnectionAsync += e => + server.ValidatingConnectionAsync += e => { e.ReasonCode = MqttConnectReasonCode.ServerMoved; e.ServerReference = "new_server"; diff --git a/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs index 45d9e066..798890aa 100644 --- a/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/Session_Tests.cs @@ -20,7 +20,7 @@ namespace MQTTnet.Tests.Server { var server = await testEnvironment.StartServer(); - server.ValidatingClientConnectionAsync += e => + server.ValidatingConnectionAsync += e => { // Don't validate anything. Just set some session items. e.SessionItems["can_subscribe_x"] = true; @@ -29,7 +29,7 @@ namespace MQTTnet.Tests.Server return Task.CompletedTask; }; - server.InterceptingClientSubscriptionAsync += e => + server.InterceptingSubscriptionAsync += e => { if (e.TopicFilter.Topic == "x") { @@ -42,7 +42,7 @@ namespace MQTTnet.Tests.Server return Task.CompletedTask; }; - server.InterceptingClientPublishAsync += e => + server.InterceptingPublishAsync += e => { e.ApplicationMessage.Payload = Encoding.UTF8.GetBytes( @@ -80,7 +80,7 @@ namespace MQTTnet.Tests.Server { var server = await testEnvironment.StartServer(); - server.ValidatingClientConnectionAsync += e => + server.ValidatingConnectionAsync += e => { // Don't validate anything. Just set some session items. e.SessionItems["can_subscribe_x"] = true; @@ -91,7 +91,7 @@ namespace MQTTnet.Tests.Server await testEnvironment.ConnectClient(); - var sessionStatus = await testEnvironment.Server.GetSessionStatusAsync(); + var sessionStatus = await testEnvironment.Server.GetSessionsAsync(); var session = sessionStatus.First(); Assert.AreEqual(true, session.Items["can_subscribe_x"]); @@ -128,7 +128,7 @@ namespace MQTTnet.Tests.Server var client = await testEnvironment.ConnectClient(); // Arrange session status tracking. - var status = await server.GetSessionStatusAsync(); + var status = await server.GetSessionsAsync(); var clientStatus = status[0]; var deletedEventFired = false; @@ -158,7 +158,7 @@ namespace MQTTnet.Tests.Server var messageReceivedHandler = testEnvironment.CreateApplicationMessageHandler(client); // Arrange session status tracking. - var status = await server.GetSessionStatusAsync(); + var status = await server.GetSessionsAsync(); var clientStatus = status[0]; // Act: Inject the application message. diff --git a/Tests/MQTTnet.Core.Tests/Server/Status_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Status_Tests.cs index 49f5a9cb..05eac932 100644 --- a/Tests/MQTTnet.Core.Tests/Server/Status_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/Status_Tests.cs @@ -23,8 +23,8 @@ namespace MQTTnet.Tests.Server await Task.Delay(500); - var clientStatus = await server.GetClientStatusAsync(); - var sessionStatus = await server.GetSessionStatusAsync(); + var clientStatus = await server.GetClientsAsync(); + var sessionStatus = await server.GetSessionsAsync(); Assert.AreEqual(2, clientStatus.Count); Assert.AreEqual(2, sessionStatus.Count); @@ -37,8 +37,8 @@ namespace MQTTnet.Tests.Server await Task.Delay(500); - clientStatus = await server.GetClientStatusAsync(); - sessionStatus = await server.GetSessionStatusAsync(); + clientStatus = await server.GetClientsAsync(); + sessionStatus = await server.GetSessionsAsync(); Assert.AreEqual(0, clientStatus.Count); Assert.AreEqual(0, sessionStatus.Count); @@ -56,7 +56,7 @@ namespace MQTTnet.Tests.Server await Task.Delay(1000); - var clientStatus = await server.GetClientStatusAsync(); + var clientStatus = await server.GetClientsAsync(); Assert.AreEqual(1, clientStatus.Count); Assert.IsTrue(clientStatus.Any(s => s.ClientId == c1.Options.ClientId)); @@ -67,7 +67,7 @@ namespace MQTTnet.Tests.Server Assert.IsFalse(c1.IsConnected); - clientStatus = await server.GetClientStatusAsync(); + clientStatus = await server.GetClientsAsync(); Assert.AreEqual(0, clientStatus.Count); } @@ -87,8 +87,8 @@ namespace MQTTnet.Tests.Server await Task.Delay(500); - var clientStatus = await server.GetClientStatusAsync(); - var sessionStatus = await server.GetSessionStatusAsync(); + var clientStatus = await server.GetClientsAsync(); + var sessionStatus = await server.GetSessionsAsync(); Assert.AreEqual(1, clientStatus.Count); Assert.AreEqual(2, sessionStatus.Count); @@ -97,8 +97,8 @@ namespace MQTTnet.Tests.Server await Task.Delay(500); - clientStatus = await server.GetClientStatusAsync(); - sessionStatus = await server.GetSessionStatusAsync(); + clientStatus = await server.GetClientsAsync(); + sessionStatus = await server.GetSessionsAsync(); Assert.AreEqual(0, clientStatus.Count); Assert.AreEqual(2, sessionStatus.Count); @@ -119,7 +119,7 @@ namespace MQTTnet.Tests.Server await c1.PublishAsync("a"); await Task.Delay(50); - var clientStatus = await server.GetClientStatusAsync(); + var clientStatus = await server.GetClientsAsync(); Assert.AreEqual(i, clientStatus.First().SentApplicationMessagesCount); Assert.AreEqual(0, clientStatus.First().ReceivedApplicationMessagesCount); } @@ -143,7 +143,7 @@ namespace MQTTnet.Tests.Server await Task.Delay(500); - var clientStatus = await server.GetClientStatusAsync(); + var clientStatus = await server.GetClientsAsync(); Assert.AreEqual(i, clientStatus.First().SentApplicationMessagesCount, "SAMC invalid!"); diff --git a/Tests/MQTTnet.Core.Tests/Server/Subscribe_Tests.cs b/Tests/MQTTnet.Core.Tests/Server/Subscribe_Tests.cs index 806b52c8..ba897225 100644 --- a/Tests/MQTTnet.Core.Tests/Server/Subscribe_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/Server/Subscribe_Tests.cs @@ -20,7 +20,7 @@ namespace MQTTnet.Tests.Server { var server = await testEnvironment.StartServer(); - server.InterceptingClientSubscriptionAsync += e => + server.InterceptingSubscriptionAsync += e => { // Set the topic to "a" regards what the client wants to subscribe. e.TopicFilter.Topic = "a"; @@ -290,7 +290,7 @@ namespace MQTTnet.Tests.Server { var server = await testEnvironment.StartServer(); - server.InterceptingClientSubscriptionAsync += e => + server.InterceptingSubscriptionAsync += e => { if (e.TopicFilter.Topic == "not_allowed_topic") { diff --git a/Tests/MQTTnet.TestApp.NetCore/Program.cs b/Tests/MQTTnet.TestApp.NetCore/Program.cs index 1a51677c..0c199be3 100644 --- a/Tests/MQTTnet.TestApp.NetCore/Program.cs +++ b/Tests/MQTTnet.TestApp.NetCore/Program.cs @@ -134,39 +134,6 @@ namespace MQTTnet.TestApp.NetCore } } - public class RetainedMessageHandler : IMqttServerStorage - { - const string Filename = "C:\\MQTT\\RetainedMessages.json"; - - public Task SaveRetainedMessagesAsync(IList messages) - { - var directory = Path.GetDirectoryName(Filename); - if (!Directory.Exists(directory)) - { - Directory.CreateDirectory(directory); - } - - File.WriteAllText(Filename, JsonConvert.SerializeObject(messages)); - return Task.FromResult(0); - } - - public Task> LoadRetainedMessagesAsync() - { - IList retainedMessages; - if (File.Exists(Filename)) - { - var json = File.ReadAllText(Filename); - retainedMessages = JsonConvert.DeserializeObject>(json); - } - else - { - retainedMessages = new List(); - } - - return Task.FromResult(retainedMessages); - } - } - public class WikiCode { public void Code() diff --git a/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs index a7e6c97c..70620ca5 100644 --- a/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs +++ b/Tests/MQTTnet.TestApp.NetCore/ServerTest.cs @@ -1,10 +1,14 @@ using System; +using System.Collections.Generic; +using System.IO; using System.Text; using System.Threading.Tasks; using MQTTnet.Diagnostics; +using MQTTnet.Extensions.ManagedClient; using MQTTnet.Implementations; using MQTTnet.Protocol; using MQTTnet.Server; +using Newtonsoft.Json; namespace MQTTnet.TestApp.NetCore { @@ -36,10 +40,7 @@ namespace MQTTnet.TestApp.NetCore { try { - var options = new MqttServerOptions - { - Storage = new RetainedMessageHandler() - }; + var options = new MqttServerOptions(); // Extend the timestamp for all messages from clients. // Protect several topics from being subscribed from every client. @@ -52,7 +53,45 @@ namespace MQTTnet.TestApp.NetCore var mqttServer = new MqttFactory().CreateMqttServer(options); - mqttServer.InterceptingClientPublishAsync += e => + const string Filename = "C:\\MQTT\\RetainedMessages.json"; + + mqttServer.RetainedMessageChangedAsync += e => + { + var directory = Path.GetDirectoryName(Filename); + if (!Directory.Exists(directory)) + { + Directory.CreateDirectory(directory); + } + + File.WriteAllText(Filename, JsonConvert.SerializeObject(e.StoredRetainedMessages)); + return Task.FromResult(0); + }; + + mqttServer.RetainedMessagesClearedAsync += e => + { + File.Delete(Filename); + return Task.FromResult(0); + }; + + mqttServer.LoadingRetainedMessageAsync += e => + { + List retainedMessages; + if (File.Exists(Filename)) + { + var json = File.ReadAllText(Filename); + retainedMessages = JsonConvert.DeserializeObject>(json); + } + else + { + retainedMessages = new List(); + } + + e.StoredRetainedMessages = retainedMessages; + + return Task.FromResult(0); + }; + + mqttServer.InterceptingPublishAsync += e => { if (MqttTopicFilterComparer.Compare(e.ApplicationMessage.Topic, "/myTopic/WithTimestamp/#") == MqttTopicFilterCompareResult.IsMatch) { @@ -70,7 +109,7 @@ namespace MQTTnet.TestApp.NetCore return PlatformAbstractionLayer.CompletedTask; }; - mqttServer.ValidatingClientConnectionAsync += e => + mqttServer.ValidatingConnectionAsync += e => { if (e.ClientId == "SpecialClient") { @@ -83,7 +122,7 @@ namespace MQTTnet.TestApp.NetCore return PlatformAbstractionLayer.CompletedTask; }; - mqttServer.InterceptingClientSubscriptionAsync += e => + mqttServer.InterceptingSubscriptionAsync += e => { if (e.TopicFilter.Topic.StartsWith("admin/foo/bar") && e.ClientId != "theAdmin") { @@ -99,7 +138,7 @@ namespace MQTTnet.TestApp.NetCore return PlatformAbstractionLayer.CompletedTask; }; - mqttServer.InterceptingClientPublishAsync += e => + mqttServer.InterceptingPublishAsync += e => { MqttNetConsoleLogger.PrintToConsole( $"'{e.ClientId}' reported '{e.ApplicationMessage.Topic}' > '{Encoding.UTF8.GetString(e.ApplicationMessage.Payload ?? new byte[0])}'",