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

Migrate retained messages to the event based approach.

Browse Source
This commit is contained in:
Christian
2021-11-01 11:20:46 +01:00
parent 9ab85b3592
commit 22ccdf10c6
57 changed files with 451 additions and 666 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Formatter/MqttConnAckPacketFactory.cs
+1 -1
View File
@@ -6,7 +6,7 @@ namespace MQTTnet.Formatter
{
public sealed class MqttConnAckPacketFactory
{
public MqttConnAckPacket Create(ValidatingMqttClientConnectionEventArgs clientConnectionValidationEventArgs)
public MqttConnAckPacket Create(ValidatingConnectionEventArgs clientConnectionValidationEventArgs)
{
if (clientConnectionValidationEventArgs == null) throw new ArgumentNullException(nameof(clientConnectionValidationEventArgs));
Source/MQTTnet/Formatter/MqttPubAckPacketFactory.cs
+1 -1
View File
@@ -8,7 +8,7 @@ namespace MQTTnet.Formatter
{
public sealed class MqttPubAckPacketFactory
{
public MqttPubAckPacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageResponse applicationMessageResponse)
public MqttPubAckPacket Create(MqttPublishPacket publishPacket, PublishResponse applicationMessageResponse)
{
if (applicationMessageResponse == null) throw new ArgumentNullException(nameof(applicationMessageResponse));
Source/MQTTnet/Formatter/MqttPubRecPacketFactory.cs
+1 -1
View File
@@ -18,7 +18,7 @@ namespace MQTTnet.Formatter
return pubRecPacket;
}
public MqttBasePacket Create(MqttPublishPacket publishPacket, MqttApplicationMessageResponse applicationMessageResponse)
public MqttBasePacket Create(MqttPublishPacket publishPacket, PublishResponse applicationMessageResponse)
{
if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
Source/MQTTnet/Formatter/MqttSubAckPacketFactory.cs
+3 -3
View File
@@ -6,17 +6,17 @@ namespace MQTTnet.Formatter
{
public sealed class MqttSubAckPacketFactory
{
public MqttSubAckPacket Create(MqttSubscribePacket subscribePacket, MqttSubscribeResult mqttSubscribeResult)
public MqttSubAckPacket Create(MqttSubscribePacket subscribePacket, SubscribeResult subscribeResult)
{
if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));
if (mqttSubscribeResult == null) throw new ArgumentNullException(nameof(mqttSubscribeResult));
if (subscribeResult == null) throw new ArgumentNullException(nameof(subscribeResult));
var subAckPacket = new MqttSubAckPacket
{
PacketIdentifier = subscribePacket.PacketIdentifier
};
subAckPacket.ReasonCodes.AddRange(mqttSubscribeResult.ReasonCodes);
subAckPacket.ReasonCodes.AddRange(subscribeResult.ReasonCodes);
return subAckPacket;
}
Source/MQTTnet/MqttFactory.cs
+1
View File
@@ -7,6 +7,7 @@ using System;
using System.Collections.Generic;
using System.Linq;
using MQTTnet.Diagnostics;
using MqttClient = MQTTnet.Client.MqttClient;
namespace MQTTnet
{
Source/MQTTnet/Server/Events/MqttApplicationMessageNotConsumedEventArgs.cs → Source/MQTTnet/Server/Events/ApplicationMessageNotConsumedEventArgs.cs
+1 -1
View File
@@ -2,7 +2,7 @@
namespace MQTTnet.Server
{
public sealed class MqttApplicationMessageNotConsumedEventArgs : EventArgs
public sealed class ApplicationMessageNotConsumedEventArgs : EventArgs
{
/// <summary>
/// Gets the application message which was not consumed by any client.
Source/MQTTnet/Server/Events/MqttClientConnectedEventArgs.cs → Source/MQTTnet/Server/Events/ClientConnectedEventArgs.cs
+1 -1
View File
@@ -3,7 +3,7 @@ using MQTTnet.Formatter;
namespace MQTTnet.Server
{
public sealed class MqttServerClientConnectedEventArgs : EventArgs
public sealed class ClientConnectedEventArgs : EventArgs
{
/// <summary>
/// Gets the client identifier of the connected client.
Source/MQTTnet/Server/Events/MqttClientDisconnectedEventArgs.cs → Source/MQTTnet/Server/Events/ClientDisconnectedEventArgs.cs
+1 -1
View File
@@ -2,7 +2,7 @@ using System;
namespace MQTTnet.Server
{
public sealed class MqttServerClientDisconnectedEventArgs : EventArgs
public sealed class ClientDisconnectedEventArgs : EventArgs
{
/// <summary>
/// Gets the client identifier.
Source/MQTTnet/Server/Events/MqttClientSubscribedTopicEventArgs.cs → Source/MQTTnet/Server/Events/ClientSubscribedTopicEventArgs.cs
+1 -1
View File
@@ -3,7 +3,7 @@ using MQTTnet.Packets;
namespace MQTTnet.Server
{
public sealed class MqttServerClientSubscribedTopicEventArgs : EventArgs
public sealed class ClientSubscribedTopicEventArgs : EventArgs
{
/// <summary>
/// Gets the client identifier.
Source/MQTTnet/Server/Events/MqttClientUnsubscribedTopicEventArgs.cs → Source/MQTTnet/Server/Events/ClientUnsubscribedTopicEventArgs.cs
+1 -1
View File
@@ -2,7 +2,7 @@
namespace MQTTnet.Server
{
public sealed class MqttServerClientUnsubscribedTopicEventArgs : EventArgs
public sealed class ClientUnsubscribedTopicEventArgs : EventArgs
{
/// <summary>
/// Gets the client identifier.
Source/MQTTnet/Server/Events/InterceptingMqttClientPublishEventArgs.cs → Source/MQTTnet/Server/Events/InterceptingPublishEventArgs.cs
+2 -2
View File
@@ -4,7 +4,7 @@ using System.Threading;
namespace MQTTnet.Server
{
public sealed class InterceptingMqttClientPublishEventArgs : EventArgs
public sealed class InterceptingPublishEventArgs : EventArgs
{
/// <summary>
/// Gets the client identifier.
@@ -22,7 +22,7 @@ namespace MQTTnet.Server
/// <summary>
/// Gets the response which will be sent to the client via the PUBACK etc. packets.
/// </summary>
public MqttApplicationMessageResponse Response { get; } = new MqttApplicationMessageResponse();
public PublishResponse Response { get; } = new PublishResponse();
/// <summary>
/// Gets or sets whether the publish should be processed internally.
Source/MQTTnet/Server/Events/InterceptingMqttClientSubscriptionEventArgs.cs → Source/MQTTnet/Server/Events/InterceptingSubscriptionEventArgs.cs
+2 -2
View File
@@ -5,7 +5,7 @@ using MQTTnet.Packets;
namespace MQTTnet.Server
{
public sealed class InterceptingMqttClientSubscriptionEventArgs : EventArgs
public sealed class InterceptingSubscriptionEventArgs : EventArgs
{
/// <summary>
/// Gets the client identifier.
@@ -44,7 +44,7 @@ namespace MQTTnet.Server
/// <summary>
/// Gets the response which will be sent to the client via the SUBACK packet.
/// </summary>
public MqttSubscribeResponse Response { get; } = new MqttSubscribeResponse();
public SubscribeResponse Response { get; } = new SubscribeResponse();
/// <summary>
/// Gets the cancellation token which can indicate that the client connection gets down.
Source/MQTTnet/Server/Events/InterceptingMqttClientUnsubscriptionEventArgs.cs → Source/MQTTnet/Server/Events/InterceptingUnsubscriptionEventArgs.cs
+2 -2
View File
@@ -4,7 +4,7 @@ using System.Threading;
namespace MQTTnet.Server
{
public sealed class InterceptingMqttClientUnsubscriptionEventArgs : EventArgs
public sealed class InterceptingUnsubscriptionEventArgs : EventArgs
{
/// <summary>
/// Gets the client identifier.
@@ -27,7 +27,7 @@ namespace MQTTnet.Server
/// <summary>
/// Gets the response which will be sent to the client via the UNSUBACK pocket.
/// </summary>
public MqttUnsubscribeResponse Response { get; } = new MqttUnsubscribeResponse();
public UnsubscribeResponse Response { get; } = new UnsubscribeResponse();
/// <summary>
/// Gets or sets whether the broker should remove an internal subscription for the client.
Source/MQTTnet/Server/Events/LoadingRetainedMessagesEventArgs.cs
+10
View File
@@ -0,0 +1,10 @@
using System;
using System.Collections.Generic;
namespace MQTTnet.Server
{
public sealed class LoadingRetainedMessagesEventArgs : EventArgs
{
public List<MqttApplicationMessage> StoredRetainedMessages { get; set; } = new List<MqttApplicationMessage>();
}
}
Source/MQTTnet/Server/Events/PreparingMqttClientSessionEventArgs.cs → Source/MQTTnet/Server/Events/PreparingSessionEventArgs.cs
+18 -2
View File
@@ -4,9 +4,9 @@ using MQTTnet.Packets;
namespace MQTTnet.Server
{
public sealed class PreparingMqttClientSessionEventArgs : EventArgs
public sealed class PreparingSessionEventArgs : EventArgs
{
public string ClientId { get; internal set; }
public string Id { get; internal set; }
// TODO: Allow adding of packets to the queue etc.
@@ -25,5 +25,21 @@ namespace MQTTnet.Server
public List<MqttSubscription> Subscriptions { get; } = new List<MqttSubscription>();
public List<MqttPublishPacket> PublishPackets { get; } = new List<MqttPublishPacket>();
IDictionary<object, object> Items { get; set; }
/// <summary>
/// Gets the last will message.
/// In MQTT, you use the last will message feature to notify other clients about an ungracefully disconnected client.
/// </summary>
MqttApplicationMessage WillMessage { get; set; }
/// <summary>
/// Gets the will delay interval.
/// This is the time between the client disconnect and the time the will message will be sent.
/// </summary>
uint? WillDelayInterval { get; set; }
DateTime? SessionExpiryTimestamp { get; set; }
}
}
Source/MQTTnet/Server/Events/RetainedMessageChangedEventArgs.cs
+14
View File
@@ -0,0 +1,14 @@
using System;
using System.Collections.Generic;
namespace MQTTnet.Server
{
public sealed class RetainedMessageChangedEventArgs : EventArgs
{
public string ClientId { get; internal set; }
public MqttApplicationMessage ChangedRetainedMessage { get; internal set; }
public List<MqttApplicationMessage> StoredRetainedMessages { get; internal set; }
}
}
Source/MQTTnet/Server/Events/ValidatingMqttClientConnectionEventArgs.cs → Source/MQTTnet/Server/Events/ValidatingConnectionEventArgs.cs
+2 -2
View File
@@ -10,12 +10,12 @@ using MQTTnet.Protocol;
namespace MQTTnet.Server
{
public sealed class ValidatingMqttClientConnectionEventArgs : EventArgs
public sealed class ValidatingConnectionEventArgs : EventArgs
{
readonly MqttConnectPacket _connectPacket;
readonly IMqttChannelAdapter _clientAdapter;
public ValidatingMqttClientConnectionEventArgs(MqttConnectPacket connectPacket, IMqttChannelAdapter clientAdapter)
public ValidatingConnectionEventArgs(MqttConnectPacket connectPacket, IMqttChannelAdapter clientAdapter)
{
_connectPacket = connectPacket ?? throw new ArgumentNullException(nameof(connectPacket));
_clientAdapter = clientAdapter ?? throw new ArgumentNullException(nameof(clientAdapter));
Source/MQTTnet/Server/Internal/MqttApplicationMessageInterceptorInvoker.cs
+4 -4
View File
@@ -20,7 +20,7 @@ namespace MQTTnet.Server
public MqttApplicationMessage ApplicationMessage { get; private set; }
public MqttApplicationMessageResponse Response { get; private set; } = new MqttApplicationMessageResponse();
public PublishResponse Response { get; private set; } = new PublishResponse();
public bool ProcessPublish { get; private set; } = true;
@@ -29,13 +29,13 @@ namespace MQTTnet.Server
public async Task Invoke(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken)
{
// Reset to defaults.
Response = new MqttApplicationMessageResponse();
Response = new PublishResponse();
ProcessPublish = true;
CloseConnection = false;
ApplicationMessage = applicationMessage;
// Intercept.
var eventArgs = new InterceptingMqttClientPublishEventArgs
var eventArgs = new InterceptingPublishEventArgs
{
ClientId = _clientId,
ApplicationMessage = applicationMessage,
@@ -53,7 +53,7 @@ namespace MQTTnet.Server
eventArgs.ProcessPublish = false;
}
await _eventContainer.InterceptingClientPublishEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
await _eventContainer.InterceptingPublishEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
// Expose results.
ProcessPublish = eventArgs.ProcessPublish && eventArgs.ApplicationMessage != null;
Source/MQTTnet/Server/Internal/MqttClientConnection.cs → Source/MQTTnet/Server/Internal/MqttClient.cs
+6 -6
View File
@@ -14,7 +14,7 @@ using MQTTnet.Protocol;
namespace MQTTnet.Server
{
public sealed class MqttClientConnection : IDisposable
public sealed class MqttClient : IDisposable
{
readonly Dictionary<ushort, string> _topicAlias = new Dictionary<ushort, string>();
readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
@@ -32,10 +32,10 @@ namespace MQTTnet.Server
CancellationTokenSource _cancellationToken;
public MqttClientConnection(
public MqttClient(
MqttConnectPacket connectPacket,
IMqttChannelAdapter channelAdapter,
MqttClientSession session,
MqttSession session,
MqttServerOptions serverOptions,
MqttServerEventContainer eventContainer,
MqttClientSessionsManager sessionsManager,
@@ -54,7 +54,7 @@ namespace MQTTnet.Server
_applicationMessageInterceptorInvoker = new MqttApplicationMessageInterceptorInvoker(eventContainer, Id, Session.Items);
if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.WithSource(nameof(MqttClientConnection));
_logger = logger.WithSource(nameof(MqttClient));
}
public string Id => _connectPacket.ClientId;
@@ -63,13 +63,13 @@ namespace MQTTnet.Server
public IMqttChannelAdapter ChannelAdapter { get; }
public MqttClientConnectionStatistics Statistics { get; } = new MqttClientConnectionStatistics();
public MqttClientStatistics Statistics { get; } = new MqttClientStatistics();
public bool IsRunning { get; private set; }
public ushort KeepAlivePeriod => _connectPacket.KeepAlivePeriod;
public MqttClientSession Session { get; }
public MqttSession Session { get; }
public bool IsTakenOver { get; set; }
Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
+83 -124
View File
@@ -20,14 +20,14 @@ namespace MQTTnet.Server
readonly MqttPacketFactories _packetFactories = new MqttPacketFactories();
readonly AsyncLock _createConnectionSyncRoot = new AsyncLock();
readonly Dictionary<string, MqttClientConnection> _clientConnections = new Dictionary<string, MqttClientConnection>(4096);
readonly Dictionary<string, MqttClientSession> _clientSessions = new Dictionary<string, MqttClientSession>(4096);
readonly Dictionary<string, MqttClient> _clients = new Dictionary<string, MqttClient>(4096);
readonly Dictionary<string, MqttSession> _sessions = new Dictionary<string, MqttSession>(4096);
readonly IDictionary<object, object> _serverSessionItems = new ConcurrentDictionary<object, object>();
readonly MqttConnAckPacketFactory _connAckPacketFactory = new MqttConnAckPacketFactory();
readonly IMqttRetainedMessagesManager _retainedMessagesManager;
readonly MqttRetainedMessagesManager _retainedMessagesManager;
readonly MqttServerEventContainer _eventContainer;
readonly MqttServerOptions _options;
readonly MqttNetSourceLogger _logger;
@@ -35,7 +35,7 @@ namespace MQTTnet.Server
public MqttClientSessionsManager(
MqttServerOptions options,
IMqttRetainedMessagesManager retainedMessagesManager,
MqttRetainedMessagesManager retainedMessagesManager,
MqttServerEventContainer eventContainer,
IMqttNetLogger logger)
{
@@ -78,7 +78,7 @@ namespace MQTTnet.Server
public async Task HandleClientConnectionAsync(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
{
MqttClientConnection clientConnection = null;
MqttClient client = null;
try
{
@@ -103,11 +103,11 @@ namespace MQTTnet.Server
connAckPacket = _connAckPacketFactory.Create(connectionValidatorContext);
// Pass connAckPacket so that IsSessionPresent flag can be set if the client session already exists
clientConnection = await CreateClientConnection(connectPacket, connAckPacket, channelAdapter, connectionValidatorContext.SessionItems).ConfigureAwait(false);
client = await CreateClientConnection(connectPacket, connAckPacket, channelAdapter, connectionValidatorContext.SessionItems).ConfigureAwait(false);
await clientConnection.SendPacketAsync(connAckPacket, cancellationToken).ConfigureAwait(false);
await client.SendPacketAsync(connAckPacket, cancellationToken).ConfigureAwait(false);
await _eventContainer.ClientConnectedEvent.InvokeAsync(() => new MqttServerClientConnectedEventArgs
await _eventContainer.ClientConnectedEvent.InvokeAsync(() => new ClientConnectedEventArgs
{
ClientId = connectPacket.ClientId,
UserName = connectPacket.Username,
@@ -115,7 +115,7 @@ namespace MQTTnet.Server
Endpoint = channelAdapter.Endpoint
}).ConfigureAwait(false);
await clientConnection.RunAsync().ConfigureAwait(false);
await client.RunAsync().ConfigureAwait(false);
}
catch (OperationCanceledException)
{
@@ -126,33 +126,33 @@ namespace MQTTnet.Server
}
finally
{
if (clientConnection != null)
if (client != null)
{
if (clientConnection.Id != null)
if (client.Id != null)
{
// in case it is a takeover _clientConnections already contains the new connection
if (!clientConnection.IsTakenOver)
if (!client.IsTakenOver)
{
lock (_clientConnections)
lock (_clients)
{
_clientConnections.Remove(clientConnection.Id);
_clients.Remove(client.Id);
}
if (!_options.EnablePersistentSessions)
{
await DeleteSessionAsync(clientConnection.Id).ConfigureAwait(false);
await DeleteSessionAsync(client.Id).ConfigureAwait(false);
}
}
}
var endpoint = clientConnection.Endpoint;
var endpoint = client.Endpoint;
if (clientConnection.Id != null && !clientConnection.IsTakenOver)
if (client.Id != null && !client.IsTakenOver)
{
await _eventContainer.ClientDisconnectedEvent.InvokeAsync(() => new MqttServerClientDisconnectedEventArgs
await _eventContainer.ClientDisconnectedEvent.InvokeAsync(() => new ClientDisconnectedEventArgs
{
ClientId = clientConnection.Id,
DisconnectType = clientConnection.IsCleanDisconnect ? MqttClientDisconnectType.Clean : MqttClientDisconnectType.NotClean,
ClientId = client.Id,
DisconnectType = client.IsCleanDisconnect ? MqttClientDisconnectType.Clean : MqttClientDisconnectType.NotClean,
Endpoint = endpoint
}).ConfigureAwait(false);
}
@@ -167,11 +167,11 @@ namespace MQTTnet.Server
public async Task CloseAllConnectionsAsync()
{
List<MqttClientConnection> connections;
lock (_clientConnections)
List<MqttClient> connections;
lock (_clients)
{
connections = _clientConnections.Values.ToList();
_clientConnections.Clear();
connections = _clients.Values.ToList();
_clients.Clear();
}
foreach (var connection in connections)
@@ -180,11 +180,11 @@ namespace MQTTnet.Server
}
}
public List<MqttClientConnection> GetConnections()
public List<MqttClient> GetConnections()
{
lock (_clientConnections)
lock (_clients)
{
return _clientConnections.Values.ToList();
return _clients.Values.ToList();
}
}
@@ -192,9 +192,9 @@ namespace MQTTnet.Server
{
var result = new List<IMqttClientStatus>();
lock (_clientConnections)
lock (_clients)
{
foreach (var connection in _clientConnections.Values)
foreach (var connection in _clients.Values)
{
var clientStatus = new MqttClientStatus(connection)
{
@@ -212,9 +212,9 @@ namespace MQTTnet.Server
{
var result = new List<IMqttSessionStatus>();
lock (_clientSessions)
lock (_sessions)
{
foreach (var session in _clientSessions.Values)
foreach (var session in _sessions.Values)
{
var sessionStatus = new MqttSessionStatus(session);
result.Add(sessionStatus);
@@ -263,18 +263,18 @@ namespace MQTTnet.Server
public async Task DeleteSessionAsync(string clientId)
{
MqttClientConnection connection;
MqttClientSession session;
MqttClient connection;
MqttSession session;
lock (_clientConnections)
lock (_clients)
{
_clientConnections.TryGetValue(clientId, out connection);
_clients.TryGetValue(clientId, out connection);
}
lock (_clientSessions)
lock (_sessions)
{
_clientSessions.TryGetValue(clientId, out session);
_clientSessions.Remove(clientId);
_sessions.TryGetValue(clientId, out session);
_sessions.Remove(clientId);
}
try
@@ -331,67 +331,16 @@ namespace MQTTnet.Server
{
try
{
// MqttPendingApplicationMessage pendingApplicationMessage;
// try
// {
// pendingApplicationMessage = _messageQueue.Take();
// }
// catch (ArgumentNullException)
// {
// return;
// }
// // catch (ThreadAbortException)
// // {
// // return;
// // }
// catch (ObjectDisposedException)
// {
// return;
// }
//var clientConnection = pendingApplicationMessage.Sender;
//var senderClientId = clientConnection?.Id ?? _options.ClientId;
// // TODO: Only use PUBLISH packets internally!
// var applicationMessage = new MqttApplicationMessageFactory().Create(pendingApplicationMessage.PublishPacket);
//
// var interceptor = _options.ApplicationMessageInterceptor;
// if (interceptor != null)
// {
// var interceptorContext = await InterceptApplicationMessageAsync(interceptor, clientConnection, applicationMessage).ConfigureAwait(false);
// if (interceptorContext != null)
// {
// if (interceptorContext.CloseConnection)
// {
// if (clientConnection != null)
// {
// await clientConnection.StopAsync(MqttClientDisconnectReason.NormalDisconnection).ConfigureAwait(false);
// }
// }
//
// if (interceptorContext.ApplicationMessage == null || !interceptorContext.ProcessPublish)
// {
// return;
// }
//
// applicationMessage = interceptorContext.ApplicationMessage;
// }
// }
// await _eventContainer.ApplicationMessageReceivedEvent.InvokeAsync(
// () => new MqttApplicationMessageReceivedEventArgs(senderClientId, applicationMessage, null, null))
// .ConfigureAwait(false);
if (applicationMessage.Retain)
{
await _retainedMessagesManager.HandleMessageAsync(senderClientId, applicationMessage).ConfigureAwait(false);
await _retainedMessagesManager.UpdateMessage(senderClientId, applicationMessage).ConfigureAwait(false);
}
var deliveryCount = 0;
List<MqttClientSession> sessions;
lock (_clientSessions)
List<MqttSession> sessions;
lock (_sessions)
{
sessions = _clientSessions.Values.ToList();
sessions = _sessions.Values.ToList();
}
foreach (var clientSession in sessions)
@@ -434,7 +383,7 @@ namespace MQTTnet.Server
if (deliveryCount == 0)
{
await _eventContainer.ApplicationMessageNotConsumedEvent.InvokeAsync(() => new MqttApplicationMessageNotConsumedEventArgs
await _eventContainer.ApplicationMessageNotConsumedEvent.InvokeAsync(() => new ApplicationMessageNotConsumedEventArgs
{
ApplicationMessage = applicationMessage,
SenderClientId = senderClientId
@@ -447,14 +396,14 @@ namespace MQTTnet.Server
}
}
async Task<ValidatingMqttClientConnectionEventArgs> ValidateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter)
async Task<ValidatingConnectionEventArgs> ValidateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter)
{
var context = new ValidatingMqttClientConnectionEventArgs(connectPacket, channelAdapter)
var context = new ValidatingConnectionEventArgs(connectPacket, channelAdapter)
{
SessionItems = new ConcurrentDictionary<object, object>()
};
await _eventContainer.ValidatingClientConnectionEvent.InvokeAsync(context).ConfigureAwait(false);
await _eventContainer.ValidatingConnectionEvent.InvokeAsync(context).ConfigureAwait(false);
// Check the client ID and set a random one if supported.
if (string.IsNullOrEmpty(connectPacket.ClientId) && channelAdapter.PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.V500)
@@ -470,22 +419,21 @@ namespace MQTTnet.Server
return context;
}
async Task<MqttClientConnection> CreateClientConnection(
async Task<MqttClient> CreateClientConnection(
MqttConnectPacket connectPacket,
MqttConnAckPacket connAckPacket,
IMqttChannelAdapter channelAdapter,
IDictionary<object, object> sessionItems)
{
MqttClientConnection connection;
MqttClient connection;
using (await _createConnectionSyncRoot.WaitAsync(CancellationToken.None).ConfigureAwait(false))
{
MqttClientSession session;
lock (_clientSessions)
MqttSession session;
lock (_sessions)
{
if (!_clientSessions.TryGetValue(connectPacket.ClientId, out session))
if (!_sessions.TryGetValue(connectPacket.ClientId, out session))
{
_logger.Verbose("Created a new session for client '{0}'.", connectPacket.ClientId);
session = CreateSession(connectPacket.ClientId, sessionItems);
}
else
@@ -503,29 +451,38 @@ namespace MQTTnet.Server
}
}
_clientSessions[connectPacket.ClientId] = session;
_sessions[connectPacket.ClientId] = session;
}
MqttClientConnection existingConnection;
lock (_clientConnections)
if (!connAckPacket.IsSessionPresent)
{
_clientConnections.TryGetValue(connectPacket.ClientId, out existingConnection);
var preparingSessionEventArgs = new PreparingSessionEventArgs();
await _eventContainer.PreparingSessionEvent.InvokeAsync(preparingSessionEventArgs).ConfigureAwait(false);
// TODO: Import subscriptions etc.
//session.SubscriptionsManager.Subscribe()
}
MqttClient existing;
lock (_clients)
{
_clients.TryGetValue(connectPacket.ClientId, out existing);
connection = CreateConnection(connectPacket, channelAdapter, session);
_clientConnections[connectPacket.ClientId] = connection;
_clients[connectPacket.ClientId] = connection;
}
if (existingConnection != null)
if (existing != null)
{
existingConnection.IsTakenOver = true;
await existingConnection.StopAsync(MqttClientDisconnectReason.SessionTakenOver).ConfigureAwait(false);
existing.IsTakenOver = true;
await existing.StopAsync(MqttClientDisconnectReason.SessionTakenOver).ConfigureAwait(false);
await _eventContainer.ClientDisconnectedEvent.InvokeAsync(() => new MqttServerClientDisconnectedEventArgs
await _eventContainer.ClientDisconnectedEvent.InvokeAsync(() => new ClientDisconnectedEventArgs
{
ClientId = existingConnection.Id,
ClientId = existing.Id,
DisconnectType = MqttClientDisconnectType.Takeover,
Endpoint = existingConnection.Endpoint
Endpoint = existing.Endpoint
}).ConfigureAwait(false);
}
}
@@ -566,11 +523,11 @@ namespace MQTTnet.Server
// return interceptorContext;
// }
MqttClientSession GetClientSession(string clientId)
MqttSession GetClientSession(string clientId)
{
lock (_clientSessions)
lock (_sessions)
{
if (!_clientSessions.TryGetValue(clientId, out var session))
if (!_sessions.TryGetValue(clientId, out var session))
{
throw new InvalidOperationException($"Client session '{clientId}' is unknown.");
}
@@ -579,9 +536,9 @@ namespace MQTTnet.Server
}
}
MqttClientConnection CreateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, MqttClientSession session)
MqttClient CreateConnection(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, MqttSession session)
{
return new MqttClientConnection(
return new MqttClient(
connectPacket,
channelAdapter,
session,
@@ -591,9 +548,11 @@ namespace MQTTnet.Server
_rootLogger);
}
MqttClientSession CreateSession(string clientId, IDictionary<object, object> sessionItems)
MqttSession CreateSession(string clientId, IDictionary<object, object> sessionItems)
{
return new MqttClientSession(
_logger.Verbose("Created a new session for client '{0}'.", clientId);
return new MqttSession(
clientId,
sessionItems,
_options,
@@ -606,12 +565,12 @@ namespace MQTTnet.Server
{
_createConnectionSyncRoot?.Dispose();
foreach (var clientConnection in _clientConnections.Values)
foreach (var clientConnection in _clients.Values)
{
clientConnection.Dispose();
}
foreach (var session in _clientSessions.Values)
foreach (var session in _sessions.Values)
{
session.Dispose();
}
Source/MQTTnet/Server/Internal/MqttClientConnectionStatistics.cs → Source/MQTTnet/Server/Internal/MqttClientStatistics.cs
+2 -2
View File
@@ -4,7 +4,7 @@ using MQTTnet.Packets;
namespace MQTTnet.Server
{
public sealed class MqttClientConnectionStatistics
public sealed class MqttClientStatistics
{
// Start with 1 because the CONNACK packet is not counted here.
long _receivedPacketsCount = 1;
@@ -15,7 +15,7 @@ namespace MQTTnet.Server
long _receivedApplicationMessagesCount;
long _sentApplicationMessagesCount;
public MqttClientConnectionStatistics()
public MqttClientStatistics()
{
ConnectedTimestamp = DateTime.UtcNow;
Source/MQTTnet/Server/Internal/MqttClientSubscriptionsManager.cs
+29 -30
View File
@@ -16,29 +16,29 @@ namespace MQTTnet.Server
readonly ReaderWriterLockSlim _subscriptionsLock = new ReaderWriterLockSlim();
readonly Dictionary<string, MqttSubscription> _subscriptions = new Dictionary<string, MqttSubscription>(4096);
readonly MqttClientSession _clientSession;
readonly MqttSession _session;
readonly MqttServerOptions _options;
readonly MqttServerEventContainer _eventContainer;
readonly IMqttRetainedMessagesManager _retainedMessagesManager;
readonly MqttRetainedMessagesManager _retainedMessagesManager;
public MqttClientSubscriptionsManager(
MqttClientSession clientSession,
MqttSession session,
MqttServerOptions serverOptions,
MqttServerEventContainer eventContainer,
IMqttRetainedMessagesManager retainedMessagesManager)
MqttRetainedMessagesManager retainedMessagesManager)
{
_clientSession = clientSession ?? throw new ArgumentNullException(nameof(clientSession));
_session = session ?? throw new ArgumentNullException(nameof(session));
_options = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
_eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));
_retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
}
public async Task<MqttSubscribeResult> Subscribe(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
public async Task<SubscribeResult> Subscribe(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
{
if (subscribePacket == null) throw new ArgumentNullException(nameof(subscribePacket));
var retainedApplicationMessages = await _retainedMessagesManager.GetMessagesAsync().ConfigureAwait(false);
var result = new MqttSubscribeResult();
var retainedApplicationMessages = await _retainedMessagesManager.GetMessages().ConfigureAwait(false);
var result = new SubscribeResult();
// The topic filters are order by its QoS so that the higher QoS will win over a
// lower one.
@@ -79,9 +79,9 @@ namespace MQTTnet.Server
subscribePacket.Properties?.SubscriptionIdentifier ?? 0,
interceptorContext.Response.ReasonCode);
await _eventContainer.ClientSubscribedTopicEvent.InvokeAsync(() => new MqttServerClientSubscribedTopicEventArgs
await _eventContainer.ClientSubscribedTopicEvent.InvokeAsync(() => new ClientSubscribedTopicEventArgs
{
ClientId = _clientSession.Id,
ClientId = _session.Id,
TopicFilter = finalTopicFilter
}).ConfigureAwait(false);
@@ -135,9 +135,9 @@ namespace MQTTnet.Server
foreach (var topicFilter in unsubscribePacket.TopicFilters)
{
await _eventContainer.ClientUnsubscribedTopicEvent.InvokeAsync(() => new MqttServerClientUnsubscribedTopicEventArgs
await _eventContainer.ClientUnsubscribedTopicEvent.InvokeAsync(() => new ClientUnsubscribedTopicEventArgs
{
ClientId = _clientSession.Id,
ClientId = _session.Id,
TopicFilter = topicFilter
}).ConfigureAwait(false);
}
@@ -159,7 +159,7 @@ namespace MQTTnet.Server
_subscriptionsLock.ExitReadLock();
}
var senderIsReceiver = string.Equals(senderClientId, _clientSession.Id);
var senderIsReceiver = string.Equals(senderClientId, _session.Id);
var qosLevels = new HashSet<MqttQualityOfServiceLevel>();
var subscriptionIdentifiers = new HashSet<uint>();
@@ -262,7 +262,7 @@ namespace MQTTnet.Server
}
static void FilterRetainedApplicationMessages(IList<MqttApplicationMessage> retainedApplicationMessages, CreateSubscriptionResult createSubscriptionResult,
MqttSubscribeResult subscribeResult)
SubscribeResult subscribeResult)
{
for (var i = retainedApplicationMessages.Count - 1; i >= 0; i--)
{
@@ -294,14 +294,13 @@ namespace MQTTnet.Server
var queuedApplicationMessage = new MqttQueuedApplicationMessage
{
ApplicationMessage = retainedApplicationMessage,
IsRetainedMessage = true,
SubscriptionQualityOfServiceLevel = createSubscriptionResult.Subscription.GrantedQualityOfServiceLevel
};
if (createSubscriptionResult.Subscription.Identifier > 0)
{
queuedApplicationMessage.SubscriptionIdentifiers = new List<uint> {createSubscriptionResult.Subscription.Identifier};
}
// if (createSubscriptionResult.Subscription.Identifier > 0)
// {
// queuedApplicationMessage.SubscriptionIdentifiers = new List<uint> {createSubscriptionResult.Subscription.Identifier};
// }
subscribeResult.RetainedApplicationMessages.Add(queuedApplicationMessage);
@@ -309,14 +308,14 @@ namespace MQTTnet.Server
}
}
async Task<InterceptingMqttClientSubscriptionEventArgs> InterceptSubscribe(MqttTopicFilter topicFilter, CancellationToken cancellationToken)
async Task<InterceptingSubscriptionEventArgs> InterceptSubscribe(MqttTopicFilter topicFilter, CancellationToken cancellationToken)
{
var subscriptionReceivedEventArgs = new InterceptingMqttClientSubscriptionEventArgs
var subscriptionReceivedEventArgs = new InterceptingSubscriptionEventArgs
{
ClientId = _clientSession.Id,
ClientId = _session.Id,
TopicFilter = topicFilter,
SessionItems = _clientSession.Items,
Session = new MqttSessionStatus(_clientSession),
SessionItems = _session.Items,
Session = new MqttSessionStatus(_session),
CancellationToken = cancellationToken
};
@@ -339,19 +338,19 @@ namespace MQTTnet.Server
}
else
{
await _eventContainer.InterceptingClientSubscriptionEvent.InvokeAsync(subscriptionReceivedEventArgs).ConfigureAwait(false);
await _eventContainer.InterceptingSubscriptionEvent.InvokeAsync(subscriptionReceivedEventArgs).ConfigureAwait(false);
}
return subscriptionReceivedEventArgs;
}
async Task<InterceptingMqttClientUnsubscriptionEventArgs> InterceptUnsubscribe(string topicFilter, MqttSubscription mqttSubscription, CancellationToken cancellationToken)
async Task<InterceptingUnsubscriptionEventArgs> InterceptUnsubscribe(string topicFilter, MqttSubscription mqttSubscription, CancellationToken cancellationToken)
{
var clientUnsubscribingTopicEventArgs = new InterceptingMqttClientUnsubscriptionEventArgs
var clientUnsubscribingTopicEventArgs = new InterceptingUnsubscriptionEventArgs
{
ClientId = _clientSession.Id,
ClientId = _session.Id,
Topic = topicFilter,
SessionItems = _clientSession.Items,
SessionItems = _session.Items,
CancellationToken = cancellationToken
};
@@ -364,7 +363,7 @@ namespace MQTTnet.Server
clientUnsubscribingTopicEventArgs.Response.ReasonCode = MqttUnsubscribeReasonCode.Success;
}
await _eventContainer.InterceptingClientUnsubscriptionEvent.InvokeAsync(clientUnsubscribingTopicEventArgs).ConfigureAwait(false);
await _eventContainer.InterceptingUnsubscriptionEvent.InvokeAsync(clientUnsubscribingTopicEventArgs).ConfigureAwait(false);
return clientUnsubscribingTopicEventArgs;
}
Source/MQTTnet/Server/Internal/MqttRetainedMessagesManager.cs
+28 -36
View File
@@ -9,45 +9,36 @@ using MQTTnet.Internal;
namespace MQTTnet.Server
{
public sealed class MqttRetainedMessagesManager : IMqttRetainedMessagesManager
public sealed class MqttRetainedMessagesManager
{
readonly AsyncLock _storageAccessLock = new AsyncLock();
readonly Dictionary<string, MqttApplicationMessage> _messages = new Dictionary<string, MqttApplicationMessage>(4096);
readonly AsyncLock _storageAccessLock = new AsyncLock();
MqttNetSourceLogger _logger;
MqttServerOptions _options;
// TODO: Get rid of the logger here!
public Task Start(MqttServerOptions options, IMqttNetLogger logger)
readonly MqttServerEventContainer _eventContainer;
readonly MqttNetSourceLogger _logger;
public MqttRetainedMessagesManager(MqttServerEventContainer eventContainer, IMqttNetLogger logger)
{
_eventContainer = eventContainer ?? throw new ArgumentNullException(nameof(eventContainer));
if (logger == null) throw new ArgumentNullException(nameof(logger));
_logger = logger.WithSource(nameof(MqttRetainedMessagesManager));
_options = options ?? throw new ArgumentNullException(nameof(options));
return PlatformAbstractionLayer.CompletedTask;
}
public async Task LoadMessagesAsync()
public async Task LoadMessages()
{
if (_options.Storage == null)
{
return;
}
try
{
var retainedMessages = await _options.Storage.LoadRetainedMessagesAsync().ConfigureAwait(false);
var eventArgs = new LoadingRetainedMessagesEventArgs();
await _eventContainer.LoadingRetainedMessagesEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
lock (_messages)
{
_messages.Clear();
if (retainedMessages != null)
foreach (var retainedMessage in eventArgs.StoredRetainedMessages ?? new List<MqttApplicationMessage>())
{
foreach (var retainedMessage in retainedMessages)
{
_messages[retainedMessage.Topic] = retainedMessage;
}
_messages[retainedMessage.Topic] = retainedMessage;
}
}
}
@@ -57,7 +48,7 @@ namespace MQTTnet.Server
}
}
public async Task HandleMessageAsync(string clientId, MqttApplicationMessage applicationMessage)
public async Task UpdateMessage(string clientId, MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
@@ -102,12 +93,16 @@ namespace MQTTnet.Server
if (saveIsRequired)
{
if (_options.Storage != null)
using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
{
using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
var eventArgs = new RetainedMessageChangedEventArgs
{
await _options.Storage.SaveRetainedMessagesAsync(messagesForSave).ConfigureAwait(false);
}
ClientId = clientId,
ChangedRetainedMessage = applicationMessage,
StoredRetainedMessages = messagesForSave
};
await _eventContainer.RetainedMessageChangedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
}
}
}
@@ -117,7 +112,7 @@ namespace MQTTnet.Server
}
}
public Task<IList<MqttApplicationMessage>> GetMessagesAsync()
public Task<IList<MqttApplicationMessage>> GetMessages()
{
lock (_messages)
{
@@ -125,19 +120,16 @@ namespace MQTTnet.Server
}
}
public async Task ClearMessagesAsync()
public async Task ClearMessages()
{
lock (_messages)
{
_messages.Clear();
}
if (_options.Storage != null)
using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
{
using (await _storageAccessLock.WaitAsync(CancellationToken.None).ConfigureAwait(false))
{
await _options.Storage.SaveRetainedMessagesAsync(new List<MqttApplicationMessage>()).ConfigureAwait(false);
}
await _eventContainer.RetainedMessagesClearedEvent.InvokeAsync(EventArgs.Empty).ConfigureAwait(false);
}
}
}
Source/MQTTnet/Server/Internal/MqttServerEventContainer.cs
+14 -12
View File
@@ -5,29 +5,31 @@ namespace MQTTnet.Server
{
public sealed class MqttServerEventContainer
{
public AsyncEvent<InterceptingMqttClientSubscriptionEventArgs> InterceptingClientSubscriptionEvent { get; } = new AsyncEvent<InterceptingMqttClientSubscriptionEventArgs>();
public AsyncEvent<InterceptingSubscriptionEventArgs> InterceptingSubscriptionEvent { get; } = new AsyncEvent<InterceptingSubscriptionEventArgs>();
public AsyncEvent<InterceptingMqttClientUnsubscriptionEventArgs> InterceptingClientUnsubscriptionEvent { get; } = new AsyncEvent<InterceptingMqttClientUnsubscriptionEventArgs>();
public AsyncEvent<InterceptingUnsubscriptionEventArgs> InterceptingUnsubscriptionEvent { get; } = new AsyncEvent<InterceptingUnsubscriptionEventArgs>();
public AsyncEvent<InterceptingMqttClientPublishEventArgs> InterceptingClientPublishEvent { get; } = new AsyncEvent<InterceptingMqttClientPublishEventArgs>();
public AsyncEvent<InterceptingPublishEventArgs> InterceptingPublishEvent { get; } = new AsyncEvent<InterceptingPublishEventArgs>();
public AsyncEvent<ValidatingMqttClientConnectionEventArgs> ValidatingClientConnectionEvent { get; } = new AsyncEvent<ValidatingMqttClientConnectionEventArgs>();
public AsyncEvent<ValidatingConnectionEventArgs> ValidatingConnectionEvent { get; } = new AsyncEvent<ValidatingConnectionEventArgs>();
public AsyncEvent<MqttServerClientConnectedEventArgs> ClientConnectedEvent { get; } = new AsyncEvent<MqttServerClientConnectedEventArgs>();
public AsyncEvent<ClientConnectedEventArgs> ClientConnectedEvent { get; } = new AsyncEvent<ClientConnectedEventArgs>();
public AsyncEvent<MqttServerClientDisconnectedEventArgs> ClientDisconnectedEvent { get; } = new AsyncEvent<MqttServerClientDisconnectedEventArgs>();
public AsyncEvent<ClientDisconnectedEventArgs> ClientDisconnectedEvent { get; } = new AsyncEvent<ClientDisconnectedEventArgs>();
public AsyncEvent<MqttServerClientSubscribedTopicEventArgs> ClientSubscribedTopicEvent { get; } = new AsyncEvent<MqttServerClientSubscribedTopicEventArgs>();
public AsyncEvent<ClientSubscribedTopicEventArgs> ClientSubscribedTopicEvent { get; } = new AsyncEvent<ClientSubscribedTopicEventArgs>();
public AsyncEvent<MqttServerClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopicEvent { get; } = new AsyncEvent<MqttServerClientUnsubscribedTopicEventArgs>();
public AsyncEvent<ClientUnsubscribedTopicEventArgs> ClientUnsubscribedTopicEvent { get; } = new AsyncEvent<ClientUnsubscribedTopicEventArgs>();
public AsyncEvent<PreparingMqttClientSessionEventArgs> PreparingClientSessionEvent { get; } = new AsyncEvent<PreparingMqttClientSessionEventArgs>();
public AsyncEvent<PreparingSessionEventArgs> PreparingSessionEvent { get; } = new AsyncEvent<PreparingSessionEventArgs>();
public AsyncEvent<MqttApplicationMessageNotConsumedEventArgs> ApplicationMessageNotConsumedEvent { get; } = new AsyncEvent<MqttApplicationMessageNotConsumedEventArgs>();
public AsyncEvent<ApplicationMessageNotConsumedEventArgs> ApplicationMessageNotConsumedEvent { get; } = new AsyncEvent<ApplicationMessageNotConsumedEventArgs>();
public AsyncEvent<EventArgs> RetainedApplicationMessageChangedEvent { get; } = new AsyncEvent<EventArgs>();
public AsyncEvent<RetainedMessageChangedEventArgs> RetainedMessageChangedEvent { get; } = new AsyncEvent<RetainedMessageChangedEventArgs>();
public AsyncEvent<EventArgs> RetainedApplicationMessageClearedEvent { get; } = new AsyncEvent<EventArgs>();
public AsyncEvent<LoadingRetainedMessagesEventArgs> LoadingRetainedMessagesEvent { get; } = new AsyncEvent<LoadingRetainedMessagesEventArgs>();
public AsyncEvent<EventArgs> RetainedMessagesClearedEvent { get; } = new AsyncEvent<EventArgs>();
public AsyncEvent<InterceptingPacketEventArgs> InterceptingInboundPacketEvent { get; } = new AsyncEvent<InterceptingPacketEventArgs>();
Source/MQTTnet/Server/Internal/MqttServerKeepAliveMonitor.cs
+1 -1
View File
@@ -66,7 +66,7 @@ namespace MQTTnet.Server
}
}
void TryMaintainConnection(MqttClientConnection connection, DateTime now)
void TryMaintainConnection(MqttClient connection, DateTime now)
{
try
{
Source/MQTTnet/Server/Internal/MqttClientSession.cs → Source/MQTTnet/Server/Internal/MqttSession.cs
+3 -3
View File
@@ -10,7 +10,7 @@ using MQTTnet.Protocol;
namespace MQTTnet.Server
{
public sealed class MqttClientSession : IDisposable
public sealed class MqttSession : IDisposable
{
readonly MqttPacketBus _packetBus = new MqttPacketBus();
@@ -19,11 +19,11 @@ namespace MQTTnet.Server
readonly MqttServerOptions _serverOptions;
readonly MqttClientSessionsManager _clientSessionsManager;
public MqttClientSession(string clientId,
public MqttSession(string clientId,
IDictionary<object, object> items,
MqttServerOptions serverOptions,
MqttServerEventContainer eventContainer,
IMqttRetainedMessagesManager retainedMessagesManager,
MqttRetainedMessagesManager retainedMessagesManager,
MqttClientSessionsManager clientSessionsManager)
{
_serverOptions = serverOptions ?? throw new ArgumentNullException(nameof(serverOptions));
Source/MQTTnet/Server/Internal/MqttSubscribeResult.cs → Source/MQTTnet/Server/Internal/SubscribeResult.cs
+1 -1
View File
@@ -3,7 +3,7 @@ using MQTTnet.Protocol;
namespace MQTTnet.Server
{
public sealed class MqttSubscribeResult
public sealed class SubscribeResult
{
public List<MqttSubscribeReturnCode> ReturnCodes { get; } = new List<MqttSubscribeReturnCode>(128);
Source/MQTTnet/Server/MqttQueuedApplicationMessage.cs
-8
View File
@@ -6,14 +6,6 @@ namespace MQTTnet.Server
public sealed class MqttQueuedApplicationMessage
{
public MqttApplicationMessage ApplicationMessage { get; set; }
public string SenderClientId { get; set; }
public bool IsRetainedMessage { get; set; }
public List<uint> SubscriptionIdentifiers { get; set; }
public bool IsDuplicate { get; set; }
/// <summary>
/// Gets or sets the subscription quality of service level.
Source/MQTTnet/Server/MqttServer.cs
+65 -38
View File
@@ -1,5 +1,4 @@
using MQTTnet.Adapter;
using MQTTnet.Exceptions;
using MQTTnet.Protocol;
using System;
using System.Collections.Generic;
@@ -24,7 +23,7 @@ namespace MQTTnet.Server
readonly MqttNetSourceLogger _logger;
MqttClientSessionsManager _clientSessionsManager;
IMqttRetainedMessagesManager _retainedMessagesManager;
MqttRetainedMessagesManager _retainedMessagesManager;
MqttServerKeepAliveMonitor _keepAliveMonitor;
CancellationTokenSource _cancellationTokenSource;
@@ -52,19 +51,19 @@ namespace MQTTnet.Server
remove => _eventContainer.StoppedEvent.RemoveHandler(value);
}
public event Func<ValidatingMqttClientConnectionEventArgs, Task> ValidatingClientConnectionAsync
public event Func<ValidatingConnectionEventArgs, Task> ValidatingConnectionAsync
{
add => _eventContainer.ValidatingClientConnectionEvent.AddHandler(value);
remove => _eventContainer.ValidatingClientConnectionEvent.RemoveHandler(value);
add => _eventContainer.ValidatingConnectionEvent.AddHandler(value);
remove => _eventContainer.ValidatingConnectionEvent.RemoveHandler(value);
}
public event Func<MqttServerClientConnectedEventArgs, Task> ClientConnectedAsync
public event Func<ClientConnectedEventArgs, Task> ClientConnectedAsync
{
add => _eventContainer.ClientConnectedEvent.AddHandler(value);
remove => _eventContainer.ClientConnectedEvent.RemoveHandler(value);
}
public event Func<MqttServerClientDisconnectedEventArgs, Task> ClientDisconnectedAsync
public event Func<ClientDisconnectedEventArgs, Task> ClientDisconnectedAsync
{
add => _eventContainer.ClientDisconnectedEvent.AddHandler(value);
remove => _eventContainer.ClientDisconnectedEvent.RemoveHandler(value);
@@ -82,63 +81,69 @@ namespace MQTTnet.Server
remove => _eventContainer.InterceptingOutboundPacketEvent.RemoveHandler(value);
}
public event Func<InterceptingMqttClientSubscriptionEventArgs, Task> InterceptingClientSubscriptionAsync
public event Func<InterceptingSubscriptionEventArgs, Task> InterceptingSubscriptionAsync
{
add => _eventContainer.InterceptingClientSubscriptionEvent.AddHandler(value);
remove => _eventContainer.InterceptingClientSubscriptionEvent.RemoveHandler(value);
add => _eventContainer.InterceptingSubscriptionEvent.AddHandler(value);
remove => _eventContainer.InterceptingSubscriptionEvent.RemoveHandler(value);
}
public event Func<InterceptingMqttClientUnsubscriptionEventArgs, Task> InterceptingClientUnsubscriptionAsync
public event Func<InterceptingUnsubscriptionEventArgs, Task> InterceptingUnsubscriptionAsync
{
add => _eventContainer.InterceptingClientUnsubscriptionEvent.AddHandler(value);
remove => _eventContainer.InterceptingClientUnsubscriptionEvent.RemoveHandler(value);
add => _eventContainer.InterceptingUnsubscriptionEvent.AddHandler(value);
remove => _eventContainer.InterceptingUnsubscriptionEvent.RemoveHandler(value);
}
public event Func<InterceptingMqttClientPublishEventArgs, Task> InterceptingClientPublishAsync
public event Func<InterceptingPublishEventArgs, Task> InterceptingPublishAsync
{
add => _eventContainer.InterceptingClientPublishEvent.AddHandler(value);
remove => _eventContainer.InterceptingClientPublishEvent.RemoveHandler(value);
add => _eventContainer.InterceptingPublishEvent.AddHandler(value);
remove => _eventContainer.InterceptingPublishEvent.RemoveHandler(value);
}
public event Func<MqttServerClientSubscribedTopicEventArgs, Task> ClientSubscribedTopicAsync
public event Func<ClientSubscribedTopicEventArgs, Task> ClientSubscribedTopicAsync
{
add => _eventContainer.ClientSubscribedTopicEvent.AddHandler(value);
remove => _eventContainer.ClientSubscribedTopicEvent.RemoveHandler(value);
}
public event Func<MqttServerClientUnsubscribedTopicEventArgs, Task> ClientUnsubscribedTopicAsync
public event Func<ClientUnsubscribedTopicEventArgs, Task> ClientUnsubscribedTopicAsync
{
add => _eventContainer.ClientUnsubscribedTopicEvent.AddHandler(value);
remove => _eventContainer.ClientUnsubscribedTopicEvent.RemoveHandler(value);
}
public event Func<PreparingMqttClientSessionEventArgs, Task> PreparingClientSessionAsync
public event Func<PreparingSessionEventArgs, Task> PreparingSessionAsync
{
add => _eventContainer.PreparingClientSessionEvent.AddHandler(value);
remove => _eventContainer.PreparingClientSessionEvent.RemoveHandler(value);
add => _eventContainer.PreparingSessionEvent.AddHandler(value);
remove => _eventContainer.PreparingSessionEvent.RemoveHandler(value);
}
public event Func<MqttApplicationMessageNotConsumedEventArgs, Task> ApplicationMessageNotConsumedAsync
public event Func<ApplicationMessageNotConsumedEventArgs, Task> ApplicationMessageNotConsumedAsync
{
add => _eventContainer.ApplicationMessageNotConsumedEvent.AddHandler(value);
remove => _eventContainer.ApplicationMessageNotConsumedEvent.RemoveHandler(value);
}
public event Func<EventArgs, Task> RetainedApplicationMessageChangedAsync
public event Func<RetainedMessageChangedEventArgs, Task> RetainedMessageChangedAsync
{
add => _eventContainer.RetainedApplicationMessageChangedEvent.AddHandler(value);
remove => _eventContainer.RetainedApplicationMessageChangedEvent.RemoveHandler(value);
add => _eventContainer.RetainedMessageChangedEvent.AddHandler(value);
remove => _eventContainer.RetainedMessageChangedEvent.RemoveHandler(value);
}
public event Func<EventArgs, Task> RetainedApplicationMessagesClearedAsync
public event Func<LoadingRetainedMessagesEventArgs, Task> LoadingRetainedMessageAsync
{
add => _eventContainer.RetainedApplicationMessageClearedEvent.AddHandler(value);
remove => _eventContainer.RetainedApplicationMessageClearedEvent.RemoveHandler(value);
add => _eventContainer.LoadingRetainedMessagesEvent.AddHandler(value);
remove => _eventContainer.LoadingRetainedMessagesEvent.RemoveHandler(value);
}
public event Func<EventArgs, Task> RetainedMessagesClearedAsync
{
add => _eventContainer.RetainedMessagesClearedEvent.AddHandler(value);
remove => _eventContainer.RetainedMessagesClearedEvent.RemoveHandler(value);
}
public bool IsStarted => _cancellationTokenSource != null;
public Task<IList<IMqttClientStatus>> GetClientStatusAsync()
public Task<IList<IMqttClientStatus>> GetClientsAsync()
{
ThrowIfDisposed();
ThrowIfNotStarted();
@@ -146,7 +151,7 @@ namespace MQTTnet.Server
return _clientSessionsManager.GetClientStatusAsync();
}
public Task<IList<IMqttSessionStatus>> GetSessionStatusAsync()
public Task<IList<IMqttSessionStatus>> GetSessionsAsync()
{
ThrowIfDisposed();
ThrowIfNotStarted();
@@ -154,20 +159,44 @@ namespace MQTTnet.Server
return _clientSessionsManager.GetSessionStatusAsync();
}
public Task<IList<MqttApplicationMessage>> GetRetainedApplicationMessagesAsync()
/// <summary>
/// Gets all currently cached retained messages.
/// </summary>
/// <returns>A list of retained messages.</returns>
public Task<IList<MqttApplicationMessage>> GetRetainedMessagesAsync()
{
ThrowIfDisposed();
ThrowIfNotStarted();
return _retainedMessagesManager.GetMessagesAsync();
return _retainedMessagesManager.GetMessages();
}
public Task ClearRetainedApplicationMessagesAsync()
/// <summary>
/// Deletes all retained messages.
/// </summary>
/// <returns>A task.</returns>
public Task DeleteRetainedMessagesAsync()
{
ThrowIfDisposed();
ThrowIfNotStarted();
return _retainedMessagesManager?.ClearMessagesAsync() ?? PlatformAbstractionLayer.CompletedTask;
return _retainedMessagesManager?.ClearMessages() ?? PlatformAbstractionLayer.CompletedTask;
}
/// <summary>
/// Updates the specified retained message.
/// Setting the payload to null or empty will delete the retained message.
/// </summary>
/// <param name="retainedMessage">The retained message.</param>
/// <returns>A task.</returns>
public Task UpdateRetainedMessageAsync(MqttApplicationMessage retainedMessage)
{
if (retainedMessage == null) throw new ArgumentNullException(nameof(retainedMessage));
ThrowIfDisposed();
ThrowIfNotStarted();
return _retainedMessagesManager?.UpdateMessage(null, retainedMessage);
}
public Task SubscribeAsync(string clientId, ICollection<MqttTopicFilter> topicFilters)
@@ -220,10 +249,8 @@ namespace MQTTnet.Server
_cancellationTokenSource = new CancellationTokenSource();
var cancellationToken = _cancellationTokenSource.Token;
_retainedMessagesManager = _options.RetainedMessagesManager ?? throw new MqttConfigurationException("options.RetainedMessagesManager should not be null.");
await _retainedMessagesManager.Start(_options, _rootLogger).ConfigureAwait(false);
await _retainedMessagesManager.LoadMessagesAsync().ConfigureAwait(false);
_retainedMessagesManager = new MqttRetainedMessagesManager(_eventContainer, _rootLogger);
await _retainedMessagesManager.LoadMessages().ConfigureAwait(false);
_clientSessionsManager = new MqttClientSessionsManager(_options, _retainedMessagesManager, _eventContainer, _rootLogger);
Source/MQTTnet/Server/MqttServerExtensions.cs
+1 -107
View File
@@ -16,16 +16,7 @@ namespace MQTTnet.Server
return server.SubscribeAsync(clientId, topicFilters);
}
public static Task SubscribeAsync(this MqttServer server, string clientId, string topic, MqttQualityOfServiceLevel qualityOfServiceLevel)
{
if (server == null) throw new ArgumentNullException(nameof(server));
if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topic == null) throw new ArgumentNullException(nameof(topic));
return server.SubscribeAsync(clientId, new MqttTopicFilterBuilder().WithTopic(topic).WithQualityOfServiceLevel(qualityOfServiceLevel).Build());
}
public static Task SubscribeAsync(this MqttServer server, string clientId, string topic)
{
if (server == null) throw new ArgumentNullException(nameof(server));
@@ -34,102 +25,5 @@ namespace MQTTnet.Server
return server.SubscribeAsync(clientId, new MqttTopicFilterBuilder().WithTopic(topic).Build());
}
public static Task UnsubscribeAsync(this MqttServer server, string clientId, params string[] topicFilters)
{
if (server == null) throw new ArgumentNullException(nameof(server));
if (clientId == null) throw new ArgumentNullException(nameof(clientId));
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
return server.UnsubscribeAsync(clientId, topicFilters);
}
// public static async Task PublishAsync(this IMqttServer server, IEnumerable<MqttApplicationMessage> applicationMessages)
// {
// if (server == null) throw new ArgumentNullException(nameof(server));
// if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));
//
// foreach (var applicationMessage in applicationMessages)
// {
// await server.PublishAsync(applicationMessage).ConfigureAwait(false);
// }
// }
// public static Task<MqttClientPublishResult> PublishAsync(this IMqttServer server, MqttApplicationMessage applicationMessage)
// {
// if (server == null) throw new ArgumentNullException(nameof(server));
// if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
//
// return server.PublishAsync(applicationMessage, CancellationToken.None);
// }
//
// public static async Task PublishAsync(this IMqttServer server, params MqttApplicationMessage[] applicationMessages)
// {
// if (server == null) throw new ArgumentNullException(nameof(server));
// if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));
//
// foreach (var applicationMessage in applicationMessages)
// {
// await server.PublishAsync(applicationMessage, CancellationToken.None).ConfigureAwait(false);
// }
// }
// public static Task<MqttClientPublishResult> PublishAsync(this IMqttServer server, string topic)
// {
// if (server == null) throw new ArgumentNullException(nameof(server));
// if (topic == null) throw new ArgumentNullException(nameof(topic));
//
// return server.PublishAsync(builder => builder
// .WithTopic(topic));
// }
// public static Task<MqttClientPublishResult> PublishAsync(this IMqttServer server, string topic, string payload)
// {
// if (server == null) throw new ArgumentNullException(nameof(server));
// if (topic == null) throw new ArgumentNullException(nameof(topic));
//
// return server.PublishAsync(builder => builder
// .WithTopic(topic)
// .WithPayload(payload));
// }
// public static Task<MqttClientPublishResult> PublishAsync(this IMqttServer server, string topic, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel)
// {
// if (server == null) throw new ArgumentNullException(nameof(server));
// if (topic == null) throw new ArgumentNullException(nameof(topic));
//
// return server.PublishAsync(builder => builder
// .WithTopic(topic)
// .WithPayload(payload)
// .WithQualityOfServiceLevel(qualityOfServiceLevel));
// }
// public static Task<MqttClientPublishResult> PublishAsync(this IMqttServer server, string topic, string payload, MqttQualityOfServiceLevel qualityOfServiceLevel, bool retain)
// {
// if (server == null) throw new ArgumentNullException(nameof(server));
// if (topic == null) throw new ArgumentNullException(nameof(topic));
//
// return server.PublishAsync(builder => builder
// .WithTopic(topic)
// .WithPayload(payload)
// .WithQualityOfServiceLevel(qualityOfServiceLevel)
// .WithRetainFlag(retain));
// }
// public static Task<MqttClientPublishResult> PublishAsync(this IMqttServer server, Func<MqttApplicationMessageBuilder, MqttApplicationMessageBuilder> builder, CancellationToken cancellationToken)
// {
// if (server == null) throw new ArgumentNullException(nameof(server));
//
// var message = builder(new MqttApplicationMessageBuilder()).Build();
// return server.PublishAsync(message, cancellationToken);
// }
//
// public static Task<MqttClientPublishResult> PublishAsync(this IMqttServer server, Func<MqttApplicationMessageBuilder, MqttApplicationMessageBuilder> builder)
// {
// if (server == null) throw new ArgumentNullException(nameof(server));
//
// var message = builder(new MqttApplicationMessageBuilder()).Build();
// return server.PublishAsync(message, CancellationToken.None);
// }
}
}
Source/MQTTnet/Server/Options/MqttServerOptions.cs
-4
View File
@@ -17,9 +17,5 @@ namespace MQTTnet.Server
public TimeSpan DefaultCommunicationTimeout { get; set; } = TimeSpan.FromSeconds(15);
public TimeSpan KeepAliveMonitorInterval { get; set; } = TimeSpan.FromMilliseconds(500);
public IMqttServerStorage Storage { get; set; }
public IMqttRetainedMessagesManager RetainedMessagesManager { get; set; } = new MqttRetainedMessagesManager();
}
}
Source/MQTTnet/Server/Options/MqttServerOptionsBuilder.cs
-12
View File
@@ -140,18 +140,6 @@ namespace MQTTnet.Server
return this;
}
#endif
public MqttServerOptionsBuilder WithStorage(IMqttServerStorage value)
{
_options.Storage = value;
return this;
}
public MqttServerOptionsBuilder WithRetainedMessagesManager(IMqttRetainedMessagesManager value)
{
_options.RetainedMessagesManager = value;
return this;
}
// public MqttServerOptionsBuilder WithApplicationMessageInterceptor(IMqttServerApplicationMessageInterceptor value)
// {
Source/MQTTnet/Server/MqttPublishResponse.cs → Source/MQTTnet/Server/PublishResponse.cs
+1 -1
View File
@@ -3,7 +3,7 @@ using MQTTnet.Packets;
namespace MQTTnet.Server
{
public sealed class MqttApplicationMessageResponse
public sealed class PublishResponse
{
public MqttApplicationMessageResponseReasonCode ReasonCode { get; set; } = MqttApplicationMessageResponseReasonCode.Success;
Source/MQTTnet/Server/Status/MqttClientStatus.cs
+2 -2
View File
@@ -7,9 +7,9 @@ namespace MQTTnet.Server
{
public sealed class MqttClientStatus : IMqttClientStatus
{
readonly MqttClientConnection _connection;
readonly MqttClient _connection;
public MqttClientStatus(MqttClientConnection connection)
public MqttClientStatus(MqttClient connection)
{
_connection = connection ?? throw new ArgumentNullException(nameof(connection));
}
Source/MQTTnet/Server/Status/MqttSessionStatus.cs
+2 -2
View File
@@ -9,9 +9,9 @@ namespace MQTTnet.Server
{
public sealed class MqttSessionStatus : IMqttSessionStatus
{
readonly MqttClientSession _session;
readonly MqttSession _session;
public MqttSessionStatus(MqttClientSession session)
public MqttSessionStatus(MqttSession session)
{
_session = session ?? throw new ArgumentNullException(nameof(session));
}
Source/MQTTnet/Server/Storage/IMqttRetainedMessagesManager.cs
-19
View File
@@ -1,19 +0,0 @@
using System.Collections.Generic;
using System.Threading.Tasks;
using MQTTnet.Diagnostics;
namespace MQTTnet.Server
{
public interface IMqttRetainedMessagesManager
{
Task Start(MqttServerOptions options, IMqttNetLogger logger);
Task LoadMessagesAsync();
Task ClearMessagesAsync();
Task HandleMessageAsync(string clientId, MqttApplicationMessage applicationMessage);
Task<IList<MqttApplicationMessage>> GetMessagesAsync();
}
}
Source/MQTTnet/Server/Storage/IMqttServerPersistedSession.cs
-35
View File
@@ -1,35 +0,0 @@
using System;
using System.Collections.Generic;
using MQTTnet.Packets;
namespace MQTTnet.Server
{
public interface IMqttServerPersistedSession
{
/// <summary>
/// Gets the client identifier.
/// Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues.
/// </summary>
string ClientId { get; }
IDictionary<object, object> Items { get; }
IList<MqttTopicFilter> Subscriptions { get; }
/// <summary>
/// Gets the last will message.
/// In MQTT, you use the last will message feature to notify other clients about an ungracefully disconnected client.
/// </summary>
MqttApplicationMessage WillMessage { get; }
/// <summary>
/// Gets the will delay interval.
/// This is the time between the client disconnect and the time the will message will be sent.
/// </summary>
uint? WillDelayInterval { get; }
DateTime? SessionExpiryTimestamp { get; }
IList<MqttQueuedApplicationMessage> PendingApplicationMessages { get; }
}
}
Source/MQTTnet/Server/Storage/IMqttServerPersistedSessionsStorage.cs
-11
View File
@@ -1,11 +0,0 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public interface IMqttServerPersistedSessionsStorage
{
Task<IList<IMqttServerPersistedSession>> LoadPersistedSessionsAsync();
}
}
Source/MQTTnet/Server/Storage/IMqttServerStorage.cs
-12
View File
@@ -1,12 +0,0 @@
using System.Collections.Generic;
using System.Threading.Tasks;
namespace MQTTnet.Server
{
public interface IMqttServerStorage
{
Task SaveRetainedMessagesAsync(IList<MqttApplicationMessage> messages);
Task<IList<MqttApplicationMessage>> LoadRetainedMessagesAsync();
}
}
Source/MQTTnet/Server/MqttSubscribeResponse.cs → Source/MQTTnet/Server/SubscribeResponse.cs
+1 -1
View File
@@ -4,7 +4,7 @@ using MQTTnet.Protocol;
namespace MQTTnet.Server
{
public sealed class MqttSubscribeResponse
public sealed class SubscribeResponse
{
/// <summary>
/// Gets or sets the reason code which is sent to the client.
Source/MQTTnet/Server/MqttUnsubscribeResponse.cs → Source/MQTTnet/Server/UnsubscribeResponse.cs
+1 -1
View File
@@ -4,7 +4,7 @@ using MQTTnet.Protocol;
namespace MQTTnet.Server
{
public sealed class MqttUnsubscribeResponse
public sealed class UnsubscribeResponse
{
/// <summary>
/// Gets or sets the reason code which is sent to the client.