From 09385b1b3600b918c134871baf51339bdc961e93 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sat, 23 Jul 2022 18:07:49 +0200 Subject: [PATCH 1/4] Pass session item to several more event handers. --- .../ApplicationMessageNotConsumedEventArgs.cs | 10 ++++- .../Server/Events/ClientConnectedEventArgs.cs | 39 +++++++++++----- .../Events/ClientDisconnectedEventArgs.cs | 26 ++++++++--- .../Events/ClientSubscribedTopicEventArgs.cs | 31 +++++++++---- .../ClientUnsubscribedTopicEventArgs.cs | 31 +++++++++---- .../Events/InterceptingPacketEventArgs.cs | 39 +++++++++++----- .../Events/InterceptingPublishEventArgs.cs | 14 ++++-- .../InterceptingSubscriptionEventArgs.cs | 20 +++++++-- .../InterceptingUnsubscriptionEventArgs.cs | 16 +++++-- .../Events/RetainedMessageChangedEventArgs.cs | 17 ++++--- .../Server/Events/SessionDeletedEventArgs.cs | 16 ++++++- Source/MQTTnet/Server/Internal/MqttClient.cs | 31 ++----------- .../Internal/MqttClientSessionsManager.cs | 44 +++++-------------- .../MqttClientSubscriptionsManager.cs | 31 ++----------- .../Internal/MqttRetainedMessagesManager.cs | 8 +--- 15 files changed, 212 insertions(+), 161 deletions(-) diff --git a/Source/MQTTnet/Server/Events/ApplicationMessageNotConsumedEventArgs.cs b/Source/MQTTnet/Server/Events/ApplicationMessageNotConsumedEventArgs.cs index 28b0e72d..d81e6bb7 100644 --- a/Source/MQTTnet/Server/Events/ApplicationMessageNotConsumedEventArgs.cs +++ b/Source/MQTTnet/Server/Events/ApplicationMessageNotConsumedEventArgs.cs @@ -8,14 +8,20 @@ namespace MQTTnet.Server { public sealed class ApplicationMessageNotConsumedEventArgs : EventArgs { + public ApplicationMessageNotConsumedEventArgs(MqttApplicationMessage applicationMessage, string senderId) + { + ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage)); + SenderId = senderId; + } + /// /// Gets the application message which was not consumed by any client. /// - public MqttApplicationMessage ApplicationMessage { get; internal set; } + public MqttApplicationMessage ApplicationMessage { get; } /// /// Gets the ID of the client which has sent the affected application message. /// - public string SenderId { get; internal set; } + public string SenderId { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Events/ClientConnectedEventArgs.cs b/Source/MQTTnet/Server/Events/ClientConnectedEventArgs.cs index 2a8845ee..39042738 100644 --- a/Source/MQTTnet/Server/Events/ClientConnectedEventArgs.cs +++ b/Source/MQTTnet/Server/Events/ClientConnectedEventArgs.cs @@ -3,31 +3,46 @@ // See the LICENSE file in the project root for more information. using System; +using System.Collections; using MQTTnet.Formatter; namespace MQTTnet.Server { public sealed class ClientConnectedEventArgs : EventArgs { - /// - /// Gets the client identifier of the connected client. - /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. - /// - public string ClientId { get; internal set; } + public ClientConnectedEventArgs(string clientId, string userName, MqttProtocolVersion protocolVersion, string endpoint, IDictionary sessionItems) + { + ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); + UserName = userName; + ProtocolVersion = protocolVersion; + Endpoint = endpoint; + SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems)); + } /// - /// Gets the user name of the connected client. + /// Gets the client identifier of the connected client. + /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. /// - public string UserName { get; internal set; } + public string ClientId { get; } /// - /// Gets the protocol version which is used by the connected client. + /// Gets the endpoint of the connected client. /// - public MqttProtocolVersion ProtocolVersion { get; internal set; } + public string Endpoint { get; } /// - /// Gets the endpoint of the connected client. + /// Gets the protocol version which is used by the connected client. /// - public string Endpoint { get; internal set; } + public MqttProtocolVersion ProtocolVersion { get; } + + /// + /// Gets or sets a key/value collection that can be used to share data within the scope of this session. + /// + public IDictionary SessionItems { get; } + + /// + /// Gets the user name of the connected client. + /// + public string UserName { get; } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Events/ClientDisconnectedEventArgs.cs b/Source/MQTTnet/Server/Events/ClientDisconnectedEventArgs.cs index aeac4138..20e2b2dc 100644 --- a/Source/MQTTnet/Server/Events/ClientDisconnectedEventArgs.cs +++ b/Source/MQTTnet/Server/Events/ClientDisconnectedEventArgs.cs @@ -3,19 +3,33 @@ // See the LICENSE file in the project root for more information. using System; +using System.Collections; namespace MQTTnet.Server { public sealed class ClientDisconnectedEventArgs : EventArgs { + public ClientDisconnectedEventArgs(string clientId, MqttClientDisconnectType disconnectType, string endpoint, IDictionary sessionItems) + { + ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); + DisconnectType = disconnectType; + Endpoint = endpoint; + SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems)); + } + /// - /// Gets the client identifier. - /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. + /// Gets the client identifier. + /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. /// - public string ClientId { get; internal set; } + public string ClientId { get; } - public MqttClientDisconnectType DisconnectType { get; internal set; } + public MqttClientDisconnectType DisconnectType { get; } - public string Endpoint { get; internal set; } + public string Endpoint { get; } + + /// + /// Gets or sets a key/value collection that can be used to share data within the scope of this session. + /// + public IDictionary SessionItems { get; } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Events/ClientSubscribedTopicEventArgs.cs b/Source/MQTTnet/Server/Events/ClientSubscribedTopicEventArgs.cs index 34105673..639ad98e 100644 --- a/Source/MQTTnet/Server/Events/ClientSubscribedTopicEventArgs.cs +++ b/Source/MQTTnet/Server/Events/ClientSubscribedTopicEventArgs.cs @@ -3,22 +3,35 @@ // See the LICENSE file in the project root for more information. using System; +using System.Collections; using MQTTnet.Packets; namespace MQTTnet.Server { public sealed class ClientSubscribedTopicEventArgs : EventArgs { - /// - /// Gets the client identifier. - /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. - /// - public string ClientId { get; internal set; } + public ClientSubscribedTopicEventArgs(string clientId, MqttTopicFilter topicFilter, IDictionary sessionItems) + { + ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); + TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter)); + SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems)); + } /// - /// Gets the topic filter. - /// The topic filter can contain topics and wildcards. + /// Gets the client identifier. + /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. /// - public MqttTopicFilter TopicFilter { get; internal set; } + public string ClientId { get; } + + /// + /// Gets or sets a key/value collection that can be used to share data within the scope of this session. + /// + public IDictionary SessionItems { get; } + + /// + /// Gets the topic filter. + /// The topic filter can contain topics and wildcards. + /// + public MqttTopicFilter TopicFilter { get; } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Events/ClientUnsubscribedTopicEventArgs.cs b/Source/MQTTnet/Server/Events/ClientUnsubscribedTopicEventArgs.cs index 9a9f73bc..d521b857 100644 --- a/Source/MQTTnet/Server/Events/ClientUnsubscribedTopicEventArgs.cs +++ b/Source/MQTTnet/Server/Events/ClientUnsubscribedTopicEventArgs.cs @@ -3,21 +3,34 @@ // See the LICENSE file in the project root for more information. using System; +using System.Collections; namespace MQTTnet.Server { public sealed class ClientUnsubscribedTopicEventArgs : EventArgs { - /// - /// Gets the client identifier. - /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. - /// - public string ClientId { get; internal set; } + public ClientUnsubscribedTopicEventArgs(string clientId, string topicFilter, IDictionary sessionItems) + { + ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); + TopicFilter = topicFilter ?? throw new ArgumentNullException(nameof(topicFilter)); + SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems)); + } /// - /// Gets or sets the topic filter. - /// The topic filter can contain topics and wildcards. + /// Gets the client identifier. + /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. /// - public string TopicFilter { get; internal set; } + public string ClientId { get; } + + /// + /// Gets or sets a key/value collection that can be used to share data within the scope of this session. + /// + public IDictionary SessionItems { get; } + + /// + /// Gets or sets the topic filter. + /// The topic filter can contain topics and wildcards. + /// + public string TopicFilter { get; } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Events/InterceptingPacketEventArgs.cs b/Source/MQTTnet/Server/Events/InterceptingPacketEventArgs.cs index 3f2661cb..df62b54a 100644 --- a/Source/MQTTnet/Server/Events/InterceptingPacketEventArgs.cs +++ b/Source/MQTTnet/Server/Events/InterceptingPacketEventArgs.cs @@ -3,6 +3,7 @@ // See the LICENSE file in the project root for more information. using System; +using System.Collections; using System.Threading; using MQTTnet.Packets; @@ -10,30 +11,44 @@ namespace MQTTnet.Server { public sealed class InterceptingPacketEventArgs : EventArgs { + public InterceptingPacketEventArgs(CancellationToken cancellationToken, string clientId, string endpoint, MqttPacket packet, IDictionary sessionItems) + { + CancellationToken = cancellationToken; + ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); + Endpoint = endpoint; + Packet = packet ?? throw new ArgumentNullException(nameof(packet)); + SessionItems = sessionItems; + } + /// - /// Gets the client ID which has sent the packet or will receive the packet. + /// Gets the cancellation token from the connection managing thread. + /// Use this in further event processing. /// - public string ClientId { get; internal set; } - + public CancellationToken CancellationToken { get; } + /// - /// Gets the endpoint of the sending or receiving client. + /// Gets the client ID which has sent the packet or will receive the packet. /// - public string Endpoint { get; internal set; } - + public string ClientId { get; } + /// - /// Gets or sets the MQTT packet which was received or will be sent. + /// Gets the endpoint of the sending or receiving client. + /// + public string Endpoint { get; } + + /// + /// Gets or sets the MQTT packet which was received or will be sent. /// public MqttPacket Packet { get; set; } /// - /// Gets or sets whether the packet should be processed or not. + /// Gets or sets whether the packet should be processed or not. /// public bool ProcessPacket { get; set; } = true; - + /// - /// Gets the cancellation token from the connection managing thread. - /// Use this in further event processing. + /// Gets or sets a key/value collection that can be used to share data within the scope of this session. /// - public CancellationToken CancellationToken { get; internal set; } + public IDictionary SessionItems { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Events/InterceptingPublishEventArgs.cs b/Source/MQTTnet/Server/Events/InterceptingPublishEventArgs.cs index 09a4d76e..34b12b56 100644 --- a/Source/MQTTnet/Server/Events/InterceptingPublishEventArgs.cs +++ b/Source/MQTTnet/Server/Events/InterceptingPublishEventArgs.cs @@ -10,18 +10,26 @@ namespace MQTTnet.Server { public sealed class InterceptingPublishEventArgs : EventArgs { + public InterceptingPublishEventArgs(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken, string clientId, IDictionary sessionItems) + { + ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage)); + CancellationToken = cancellationToken; + ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); + SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems)); + } + public MqttApplicationMessage ApplicationMessage { get; set; } /// /// Gets the cancellation token which can indicate that the client connection gets down. /// - public CancellationToken CancellationToken { get; internal set; } + public CancellationToken CancellationToken { get; } /// /// Gets the client identifier. /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. /// - public string ClientId { get; internal set; } + public string ClientId { get; } public bool CloseConnection { get; set; } @@ -38,6 +46,6 @@ namespace MQTTnet.Server /// /// Gets or sets a key/value collection that can be used to share data within the scope of this session. /// - public IDictionary SessionItems { get; internal set; } + public IDictionary SessionItems { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs b/Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs index 8b173f1d..d7f787e1 100644 --- a/Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs +++ b/Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs @@ -12,16 +12,28 @@ namespace MQTTnet.Server { public sealed class InterceptingSubscriptionEventArgs : EventArgs { + public InterceptingSubscriptionEventArgs( + CancellationToken cancellationToken, + string clientId, + MqttSessionStatus session, + MqttTopicFilter topicFilter) + { + CancellationToken = cancellationToken; + ClientId = clientId; + Session = session; + TopicFilter = topicFilter; + } + /// /// Gets the cancellation token which can indicate that the client connection gets down. /// - public CancellationToken CancellationToken { get; internal set; } + public CancellationToken CancellationToken { get; } /// /// Gets the client identifier. /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. /// - public string ClientId { get; internal set; } + public string ClientId { get; } /// /// Gets or sets whether the broker should close the client connection. @@ -48,12 +60,12 @@ namespace MQTTnet.Server /// /// Gets the current client session. /// - public MqttSessionStatus Session { get; internal set; } + public MqttSessionStatus Session { get; } /// /// Gets or sets a key/value collection that can be used to share data within the scope of this session. /// - public IDictionary SessionItems { get; internal set; } + public IDictionary SessionItems => Session.Items; /// /// Gets or sets the topic filter. diff --git a/Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs b/Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs index 648ccd6f..63e8cad5 100644 --- a/Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs +++ b/Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs @@ -10,16 +10,24 @@ namespace MQTTnet.Server { public sealed class InterceptingUnsubscriptionEventArgs : EventArgs { + public InterceptingUnsubscriptionEventArgs(CancellationToken cancellationToken, string clientId, IDictionary sessionItems, string topic) + { + CancellationToken = cancellationToken; + ClientId = clientId; + SessionItems = sessionItems; + Topic = topic; + } + /// /// Gets the cancellation token which can indicate that the client connection gets down. /// - public CancellationToken CancellationToken { get; internal set; } + public CancellationToken CancellationToken { get; } /// /// Gets the client identifier. /// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues. /// - public string ClientId { get; internal set; } + public string ClientId { get; } /// /// Gets or sets whether the broker should close the client connection. @@ -41,7 +49,7 @@ namespace MQTTnet.Server /// /// Gets or sets a key/value collection that can be used to share data within the scope of this session. /// - public IDictionary SessionItems { get; internal set; } + public IDictionary SessionItems { get; } /// /// Gets or sets the MQTT topic. @@ -50,6 +58,6 @@ namespace MQTTnet.Server /// The topic consists of one or more topic levels. Each topic level is separated by a forward slash (topic level /// separator). /// - public string Topic { get; internal set; } + public string Topic { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Events/RetainedMessageChangedEventArgs.cs b/Source/MQTTnet/Server/Events/RetainedMessageChangedEventArgs.cs index 1d436fc3..c7beb807 100644 --- a/Source/MQTTnet/Server/Events/RetainedMessageChangedEventArgs.cs +++ b/Source/MQTTnet/Server/Events/RetainedMessageChangedEventArgs.cs @@ -9,10 +9,17 @@ 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; } + public RetainedMessageChangedEventArgs(string clientId, MqttApplicationMessage changedRetainedMessage, List storedRetainedMessages) + { + ClientId = clientId ?? throw new ArgumentNullException(nameof(clientId)); + ChangedRetainedMessage = changedRetainedMessage ?? throw new ArgumentNullException(nameof(changedRetainedMessage)); + StoredRetainedMessages = storedRetainedMessages ?? throw new ArgumentNullException(nameof(storedRetainedMessages)); + } + + public MqttApplicationMessage ChangedRetainedMessage { get; } + + public string ClientId { get; } + + public List StoredRetainedMessages { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Events/SessionDeletedEventArgs.cs b/Source/MQTTnet/Server/Events/SessionDeletedEventArgs.cs index 844f7c15..3d27036c 100644 --- a/Source/MQTTnet/Server/Events/SessionDeletedEventArgs.cs +++ b/Source/MQTTnet/Server/Events/SessionDeletedEventArgs.cs @@ -3,14 +3,26 @@ // See the LICENSE file in the project root for more information. using System; +using System.Collections; namespace MQTTnet.Server { public sealed class SessionDeletedEventArgs : EventArgs { + public SessionDeletedEventArgs(string id, IDictionary sessionItems) + { + Id = id ?? throw new ArgumentNullException(nameof(id)); + SessionItems = sessionItems ?? throw new ArgumentNullException(nameof(sessionItems)); + } + /// - /// Gets the ID of the session. + /// Gets the ID of the session. /// - public string Id { get; internal set; } + public string Id { get; } + + /// + /// Gets or sets a key/value collection that can be used to share data within the scope of this session. + /// + public IDictionary SessionItems { get; } } } \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttClient.cs b/Source/MQTTnet/Server/Internal/MqttClient.cs index 6d2f59e1..9e97b671 100644 --- a/Source/MQTTnet/Server/Internal/MqttClient.cs +++ b/Source/MQTTnet/Server/Internal/MqttClient.cs @@ -160,15 +160,8 @@ namespace MQTTnet.Server { return packet; } - - var interceptingPacketEventArgs = new InterceptingPacketEventArgs - { - ClientId = Id, - Endpoint = Endpoint, - Packet = packet, - CancellationToken = cancellationToken - }; - + + var interceptingPacketEventArgs = new InterceptingPacketEventArgs(cancellationToken, Id, Endpoint, packet, Session.Items); await _eventContainer.InterceptingOutboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false); if (!interceptingPacketEventArgs.ProcessPacket || packet == null) @@ -200,14 +193,7 @@ namespace MQTTnet.Server if (_eventContainer.InterceptingInboundPacketEvent.HasHandlers) { - var interceptingPacketEventArgs = new InterceptingPacketEventArgs - { - ClientId = Id, - Endpoint = Endpoint, - Packet = packet, - CancellationToken = cancellationToken - }; - + var interceptingPacketEventArgs = new InterceptingPacketEventArgs(cancellationToken, Id, Endpoint, packet, Session.Items); await _eventContainer.InterceptingInboundPacketEvent.InvokeAsync(interceptingPacketEventArgs).ConfigureAwait(false); packet = interceptingPacketEventArgs.Packet; processPacket = interceptingPacketEventArgs.ProcessPacket; @@ -428,16 +414,7 @@ namespace MQTTnet.Server if (_eventContainer.InterceptingPublishEvent.HasHandlers) { - interceptingPublishEventArgs = new InterceptingPublishEventArgs - { - ClientId = Id, - ApplicationMessage = applicationMessage, - SessionItems = Session.Items, - ProcessPublish = true, - CloseConnection = false, - CancellationToken = cancellationToken - }; - + interceptingPublishEventArgs = new InterceptingPublishEventArgs(applicationMessage, cancellationToken, Id, Session.Items); if (string.IsNullOrEmpty(interceptingPublishEventArgs.ApplicationMessage.Topic)) { // This can happen if a topic alias us used but the topic is diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 7d2e47ec..4787e28c 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -109,13 +109,9 @@ namespace MQTTnet.Server try { - if (_eventContainer.SessionDeletedEvent.HasHandlers) + if (_eventContainer.SessionDeletedEvent.HasHandlers && session != null) { - var eventArgs = new SessionDeletedEventArgs - { - Id = session?.Id - }; - + var eventArgs = new SessionDeletedEventArgs(clientId, session.Items); await _eventContainer.SessionDeletedEvent.TryInvokeAsync(eventArgs, _logger).ConfigureAwait(false); } } @@ -295,13 +291,12 @@ namespace MQTTnet.Server if (_eventContainer.ClientConnectedEvent.HasHandlers) { - var eventArgs = new ClientConnectedEventArgs - { - ClientId = connectPacket.ClientId, - UserName = connectPacket.Username, - ProtocolVersion = channelAdapter.PacketFormatterAdapter.ProtocolVersion, - Endpoint = channelAdapter.Endpoint - }; + var eventArgs = new ClientConnectedEventArgs( + connectPacket.ClientId, + connectPacket.Username, + channelAdapter.PacketFormatterAdapter.ProtocolVersion, + channelAdapter.Endpoint, + client.Session.Items); await _eventContainer.ClientConnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); } @@ -340,12 +335,8 @@ namespace MQTTnet.Server if (client.Id != null && !client.IsTakenOver && _eventContainer.ClientDisconnectedEvent.HasHandlers) { - var eventArgs = new ClientDisconnectedEventArgs - { - ClientId = client.Id, - DisconnectType = client.IsCleanDisconnect ? MqttClientDisconnectType.Clean : MqttClientDisconnectType.NotClean, - Endpoint = endpoint - }; + var disconnectType = client.IsCleanDisconnect ? MqttClientDisconnectType.Clean : MqttClientDisconnectType.NotClean; + var eventArgs = new ClientDisconnectedEventArgs(client.Id, disconnectType, endpoint, client.Session.Items); await _eventContainer.ClientDisconnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); } @@ -533,13 +524,7 @@ namespace MQTTnet.Server if (_eventContainer.ClientConnectedEvent.HasHandlers) { - var eventArgs = new ClientDisconnectedEventArgs - { - ClientId = existing.Id, - DisconnectType = MqttClientDisconnectType.Takeover, - Endpoint = existing.Endpoint - }; - + var eventArgs = new ClientDisconnectedEventArgs(existing.Id, MqttClientDisconnectType.Takeover, existing.Endpoint, existing.Session.Items); await _eventContainer.ClientDisconnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); } } @@ -572,12 +557,7 @@ namespace MQTTnet.Server return; } - var eventArgs = new ApplicationMessageNotConsumedEventArgs - { - ApplicationMessage = applicationMessage, - SenderId = senderId - }; - + var eventArgs = new ApplicationMessageNotConsumedEventArgs(applicationMessage, senderId); await _eventContainer.ApplicationMessageNotConsumedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); } diff --git a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs index 9f2f4c71..3a505a40 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs @@ -217,12 +217,7 @@ namespace MQTTnet.Server { foreach (var finalTopicFilter in finalTopicFilters) { - var eventArgs = new ClientSubscribedTopicEventArgs - { - ClientId = _session.Id, - TopicFilter = finalTopicFilter - }; - + var eventArgs = new ClientSubscribedTopicEventArgs(_session.Id, finalTopicFilter, _session.Items); await _eventContainer.ClientSubscribedTopicEvent.InvokeAsync(eventArgs).ConfigureAwait(false); } } @@ -312,12 +307,7 @@ namespace MQTTnet.Server { foreach (var topicFilter in unsubscribePacket.TopicFilters) { - var eventArgs = new ClientUnsubscribedTopicEventArgs - { - ClientId = _session.Id, - TopicFilter = topicFilter - }; - + var eventArgs = new ClientUnsubscribedTopicEventArgs(_session.Id, topicFilter, _session.Items); await _eventContainer.ClientUnsubscribedTopicEvent.InvokeAsync(eventArgs).ConfigureAwait(false); } } @@ -472,14 +462,7 @@ namespace MQTTnet.Server async Task InterceptSubscribe(MqttTopicFilter topicFilter, CancellationToken cancellationToken) { - var eventArgs = new InterceptingSubscriptionEventArgs - { - ClientId = _session.Id, - TopicFilter = topicFilter, - SessionItems = _session.Items, - Session = new MqttSessionStatus(_session), - CancellationToken = cancellationToken - }; + var eventArgs = new InterceptingSubscriptionEventArgs(cancellationToken, _session.Id, new MqttSessionStatus(_session), topicFilter); if (topicFilter.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce) { @@ -508,13 +491,7 @@ namespace MQTTnet.Server async Task InterceptUnsubscribe(string topicFilter, MqttSubscription mqttSubscription, CancellationToken cancellationToken) { - var clientUnsubscribingTopicEventArgs = new InterceptingUnsubscriptionEventArgs - { - ClientId = _session.Id, - Topic = topicFilter, - SessionItems = _session.Items, - CancellationToken = cancellationToken - }; + var clientUnsubscribingTopicEventArgs = new InterceptingUnsubscriptionEventArgs(cancellationToken, topicFilter, _session.Items, topicFilter); if (mqttSubscription == null) { diff --git a/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs b/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs index 36737349..acc4672d 100644 --- a/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs @@ -105,13 +105,7 @@ namespace MQTTnet.Server { using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false)) { - var eventArgs = new RetainedMessageChangedEventArgs - { - ClientId = clientId, - ChangedRetainedMessage = applicationMessage, - StoredRetainedMessages = messagesForSave - }; - + var eventArgs = new RetainedMessageChangedEventArgs(clientId, applicationMessage, messagesForSave); await _eventContainer.RetainedMessageChangedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); } } From 02ede5a2d8d790b9185f73d01b3beb5fd77e3ff5 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sun, 24 Jul 2022 10:44:27 +0200 Subject: [PATCH 2/4] Update MqttClient.cs --- Source/MQTTnet/Server/Internal/MqttClient.cs | 163 +------------------ 1 file changed, 7 insertions(+), 156 deletions(-) diff --git a/Source/MQTTnet/Server/Internal/MqttClient.cs b/Source/MQTTnet/Server/Internal/MqttClient.cs index 3d4c5d82..087473c5 100644 --- a/Source/MQTTnet/Server/Internal/MqttClient.cs +++ b/Source/MQTTnet/Server/Internal/MqttClient.cs @@ -170,7 +170,7 @@ namespace MQTTnet.Server ClientId = Id, SessionItems = Session.Items }; - + return _eventContainer.ClientAcknowledgedPublishPacketEvent.TryInvokeAsync(eventArgs, _logger); } @@ -180,7 +180,7 @@ namespace MQTTnet.Server Task HandleIncomingPubAckPacket(MqttPubAckPacket pubAckPacket) { var acknowledgedPublishPacket = Session.AcknowledgePublishPacket(pubAckPacket.PacketIdentifier); - + if (acknowledgedPublishPacket != null) { return ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubAckPacket); @@ -192,7 +192,7 @@ namespace MQTTnet.Server Task HandleIncomingPubCompPacket(MqttPubCompPacket pubCompPacket) { var acknowledgedPublishPacket = Session.AcknowledgePublishPacket(pubCompPacket.PacketIdentifier); - + if (acknowledgedPublishPacket != null) { return ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubCompPacket); @@ -212,16 +212,7 @@ namespace MQTTnet.Server if (_eventContainer.InterceptingPublishEvent.HasHandlers) { - interceptingPublishEventArgs = new InterceptingPublishEventArgs - { - ClientId = Id, - ApplicationMessage = applicationMessage, - SessionItems = Session.Items, - ProcessPublish = true, - CloseConnection = false, - CancellationToken = cancellationToken - }; - + interceptingPublishEventArgs = new InterceptingPublishEventArgs(applicationMessage, cancellationToken, Id, Session.Items); if (string.IsNullOrEmpty(interceptingPublishEventArgs.ApplicationMessage.Topic)) { // This can happen if a topic alias us used but the topic is @@ -277,7 +268,7 @@ namespace MQTTnet.Server async Task HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket) { var acknowledgedPublishPacket = Session.PeekAcknowledgePublishPacket(pubRecPacket.PacketIdentifier); - + if (acknowledgedPublishPacket != null) { await ClientAcknowledgedPublishPacket(acknowledgedPublishPacket, pubRecPacket).ConfigureAwait(false); @@ -317,6 +308,7 @@ namespace MQTTnet.Server } } + async Task HandleIncomingUnsubscribePacket(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken) { var unsubscribeResult = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket, cancellationToken).ConfigureAwait(false); @@ -386,7 +378,7 @@ namespace MQTTnet.Server while (!cancellationToken.IsCancellationRequested) { await Task.Yield(); - + var packet = await ChannelAdapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false); if (packet == null) { @@ -557,147 +549,6 @@ namespace MQTTnet.Server } } - void HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket) - { - var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); - Session.EnqueueControlPacket(new MqttPacketBusItem(pubRelPacket)); - } - - void HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket) - { - var pubCompPacket = _packetFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success); - Session.EnqueueControlPacket(new MqttPacketBusItem(pubCompPacket)); - } - - async Task HandleIncomingSubscribePacket(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) - { - var subscribeResult = await Session.SubscriptionsManager.Subscribe(subscribePacket, cancellationToken).ConfigureAwait(false); - - var subAckPacket = _packetFactories.SubAck.Create(subscribePacket, subscribeResult); - - Session.EnqueueControlPacket(new MqttPacketBusItem(subAckPacket)); - - if (subscribeResult.CloseConnection) - { - StopInternal(); - return; - } - - if (subscribeResult.RetainedMessages != null) - { - foreach (var retainedApplicationMessage in subscribeResult.RetainedMessages) - { - var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage); - Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket)); - } - } - } - - async Task HandleIncomingUnsubscribePacket(MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken) - { - var unsubscribeResult = await Session.SubscriptionsManager.Unsubscribe(unsubscribePacket, cancellationToken).ConfigureAwait(false); - - var unsubAckPacket = _packetFactories.UnsubAck.Create(unsubscribePacket, unsubscribeResult); - - Session.EnqueueControlPacket(new MqttPacketBusItem(unsubAckPacket)); - - if (unsubscribeResult.CloseConnection) - { - StopInternal(); - } - } - - async Task HandleIncomingPublishPacket(MqttPublishPacket publishPacket, CancellationToken cancellationToken) - { - HandleTopicAlias(publishPacket); - - InterceptingPublishEventArgs interceptingPublishEventArgs = null; - var applicationMessage = _applicationMessageFactory.Create(publishPacket); - var closeConnection = false; - var processPublish = true; - - if (_eventContainer.InterceptingPublishEvent.HasHandlers) - { - interceptingPublishEventArgs = new InterceptingPublishEventArgs(applicationMessage, cancellationToken, Id, Session.Items); - if (string.IsNullOrEmpty(interceptingPublishEventArgs.ApplicationMessage.Topic)) - { - // This can happen if a topic alias us used but the topic is - // unknown to the server. - interceptingPublishEventArgs.Response.ReasonCode = MqttPubAckReasonCode.TopicNameInvalid; - interceptingPublishEventArgs.ProcessPublish = false; - } - - await _eventContainer.InterceptingPublishEvent.InvokeAsync(interceptingPublishEventArgs).ConfigureAwait(false); - - applicationMessage = interceptingPublishEventArgs.ApplicationMessage; - closeConnection = interceptingPublishEventArgs.CloseConnection; - processPublish = interceptingPublishEventArgs.ProcessPublish; - } - - if (closeConnection) - { - await StopAsync(MqttDisconnectReasonCode.UnspecifiedError); - return; - } - - if (processPublish && applicationMessage != null) - { - await _sessionsManager.DispatchApplicationMessage(Id, applicationMessage).ConfigureAwait(false); - } - - switch (publishPacket.QualityOfServiceLevel) - { - case MqttQualityOfServiceLevel.AtMostOnce: - { - // Do nothing since QoS 0 has no ack at all! - break; - } - case MqttQualityOfServiceLevel.AtLeastOnce: - { - var pubAckPacket = _packetFactories.PubAck.Create(publishPacket, interceptingPublishEventArgs); - Session.EnqueueControlPacket(new MqttPacketBusItem(pubAckPacket)); - break; - } - case MqttQualityOfServiceLevel.ExactlyOnce: - { - var pubRecPacket = _packetFactories.PubRec.Create(publishPacket, interceptingPublishEventArgs); - Session.EnqueueControlPacket(new MqttPacketBusItem(pubRecPacket)); - break; - } - default: - { - throw new MqttCommunicationException("Received a not supported QoS level."); - } - } - } - - void HandleTopicAlias(MqttPublishPacket publishPacket) - { - if (publishPacket.TopicAlias == 0) - { - return; - } - - lock (_topicAlias) - { - if (!string.IsNullOrEmpty(publishPacket.Topic)) - { - _topicAlias[publishPacket.TopicAlias] = publishPacket.Topic; - } - else - { - if (_topicAlias.TryGetValue(publishPacket.TopicAlias, out var topic)) - { - publishPacket.Topic = topic; - } - else - { - _logger.Warning("Client '{0}': Received invalid topic alias ({1}).", Id, publishPacket.TopicAlias); - } - } - } - } - async Task TrySendDisconnectPacket(MqttDisconnectReasonCode reasonCode) { try From 5a21939e29de0e17adbea04b07f98b5e0fa10997 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sun, 31 Jul 2022 11:09:39 +0200 Subject: [PATCH 3/4] Squashed commit of the following: commit 469caf656e36b50591c3ad01a1d1eae087d8ef9c Author: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sun Jul 31 10:55:40 2022 +0200 Fix build errors. commit 42402fba04faa6e2b60f4ae61d68de8804211a3c Author: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sun Jul 31 10:22:23 2022 +0200 Allow interception of injected application messages. commit a0dc8ab0d721b705aae9fda583c7152fae68759b Author: William Lee <31092948+william19821101@users.noreply.github.com> Date: Sun Jul 31 15:50:14 2022 +0800 Update Server_Simple_Samples.cs (#1474) Change e.UserName to e.Username --- .github/workflows/ReleaseNotes.md | 3 +- Samples/Server/Server_Simple_Samples.cs | 4 +- .../Internal/MqttServerEventContainer.cs | 1 + Source/MQTTnet/Server/MqttServer.cs | 38 +++++++++++++++++-- 4 files changed, 40 insertions(+), 6 deletions(-) diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index 67631049..883446ea 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1,4 +1,5 @@ -* [Client] Added support for passing the local endpoint which should be used (network card). +* [Client] Added support for passing the local endpoint (network card) which should be used (#1311). * [Client] Exposed _PackageIdentifier_ in _MqttApplicationMessageReceivedEventArgs_ (thanks to @koepalex, #1466). * [Server] Added a new event (_ClientAcknowledgedPublishPacketAsync_) which is fired whenever a client acknowledges a PUBLISH packet (QoS 1 and 2, #487). * [Server] Exposed channel adapter (HTTP context etc.) in connection validation (#1125). +* [Server] The event _InterceptingPublishAsync_ is now also called for injected application messages (#Jeanot-Zubler). diff --git a/Samples/Server/Server_Simple_Samples.cs b/Samples/Server/Server_Simple_Samples.cs index 0ea4b6b7..167aab64 100644 --- a/Samples/Server/Server_Simple_Samples.cs +++ b/Samples/Server/Server_Simple_Samples.cs @@ -138,7 +138,7 @@ public static class Server_Simple_Samples e.ReasonCode = MqttConnectReasonCode.ClientIdentifierNotValid; } - if (e.UserName != "ValidUser") + if (e.Username != "ValidUser") { e.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword; } @@ -216,4 +216,4 @@ public static class Server_Simple_Samples } } } -} \ No newline at end of file +} diff --git a/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs b/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs index d363ca54..e60a15e8 100644 --- a/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs +++ b/Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs @@ -26,6 +26,7 @@ namespace MQTTnet.Server public AsyncEvent InterceptingOutboundPacketEvent { get; } = new AsyncEvent(); public AsyncEvent InterceptingPublishEvent { get; } = new AsyncEvent(); + public AsyncEvent InterceptingSubscriptionEvent { get; } = new AsyncEvent(); public AsyncEvent InterceptingUnsubscriptionEvent { get; } = new AsyncEvent(); diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 9903be8a..054467a7 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -3,6 +3,8 @@ // See the LICENSE file in the project root for more information. using System; +using System.Collections; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; using System.Threading; @@ -19,7 +21,8 @@ namespace MQTTnet.Server public class MqttServer : Disposable { readonly MqttServerEventContainer _eventContainer = new MqttServerEventContainer(); - + + readonly IDictionary _sessionItems = new ConcurrentDictionary(); readonly ICollection _adapters; readonly MqttNetSourceLogger _logger; readonly MqttServerOptions _options; @@ -205,7 +208,7 @@ namespace MQTTnet.Server return _clientSessionsManager.GetSessionStatusAsync(); } - public Task InjectApplicationMessage(InjectedMqttApplicationMessage injectedApplicationMessage) + public async Task InjectApplicationMessage(InjectedMqttApplicationMessage injectedApplicationMessage) { if (injectedApplicationMessage == null) { @@ -221,7 +224,36 @@ namespace MQTTnet.Server ThrowIfNotStarted(); - return _clientSessionsManager.DispatchApplicationMessage(injectedApplicationMessage.SenderClientId, injectedApplicationMessage.ApplicationMessage); + var processPublish = true; + var applicationMessage = injectedApplicationMessage.ApplicationMessage; + + if (_eventContainer.InterceptingPublishEvent.HasHandlers) + { + var interceptingPublishEventArgs = new InterceptingPublishEventArgs + { + ApplicationMessage = applicationMessage, + CancellationToken = _cancellationTokenSource.Token, + ClientId = injectedApplicationMessage.SenderClientId, + SessionItems = _sessionItems + }; + + await _eventContainer.InterceptingPublishEvent.InvokeAsync(interceptingPublishEventArgs).ConfigureAwait(false); + + applicationMessage = interceptingPublishEventArgs.ApplicationMessage; + processPublish = interceptingPublishEventArgs.ProcessPublish; + } + + if (!processPublish) + { + return; + } + + if (string.IsNullOrEmpty(applicationMessage.Topic)) + { + throw new NotSupportedException("Injected application messages must contain a topic. Topic alias is not supported."); + } + + await _clientSessionsManager.DispatchApplicationMessage(injectedApplicationMessage.SenderClientId, applicationMessage).ConfigureAwait(false); } public async Task StartAsync() From 29a9941d86e6d82d5505d11a25911cb8ce34df0e Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Sun, 31 Jul 2022 11:10:38 +0200 Subject: [PATCH 4/4] Fix breaking changes. --- Source/MQTTnet/Server/MqttServer.cs | 64 ++++++++++++++--------------- 1 file changed, 31 insertions(+), 33 deletions(-) diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 054467a7..37115b99 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -20,19 +20,19 @@ namespace MQTTnet.Server { public class MqttServer : Disposable { - readonly MqttServerEventContainer _eventContainer = new MqttServerEventContainer(); - - readonly IDictionary _sessionItems = new ConcurrentDictionary(); readonly ICollection _adapters; + readonly MqttClientSessionsManager _clientSessionsManager; + readonly MqttServerEventContainer _eventContainer = new MqttServerEventContainer(); + readonly MqttServerKeepAliveMonitor _keepAliveMonitor; readonly MqttNetSourceLogger _logger; readonly MqttServerOptions _options; - readonly IMqttNetLogger _rootLogger; readonly MqttRetainedMessagesManager _retainedMessagesManager; - readonly MqttServerKeepAliveMonitor _keepAliveMonitor; - readonly MqttClientSessionsManager _clientSessionsManager; - + readonly IMqttNetLogger _rootLogger; + + readonly IDictionary _sessionItems = new ConcurrentDictionary(); + CancellationTokenSource _cancellationTokenSource; - + public MqttServer(MqttServerOptions options, IEnumerable adapters, IMqttNetLogger logger) { _options = options ?? throw new ArgumentNullException(nameof(options)); @@ -58,18 +58,18 @@ namespace MQTTnet.Server remove => _eventContainer.ApplicationMessageNotConsumedEvent.RemoveHandler(value); } - public event Func ClientConnectedAsync - { - add => _eventContainer.ClientConnectedEvent.AddHandler(value); - remove => _eventContainer.ClientConnectedEvent.RemoveHandler(value); - } - public event Func ClientAcknowledgedPublishPacketAsync { add => _eventContainer.ClientAcknowledgedPublishPacketEvent.AddHandler(value); remove => _eventContainer.ClientAcknowledgedPublishPacketEvent.RemoveHandler(value); } + public event Func ClientConnectedAsync + { + add => _eventContainer.ClientConnectedEvent.AddHandler(value); + remove => _eventContainer.ClientConnectedEvent.RemoveHandler(value); + } + public event Func ClientDisconnectedAsync { add => _eventContainer.ClientDisconnectedEvent.AddHandler(value); @@ -167,7 +167,7 @@ namespace MQTTnet.Server } public bool IsStarted => _cancellationTokenSource != null; - + public Task DeleteRetainedMessagesAsync() { ThrowIfNotStarted(); @@ -183,10 +183,10 @@ namespace MQTTnet.Server } ThrowIfNotStarted(); - + return _clientSessionsManager.GetClient(id).StopAsync(reasonCode); } - + public Task> GetClientsAsync() { ThrowIfNotStarted(); @@ -226,17 +226,15 @@ namespace MQTTnet.Server var processPublish = true; var applicationMessage = injectedApplicationMessage.ApplicationMessage; - + if (_eventContainer.InterceptingPublishEvent.HasHandlers) { - var interceptingPublishEventArgs = new InterceptingPublishEventArgs - { - ApplicationMessage = applicationMessage, - CancellationToken = _cancellationTokenSource.Token, - ClientId = injectedApplicationMessage.SenderClientId, - SessionItems = _sessionItems - }; - + var interceptingPublishEventArgs = new InterceptingPublishEventArgs( + applicationMessage, + _cancellationTokenSource.Token, + injectedApplicationMessage.SenderClientId, + _sessionItems); + await _eventContainer.InterceptingPublishEvent.InvokeAsync(interceptingPublishEventArgs).ConfigureAwait(false); applicationMessage = interceptingPublishEventArgs.ApplicationMessage; @@ -247,12 +245,12 @@ namespace MQTTnet.Server { return; } - + if (string.IsNullOrEmpty(applicationMessage.Topic)) { throw new NotSupportedException("Injected application messages must contain a topic. Topic alias is not supported."); } - + await _clientSessionsManager.DispatchApplicationMessage(injectedApplicationMessage.SenderClientId, applicationMessage).ConfigureAwait(false); } @@ -274,7 +272,7 @@ namespace MQTTnet.Server } await _eventContainer.StartedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false); - + _logger.Info("Started."); } @@ -304,7 +302,7 @@ namespace MQTTnet.Server } await _eventContainer.StoppedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false); - + _logger.Info("Stopped."); } @@ -348,7 +346,7 @@ namespace MQTTnet.Server return _clientSessionsManager.UnsubscribeAsync(clientId, topicFilters); } - + public Task UpdateRetainedMessageAsync(MqttApplicationMessage retainedMessage) { if (retainedMessage == null) @@ -385,7 +383,7 @@ namespace MQTTnet.Server void ThrowIfNotStarted() { ThrowIfDisposed(); - + if (_cancellationTokenSource == null) { throw new InvalidOperationException("The MQTT server is not started."); @@ -395,7 +393,7 @@ namespace MQTTnet.Server void ThrowIfStarted() { ThrowIfDisposed(); - + if (_cancellationTokenSource != null) { throw new InvalidOperationException("The MQTT server is already started.");