diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index 72eb8188..f2cc64d7 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -4,3 +4,4 @@ * [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). +* [Server] Exposed session items in multipe events (#1291). 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 483ff8af..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); @@ -365,14 +357,7 @@ 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) @@ -393,7 +378,7 @@ namespace MQTTnet.Server while (!cancellationToken.IsCancellationRequested) { await Task.Yield(); - + var packet = await ChannelAdapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false); if (packet == null) { @@ -404,14 +389,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; diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index cea0e381..eb2359c7 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); } } diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 054467a7..3e1f1671 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -24,15 +24,18 @@ namespace MQTTnet.Server 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 +61,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 +170,7 @@ namespace MQTTnet.Server } public bool IsStarted => _cancellationTokenSource != null; - + public Task DeleteRetainedMessagesAsync() { ThrowIfNotStarted(); @@ -183,10 +186,10 @@ namespace MQTTnet.Server } ThrowIfNotStarted(); - + return _clientSessionsManager.GetClient(id).StopAsync(reasonCode); } - + public Task> GetClientsAsync() { ThrowIfNotStarted(); @@ -226,17 +229,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; @@ -274,7 +275,7 @@ namespace MQTTnet.Server } await _eventContainer.StartedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false); - + _logger.Info("Started."); } @@ -304,7 +305,7 @@ namespace MQTTnet.Server } await _eventContainer.StoppedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false); - + _logger.Info("Stopped."); } @@ -348,7 +349,7 @@ namespace MQTTnet.Server return _clientSessionsManager.UnsubscribeAsync(clientId, topicFilters); } - + public Task UpdateRetainedMessageAsync(MqttApplicationMessage retainedMessage) { if (retainedMessage == null) @@ -385,7 +386,7 @@ namespace MQTTnet.Server void ThrowIfNotStarted() { ThrowIfDisposed(); - + if (_cancellationTokenSource == null) { throw new InvalidOperationException("The MQTT server is not started."); @@ -395,7 +396,7 @@ namespace MQTTnet.Server void ThrowIfStarted() { ThrowIfDisposed(); - + if (_cancellationTokenSource != null) { throw new InvalidOperationException("The MQTT server is already started.");