Add an event to receive all application messages at server level.

This commit is contained in:
Christian Kratky
2017-07-13 20:28:49 +02:00
parent 2ff449b241
commit 98d7bc980c
10 changed files with 110 additions and 14 deletions
@@ -0,0 +1,17 @@
using System;
namespace MQTTnet.Core.Server
{
public sealed class MqttApplicationMessageReceivedEventArgs : EventArgs
{
public MqttApplicationMessageReceivedEventArgs(string clientId, MqttApplicationMessage applicationMessage)
{
ClientId = clientId;
ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
}
public string ClientId { get; }
public MqttApplicationMessage ApplicationMessage { get; }
}
}
+5 -2
View File
@@ -19,20 +19,23 @@ namespace MQTTnet.Core.Server
private readonly MqttClientMessageQueue _messageQueue;
private readonly Action<MqttClientSession, MqttPublishPacket> _publishPacketReceivedCallback;
private readonly MqttServerOptions _options;
private CancellationTokenSource _cancellationTokenSource;
private IMqttCommunicationAdapter _adapter;
private string _identifier;
private MqttApplicationMessage _willApplicationMessage;
public MqttClientSession(MqttServerOptions options, Action<MqttClientSession, MqttPublishPacket> publishPacketReceivedCallback)
public MqttClientSession(string clientId, MqttServerOptions options, Action<MqttClientSession, MqttPublishPacket> publishPacketReceivedCallback)
{
ClientId = clientId;
_options = options ?? throw new ArgumentNullException(nameof(options));
_publishPacketReceivedCallback = publishPacketReceivedCallback ?? throw new ArgumentNullException(nameof(publishPacketReceivedCallback));
_messageQueue = new MqttClientMessageQueue(options);
}
public string ClientId { get; }
public bool IsConnected => _adapter != null;
public async Task RunAsync(string identifier, MqttApplicationMessage willApplicationMessage, IMqttCommunicationAdapter adapter)
@@ -5,6 +5,7 @@ using System.Threading.Tasks;
using MQTTnet.Core.Adapter;
using MQTTnet.Core.Diagnostics;
using MQTTnet.Core.Exceptions;
using MQTTnet.Core.Internal;
using MQTTnet.Core.Packets;
using MQTTnet.Core.Protocol;
@@ -21,6 +22,8 @@ namespace MQTTnet.Core.Server
_options = options ?? throw new ArgumentNullException(nameof(options));
}
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public async Task RunClientSessionAsync(MqttClientConnectedEventArgs eventArgs)
{
try
@@ -114,7 +117,7 @@ namespace MQTTnet.Core.Server
{
isExistingSession = false;
clientSession = new MqttClientSession(_options, DispatchPublishPacket);
clientSession = new MqttClientSession(connectPacket.ClientId, _options, DispatchPublishPacket);
_clientSessions[connectPacket.ClientId] = clientSession;
MqttTrace.Verbose(nameof(MqttClientSessionsManager), $"Created a new session for client '{connectPacket.ClientId}'.");
@@ -126,6 +129,9 @@ namespace MQTTnet.Core.Server
private void DispatchPublishPacket(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
{
var eventArgs = new MqttApplicationMessageReceivedEventArgs(senderClientSession.ClientId, publishPacket.ToApplicationMessage());
ApplicationMessageReceived?.Invoke(this, eventArgs);
foreach (var clientSession in _clientSessions.Values.ToList())
{
clientSession.EnqueuePublishPacket(senderClientSession, publishPacket);
+3
View File
@@ -21,6 +21,7 @@ namespace MQTTnet.Core.Server
_adapters = adapters ?? throw new ArgumentNullException(nameof(adapters));
_clientSessionsManager = new MqttClientSessionsManager(options);
_clientSessionsManager.ApplicationMessageReceived += (s, e) => ApplicationMessageReceived?.Invoke(s, e);
}
public IList<string> GetConnectedClients()
@@ -30,6 +31,8 @@ namespace MQTTnet.Core.Server
public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public void InjectClient(string identifier, IMqttCommunicationAdapter adapter)
{
if (adapter == null) throw new ArgumentNullException(nameof(adapter));