diff --git a/Source/MQTTnet.Extensions.ManagedClient/InterceptingPublishMessageEventArgs.cs b/Source/MQTTnet.Extensions.ManagedClient/InterceptingPublishMessageEventArgs.cs new file mode 100644 index 00000000..c688082f --- /dev/null +++ b/Source/MQTTnet.Extensions.ManagedClient/InterceptingPublishMessageEventArgs.cs @@ -0,0 +1,20 @@ +// Licensed to the .NET Foundation under one or more agreements. +// 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; + +namespace MQTTnet.Extensions.ManagedClient +{ + public sealed class InterceptingPublishMessageEventArgs : EventArgs + { + public InterceptingPublishMessageEventArgs(ManagedMqttApplicationMessage applicationMessage) + { + ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage)); + } + + public ManagedMqttApplicationMessage ApplicationMessage { get; } + + public bool AcceptPublish { get; set; } = true; + } +} diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 3690be8c..32a93807 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -19,6 +19,7 @@ namespace MQTTnet.Extensions.ManagedClient { public sealed class ManagedMqttClient : Disposable, IManagedMqttClient { + readonly AsyncEvent _interceptingPublishMessageEvent = new AsyncEvent(); readonly AsyncEvent _applicationMessageProcessedEvent = new AsyncEvent(); readonly AsyncEvent _applicationMessageSkippedEvent = new AsyncEvent(); readonly AsyncEvent _connectingFailedEvent = new AsyncEvent(); @@ -88,6 +89,14 @@ namespace MQTTnet.Extensions.ManagedClient remove => _applicationMessageProcessedEvent.RemoveHandler(value); } + + public event Func InterceptPublishMessageAsync + { + add => _interceptingPublishMessageEvent.AddHandler(value); + remove => _interceptingPublishMessageEvent.RemoveHandler(value); + } + + public event Func ApplicationMessageReceivedAsync { add => InternalClient.ApplicationMessageReceivedAsync += value; @@ -700,9 +709,20 @@ namespace MQTTnet.Extensions.ManagedClient async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage message) { Exception transmitException = null; + bool acceptPublish = true; try { - await InternalClient.PublishAsync(message.ApplicationMessage).ConfigureAwait(false); + if (_interceptingPublishMessageEvent.HasHandlers) + { + var interceptEventArgs = new InterceptingPublishMessageEventArgs(message); + await _interceptingPublishMessageEvent.InvokeAsync(interceptEventArgs).ConfigureAwait(false); + acceptPublish = interceptEventArgs.AcceptPublish; + } + + if (acceptPublish) + { + await InternalClient.PublishAsync(message.ApplicationMessage).ConfigureAwait(false); + } using (await _messageQueueLock.EnterAsync().ConfigureAwait(false)) //lock to avoid conflict with this.PublishAsync { diff --git a/Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs b/Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs index 5726b148..e8e10005 100644 --- a/Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs +++ b/Source/MQTTnet.Tests/Clients/ManagedMqttClient/ManagedMqttClient_Tests.cs @@ -437,6 +437,109 @@ namespace MQTTnet.Tests.Clients.ManagedMqttClient } } + + [TestMethod] + public async Task ManagedClients_CanInterceptPublishedMessage_PreventingPublish() + { + using (var testEnvironment = CreateTestEnvironment()) + { + await testEnvironment.StartServer(); + + var receivedMessagesCount = 0; + var interceptedMessagesCount = 0; + + var clientOptions = new MqttClientOptionsBuilder() + .WithTcpServer("127.0.0.1", testEnvironment.ServerPort) + .Build(); + + var sendingClient = testEnvironment.CreateClient(); + var sendingManagedClient = new MQTTnet.Extensions.ManagedClient.ManagedMqttClient(sendingClient, testEnvironment.ClientLogger); + + await sendingManagedClient.StartAsync(new ManagedMqttClientOptionsBuilder().WithClientOptions(clientOptions).Build()); + + // Wait until the managed client is fully set up and running. + await Task.Delay(1000); + + var receivingClient = await testEnvironment.ConnectClient(); + + receivingClient.ApplicationMessageReceivedAsync += e => + { + Interlocked.Increment(ref receivedMessagesCount); + return CompletedTask.Instance; + }; + sendingManagedClient.InterceptPublishMessageAsync += e => + { + Interlocked.Increment(ref interceptedMessagesCount); + e.AcceptPublish = false; + return CompletedTask.Instance; + }; + + await receivingClient.SubscribeAsync("My/last/will"); + + // Wait for arrival of the will message at the receiver. + await Task.Delay(5000); + + await sendingManagedClient.EnqueueAsync("My/last/will", "hello world"); + + // Wait for arrival of the will message at the receiver. + await Task.Delay(5000); + + Assert.AreEqual(0, receivedMessagesCount); + Assert.AreEqual(1, interceptedMessagesCount); + } + } + + [TestMethod] + public async Task ManagedClients_CanInterceptPublishedMessage_AllowingPublish() + { + using (var testEnvironment = CreateTestEnvironment()) + { + await testEnvironment.StartServer(); + + var receivedMessagesCount = 0; + var interceptedMessagesCount = 0; + + var clientOptions = new MqttClientOptionsBuilder() + .WithTcpServer("127.0.0.1", testEnvironment.ServerPort) + .Build(); + + var sendingClient = testEnvironment.CreateClient(); + var sendingManagedClient = new MQTTnet.Extensions.ManagedClient.ManagedMqttClient(sendingClient, testEnvironment.ClientLogger); + + await sendingManagedClient.StartAsync(new ManagedMqttClientOptionsBuilder().WithClientOptions(clientOptions).Build()); + + // Wait until the managed client is fully set up and running. + await Task.Delay(1000); + + var receivingClient = await testEnvironment.ConnectClient(); + + receivingClient.ApplicationMessageReceivedAsync += e => + { + Interlocked.Increment(ref receivedMessagesCount); + return CompletedTask.Instance; + }; + sendingManagedClient.InterceptPublishMessageAsync += e => + { + Interlocked.Increment(ref interceptedMessagesCount); + e.AcceptPublish = true; + return CompletedTask.Instance; + }; + + await receivingClient.SubscribeAsync("My/last/will"); + + // Wait for arrival of the will message at the receiver. + await Task.Delay(5000); + + await sendingManagedClient.EnqueueAsync("My/last/will", "hello world"); + + // Wait for arrival of the will message at the receiver. + await Task.Delay(5000); + + Assert.AreEqual(1, receivedMessagesCount); + Assert.AreEqual(1, interceptedMessagesCount); + } + } + // This case also serves as a regression test for the previous behavior which re-published // each and every existing subscriptions with every new subscription that was made // (causing performance problems and having the visible symptom of retained messages being received again)