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()