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

Squashed commit of the following:

Browse Source 
commit 469caf656e
Author: Christian <6939810+chkr1011@users.noreply.github.com>
Date:   Sun Jul 31 10:55:40 2022 +0200

    Fix build errors.

commit 42402fba04
Author: Christian <6939810+chkr1011@users.noreply.github.com>
Date:   Sun Jul 31 10:22:23 2022 +0200

    Allow interception of injected application messages.

commit a0dc8ab0d7
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
This commit is contained in:
Christian
2022-07-31 11:09:39 +02:00
parent 02ede5a2d8
commit 5a21939e29
4 changed files with 40 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs
+1
View File
@@ -26,6 +26,7 @@ namespace MQTTnet.Server
public AsyncEvent<InterceptingPacketEventArgs> InterceptingOutboundPacketEvent { get; } = new AsyncEvent<InterceptingPacketEventArgs>();
public AsyncEvent<InterceptingPublishEventArgs> InterceptingPublishEvent { get; } = new AsyncEvent<InterceptingPublishEventArgs>();
public AsyncEvent<InterceptingSubscriptionEventArgs> InterceptingSubscriptionEvent { get; } = new AsyncEvent<InterceptingSubscriptionEventArgs>();
public AsyncEvent<InterceptingUnsubscriptionEventArgs> InterceptingUnsubscriptionEvent { get; } = new AsyncEvent<InterceptingUnsubscriptionEventArgs>();
Source/MQTTnet/Server/MqttServer.cs
+35 -3
View File
@@ -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<object, object>();
readonly ICollection<IMqttServerAdapter> _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()