YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-19 11:03:07 +08:00
Code Issues Packages Projects Releases Wiki Activity

Allow interception of will messages (#1613)

Browse Source
This commit is contained in:
Christian
2022-12-06 20:28:01 +01:00
committed by GitHub
parent c17a350796
commit d7eb4bbe6e
10 changed files with 230 additions and 215 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Server/Internal/DispatchApplicationMessageResult.cs
+15 -3
View File
@@ -2,15 +2,27 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Collections.Generic;
using MQTTnet.Packets;
namespace MQTTnet.Server
{
public sealed class DispatchApplicationMessageResult
{
public DispatchApplicationMessageResult(int matchingSubscribersCount)
public DispatchApplicationMessageResult(int reasonCode, bool closeConnection, string reasonString, List<MqttUserProperty> userProperties)
{
MatchingSubscribersCount = matchingSubscribersCount;
ReasonCode = reasonCode;
CloseConnection = closeConnection;
ReasonString = reasonString;
UserProperties = userProperties;
}
public int MatchingSubscribersCount { get; }
public bool CloseConnection { get; }
public int ReasonCode { get; }
public string ReasonString { get; }
public List<MqttUserProperty> UserProperties { get; }
}
}
Source/MQTTnet/Server/Internal/MqttClient.cs
+9 -33
View File
@@ -116,7 +116,7 @@ namespace MQTTnet.Server
var willPublishPacket = MqttPacketFactories.Publish.Create(Session.LatestConnectPacket);
var willApplicationMessage = MqttApplicationMessageFactory.Create(willPublishPacket);
_ = _sessionsManager.DispatchApplicationMessage(Id, willApplicationMessage);
_ = _sessionsManager.DispatchApplicationMessage(Id, Session.Items, willApplicationMessage, CancellationToken.None);
Session.WillMessageSent = true;
_logger.Info("Client '{0}': Published will message", Id);
@@ -210,41 +210,17 @@ namespace MQTTnet.Server
{
HandleTopicAlias(publishPacket);
InterceptingPublishEventArgs interceptingPublishEventArgs = null;
var applicationMessage = MqttApplicationMessageFactory.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;
}
var dispatchApplicationMessageResult =
await _sessionsManager.DispatchApplicationMessage(Id, Session.Items, applicationMessage, cancellationToken).ConfigureAwait(false);
await _eventContainer.InterceptingPublishEvent.InvokeAsync(interceptingPublishEventArgs).ConfigureAwait(false);
applicationMessage = interceptingPublishEventArgs.ApplicationMessage;
closeConnection = interceptingPublishEventArgs.CloseConnection;
processPublish = interceptingPublishEventArgs.ProcessPublish;
}
if (closeConnection)
if (dispatchApplicationMessageResult.CloseConnection)
{
await StopAsync(MqttDisconnectReasonCode.UnspecifiedError);
return;
}
DispatchApplicationMessageResult dispatchResult = null;
if (processPublish && applicationMessage != null)
{
dispatchResult = await _sessionsManager.DispatchApplicationMessage(Id, applicationMessage).ConfigureAwait(false);
}
switch (publishPacket.QualityOfServiceLevel)
{
case MqttQualityOfServiceLevel.AtMostOnce:
@@ -254,13 +230,13 @@ namespace MQTTnet.Server
}
case MqttQualityOfServiceLevel.AtLeastOnce:
{
var pubAckPacket = MqttPacketFactories.PubAck.Create(publishPacket, interceptingPublishEventArgs, dispatchResult);
var pubAckPacket = MqttPacketFactories.PubAck.Create(publishPacket, dispatchApplicationMessageResult);
Session.EnqueueControlPacket(new MqttPacketBusItem(pubAckPacket));
break;
}
case MqttQualityOfServiceLevel.ExactlyOnce:
{
var pubRecPacket = MqttPacketFactories.PubRec.Create(publishPacket, interceptingPublishEventArgs, dispatchResult);
var pubRecPacket = MqttPacketFactories.PubRec.Create(publishPacket, dispatchApplicationMessageResult);
Session.EnqueueControlPacket(new MqttPacketBusItem(pubRecPacket));
break;
}
@@ -478,16 +454,16 @@ namespace MQTTnet.Server
}
var logLevel = MqttNetLogLevel.Error;
if (!IsRunning)
{
// There was an exception but the connection is already closed. So there is no chance to send a response to the client etc.
logLevel = MqttNetLogLevel.Warning;
}
if (currentPacket == null)
{
_logger.Publish(logLevel, exception, "Client '{0}': Error while receiving packets", Id);
_logger.Publish(logLevel, exception, "Client '{0}': Error while receiving packets", Id);
}
else
{
Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
+112 -79
View File
@@ -126,81 +126,118 @@ namespace MQTTnet.Server
_logger.Verbose("Session for client '{0}' deleted.", clientId);
}
public async Task<DispatchApplicationMessageResult> DispatchApplicationMessage(string senderId, MqttApplicationMessage applicationMessage)
public async Task<DispatchApplicationMessageResult> DispatchApplicationMessage(
string senderId,
IDictionary senderSessionItems,
MqttApplicationMessage applicationMessage,
CancellationToken cancellationToken)
{
var matchingSubscribersCount = 0;
try
var processPublish = true;
var closeConnection = false;
string reasonString = null;
List<MqttUserProperty> userProperties = null;
var reasonCode = 0; // The reason code is later converted into several different but compatible enums!
// Allow the user to intercept application message...
if (_eventContainer.InterceptingPublishEvent.HasHandlers)
{
if (applicationMessage.Retain)
var interceptingPublishEventArgs = new InterceptingPublishEventArgs(applicationMessage, cancellationToken, senderId, senderSessionItems);
if (string.IsNullOrEmpty(interceptingPublishEventArgs.ApplicationMessage.Topic))
{
await _retainedMessagesManager.UpdateMessage(senderId, applicationMessage).ConfigureAwait(false);
}
List<MqttSession> subscriberSessions;
lock (_sessionsManagementLock)
{
subscriberSessions = _subscriberSessions.ToList();
// 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;
}
// Calculate application message topic hash once for subscription checks
MqttSubscription.CalculateTopicHash(applicationMessage.Topic, out var topicHash, out _, out _);
await _eventContainer.InterceptingPublishEvent.InvokeAsync(interceptingPublishEventArgs).ConfigureAwait(false);
foreach (var session in subscriberSessions)
{
if(!session.TryCheckSubscriptions(
applicationMessage.Topic,
topicHash,
applicationMessage.QualityOfServiceLevel,
senderId,
out var checkSubscriptionsResult))
{
// Checking the subscriptions has failed for the session. The session
// will be ignored.
continue;
}
if (!checkSubscriptionsResult.IsSubscribed)
{
continue;
}
var newPublishPacket = MqttPacketFactories.Publish.Create(applicationMessage);
newPublishPacket.QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel;
newPublishPacket.SubscriptionIdentifiers = checkSubscriptionsResult.SubscriptionIdentifiers;
if (newPublishPacket.QualityOfServiceLevel > 0)
{
newPublishPacket.PacketIdentifier = session.PacketIdentifierProvider.GetNextPacketIdentifier();
}
if (checkSubscriptionsResult.RetainAsPublished)
{
// Transfer the original retain state from the publisher. This is a MQTTv5 feature.
newPublishPacket.Retain = applicationMessage.Retain;
}
else
{
newPublishPacket.Retain = false;
}
session.EnqueueDataPacket(new MqttPacketBusItem(newPublishPacket));
matchingSubscribersCount++;
_logger.Verbose("Client '{0}': Queued PUBLISH packet with topic '{1}'.", session.Id, applicationMessage.Topic);
}
if (matchingSubscribersCount == 0)
{
await FireApplicationMessageNotConsumedEvent(applicationMessage, senderId).ConfigureAwait(false);
}
}
catch (Exception exception)
{
_logger.Error(exception, "Unhandled exception while processing next queued application message.");
applicationMessage = interceptingPublishEventArgs.ApplicationMessage;
closeConnection = interceptingPublishEventArgs.CloseConnection;
processPublish = interceptingPublishEventArgs.ProcessPublish;
reasonString = interceptingPublishEventArgs.Response.ReasonString;
userProperties = interceptingPublishEventArgs.Response.UserProperties;
reasonCode = (int)interceptingPublishEventArgs.Response.ReasonCode;
}
return new DispatchApplicationMessageResult(matchingSubscribersCount);
// Process the application message...
if (processPublish && applicationMessage != null)
{
var matchingSubscribersCount = 0;
try
{
if (applicationMessage.Retain)
{
await _retainedMessagesManager.UpdateMessage(senderId, applicationMessage).ConfigureAwait(false);
}
List<MqttSession> subscriberSessions;
lock (_sessionsManagementLock)
{
subscriberSessions = _subscriberSessions.ToList();
}
// Calculate application message topic hash once for subscription checks
MqttSubscription.CalculateTopicHash(applicationMessage.Topic, out var topicHash, out _, out _);
foreach (var session in subscriberSessions)
{
if (!session.TryCheckSubscriptions(
applicationMessage.Topic,
topicHash,
applicationMessage.QualityOfServiceLevel,
senderId,
out var checkSubscriptionsResult))
{
// Checking the subscriptions has failed for the session. The session
// will be ignored.
continue;
}
if (!checkSubscriptionsResult.IsSubscribed)
{
continue;
}
var publishPacketCopy = MqttPacketFactories.Publish.Create(applicationMessage);
publishPacketCopy.QualityOfServiceLevel = checkSubscriptionsResult.QualityOfServiceLevel;
publishPacketCopy.SubscriptionIdentifiers = checkSubscriptionsResult.SubscriptionIdentifiers;
if (publishPacketCopy.QualityOfServiceLevel > 0)
{
publishPacketCopy.PacketIdentifier = session.PacketIdentifierProvider.GetNextPacketIdentifier();
}
if (checkSubscriptionsResult.RetainAsPublished)
{
// Transfer the original retain state from the publisher. This is a MQTTv5 feature.
publishPacketCopy.Retain = applicationMessage.Retain;
}
else
{
publishPacketCopy.Retain = false;
}
session.EnqueueDataPacket(new MqttPacketBusItem(publishPacketCopy));
matchingSubscribersCount++;
_logger.Verbose("Client '{0}': Queued PUBLISH packet with topic '{1}'.", session.Id, applicationMessage.Topic);
}
if (matchingSubscribersCount == 0)
{
reasonCode = (int)MqttPubAckReasonCode.NoMatchingSubscribers;
await FireApplicationMessageNotConsumedEvent(applicationMessage, senderId).ConfigureAwait(false);
}
}
catch (Exception exception)
{
_logger.Error(exception, "Unhandled exception while processing next queued application message.");
}
}
return new DispatchApplicationMessageResult(reasonCode, closeConnection, reasonString, userProperties);
}
public void Dispose()
@@ -495,11 +532,11 @@ namespace MQTTnet.Server
{
MqttSession oldSession;
MqttClient oldClient;
lock (_sessionsManagementLock)
{
MqttSession session;
// Create a new session (if required).
if (!_sessions.TryGetValue(connectPacket.ClientId, out oldSession))
{
@@ -518,7 +555,7 @@ namespace MQTTnet.Server
_logger.Verbose("Reusing existing session of client '{0}'.", connectPacket.ClientId);
session = oldSession;
oldSession = null;
// Session persistence could change for MQTT 5 clients that reconnect with different SessionExpiryInterval
session.IsPersistent = sessionShouldPersist;
connAckPacket.IsSessionPresent = true;
@@ -527,7 +564,7 @@ namespace MQTTnet.Server
}
_sessions[connectPacket.ClientId] = session;
// Create a new client (always required).
_clients.TryGetValue(connectPacket.ClientId, out oldClient);
if (oldClient != null)
@@ -536,7 +573,7 @@ namespace MQTTnet.Server
// for a later DISCONNECT packet.
oldClient.IsTakenOver = true;
}
client = CreateClient(connectPacket, channelAdapter, session);
_clients[connectPacket.ClientId] = client;
@@ -548,19 +585,15 @@ namespace MQTTnet.Server
var preparingSessionEventArgs = new PreparingSessionEventArgs();
await _eventContainer.PreparingSessionEvent.TryInvokeAsync(preparingSessionEventArgs, _logger).ConfigureAwait(false);
}
if (oldClient != null)
{
await oldClient.StopAsync(MqttDisconnectReasonCode.SessionTakenOver).ConfigureAwait(false);
if (_eventContainer.ClientDisconnectedEvent.HasHandlers)
{
var eventArgs = new ClientDisconnectedEventArgs(
oldClient.Id,
MqttClientDisconnectType.Takeover,
oldClient.Endpoint,
oldClient.Session.Items);
var eventArgs = new ClientDisconnectedEventArgs(oldClient.Id, MqttClientDisconnectType.Takeover, oldClient.Endpoint, oldClient.Session.Items);
await _eventContainer.ClientDisconnectedEvent.TryInvokeAsync(eventArgs, _logger).ConfigureAwait(false);
}
}
Source/MQTTnet/Server/MqttServer.cs
+11 -29
View File
@@ -19,17 +19,17 @@ namespace MQTTnet.Server
{
public class MqttServer : Disposable
{
readonly MqttServerEventContainer _eventContainer = new MqttServerEventContainer();
readonly IDictionary _sessionItems = new ConcurrentDictionary<object, object>();
readonly ICollection<IMqttServerAdapter> _adapters;
readonly MqttClientSessionsManager _clientSessionsManager;
readonly MqttServerEventContainer _eventContainer = new MqttServerEventContainer();
readonly MqttServerKeepAliveMonitor _keepAliveMonitor;
readonly MqttNetSourceLogger _logger;
readonly MqttServerOptions _options;
readonly MqttRetainedMessagesManager _retainedMessagesManager;
readonly IMqttNetLogger _rootLogger;
readonly IDictionary _sessionItems = new ConcurrentDictionary<object, object>();
CancellationTokenSource _cancellationTokenSource;
public MqttServer(MqttServerOptions options, IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
@@ -207,7 +207,7 @@ namespace MQTTnet.Server
return _clientSessionsManager.GetSessionStatusAsync();
}
public async Task InjectApplicationMessage(InjectedMqttApplicationMessage injectedApplicationMessage)
public Task InjectApplicationMessage(InjectedMqttApplicationMessage injectedApplicationMessage, CancellationToken cancellationToken = default)
{
if (injectedApplicationMessage == null)
{
@@ -223,34 +223,16 @@ namespace MQTTnet.Server
ThrowIfNotStarted();
var processPublish = true;
var applicationMessage = injectedApplicationMessage.ApplicationMessage;
if (_eventContainer.InterceptingPublishEvent.HasHandlers)
{
var interceptingPublishEventArgs = new InterceptingPublishEventArgs(
applicationMessage,
_cancellationTokenSource.Token,
injectedApplicationMessage.SenderClientId,
_sessionItems);
await _eventContainer.InterceptingPublishEvent.InvokeAsync(interceptingPublishEventArgs).ConfigureAwait(false);
applicationMessage = interceptingPublishEventArgs.ApplicationMessage;
processPublish = interceptingPublishEventArgs.ProcessPublish;
}
if (!processPublish)
{
return;
}
if (string.IsNullOrEmpty(applicationMessage.Topic))
if (string.IsNullOrEmpty(injectedApplicationMessage.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);
return _clientSessionsManager.DispatchApplicationMessage(
injectedApplicationMessage.SenderClientId,
_sessionItems,
injectedApplicationMessage.ApplicationMessage,
cancellationToken);
}
public async Task StartAsync()
Source/MQTTnet/Server/PublishResponse.cs
+1 -1
View File
@@ -14,6 +14,6 @@ namespace MQTTnet.Server
public string ReasonString { get; set; }
public List<MqttUserProperty> UserProperties { get; }
public List<MqttUserProperty> UserProperties { get; set; }
}
}