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

initial version

Browse Source
This commit is contained in:
JanEggers
2017-10-22 19:22:14 +02:00
parent 41edc3a1ec
commit d10f29bbc6
42 changed files with 580 additions and 487 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MQTTnet.Core/Server/IMqttClientSesssionFactory.cs
+7
View File
@@ -0,0 +1,7 @@
namespace MQTTnet.Core.Server
{
public interface IMqttClientSesssionFactory
{
MqttClientSession CreateClientSession(string sessionId, MqttClientSessionsManager mqttClientSessionsManager);
}
}
MQTTnet.Core/Server/IMqttServerFactory.cs
+2 -4
View File
@@ -1,9 +1,7 @@
using MQTTnet.Core.Diagnostics;
namespace MQTTnet.Core.Server
namespace MQTTnet.Core.Server
{
public interface IMqttServerFactory
{
IMqttServer CreateMqttServer(MqttServerOptions options, IMqttNetTraceHandler traceHandler = null);
IMqttServer CreateMqttServer();
}
}
MQTTnet.Core/Server/MqttClientPendingMessagesQueue.cs
+8 -8
View File
@@ -3,10 +3,10 @@ using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Core.Adapter;
using MQTTnet.Core.Diagnostics;
using MQTTnet.Core.Exceptions;
using MQTTnet.Core.Packets;
using MQTTnet.Core.Protocol;
using Microsoft.Extensions.Logging;
namespace MQTTnet.Core.Server
{
@@ -15,11 +15,11 @@ namespace MQTTnet.Core.Server
private readonly BlockingCollection<MqttPublishPacket> _pendingPublishPackets = new BlockingCollection<MqttPublishPacket>();
private readonly MqttClientSession _session;
private readonly MqttServerOptions _options;
private readonly MqttNetTrace _trace;
private readonly ILogger<MqttClientPendingMessagesQueue> _logger;
public MqttClientPendingMessagesQueue(MqttServerOptions options, MqttClientSession session, MqttNetTrace trace)
public MqttClientPendingMessagesQueue(MqttServerOptions options, MqttClientSession session, ILogger<MqttClientPendingMessagesQueue> logger)
{
_trace = trace ?? throw new ArgumentNullException(nameof(trace));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_session = session ?? throw new ArgumentNullException(nameof(session));
_options = options ?? throw new ArgumentNullException(nameof(options));
}
@@ -52,7 +52,7 @@ namespace MQTTnet.Core.Server
}
catch (Exception exception)
{
_trace.Error(nameof(MqttClientPendingMessagesQueue), exception, "Unhandled exception while sending pending publish packets.");
_logger.LogError(new EventId(), exception, "Unhandled exception while sending pending publish packets.");
}
}
@@ -68,18 +68,18 @@ namespace MQTTnet.Core.Server
{
if (exception is MqttCommunicationTimedOutException)
{
_trace.Warning(nameof(MqttClientPendingMessagesQueue), exception, "Sending publish packet failed due to timeout.");
_logger.LogWarning(new EventId(), exception, "Sending publish packet failed due to timeout.");
}
else if (exception is MqttCommunicationException)
{
_trace.Warning(nameof(MqttClientPendingMessagesQueue), exception, "Sending publish packet failed due to communication exception.");
_logger.LogWarning(new EventId(), exception, "Sending publish packet failed due to communication exception.");
}
if (exception is OperationCanceledException)
{
}
else
{
_trace.Error(nameof(MqttClientPendingMessagesQueue), exception, "Sending publish packet failed.");
_logger.LogError(new EventId(), exception, "Sending publish packet failed.");
}
if (packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
MQTTnet.Core/Server/MqttClientRetainedMessagesManager.cs
+10 -9
View File
@@ -2,21 +2,22 @@
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using MQTTnet.Core.Diagnostics;
using MQTTnet.Core.Packets;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
namespace MQTTnet.Core.Server
{
public sealed class MqttClientRetainedMessagesManager
{
private readonly Dictionary<string, MqttApplicationMessage> _retainedMessages = new Dictionary<string, MqttApplicationMessage>();
private readonly MqttNetTrace _trace;
private readonly ILogger<MqttClientRetainedMessagesManager> _logger;
private readonly MqttServerOptions _options;
public MqttClientRetainedMessagesManager(MqttServerOptions options, MqttNetTrace trace)
public MqttClientRetainedMessagesManager(IOptions<MqttServerOptions> options, ILogger<MqttClientRetainedMessagesManager> logger)
{
_trace = trace ?? throw new ArgumentNullException(nameof(trace));
_options = options ?? throw new ArgumentNullException(nameof(options));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_options = options.Value ?? throw new ArgumentNullException(nameof(options));
}
public async Task LoadMessagesAsync()
@@ -40,7 +41,7 @@ namespace MQTTnet.Core.Server
}
catch (Exception exception)
{
_trace.Error(nameof(MqttClientRetainedMessagesManager), exception, "Unhandled exception while loading retained messages.");
_logger.LogError(new EventId(), exception, "Unhandled exception while loading retained messages.");
}
}
@@ -54,12 +55,12 @@ namespace MQTTnet.Core.Server
if (applicationMessage.Payload?.Any() == false)
{
_retainedMessages.Remove(applicationMessage.Topic);
_trace.Information(nameof(MqttClientRetainedMessagesManager), "Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);
_logger.LogInformation("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);
}
else
{
_retainedMessages[applicationMessage.Topic] = applicationMessage;
_trace.Information(nameof(MqttClientRetainedMessagesManager), "Client '{0}' updated retained message for topic '{1}'.", clientId, applicationMessage.Topic);
_logger.LogInformation("Client '{0}' updated retained message for topic '{1}'.", clientId, applicationMessage.Topic);
}
allRetainedMessages = new List<MqttApplicationMessage>(_retainedMessages.Values);
@@ -75,7 +76,7 @@ namespace MQTTnet.Core.Server
}
catch (Exception exception)
{
_trace.Error(nameof(MqttClientRetainedMessagesManager), exception, "Unhandled exception while saving retained messages.");
_logger.LogError(new EventId(), exception, "Unhandled exception while saving retained messages.");
}
}
MQTTnet.Core/Server/MqttClientSession.cs
+14 -13
View File
@@ -3,12 +3,12 @@ using System.Collections.Generic;
using System.Threading;
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;
using MQTTnet.Core.Serializer;
using Microsoft.Extensions.Logging;
namespace MQTTnet.Core.Server
{
@@ -20,19 +20,20 @@ namespace MQTTnet.Core.Server
private readonly MqttClientSessionsManager _mqttClientSessionsManager;
private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue;
private readonly MqttServerOptions _options;
private readonly MqttNetTrace _trace;
private readonly ILogger<MqttClientSession> _logger;
private IMqttCommunicationAdapter _adapter;
private CancellationTokenSource _cancellationTokenSource;
private MqttApplicationMessage _willMessage;
public MqttClientSession(string clientId, MqttServerOptions options, MqttClientSessionsManager mqttClientSessionsManager, MqttNetTrace trace)
public MqttClientSession(string clientId, MqttClientSessionsManager mqttClientSessionsManager, ILogger<MqttClientSession> logger, ILogger<MqttClientPendingMessagesQueue> msgQueueLogger)
{
_trace = trace ?? throw new ArgumentNullException(nameof(trace));
ClientId = clientId;
_options = options ?? throw new ArgumentNullException(nameof(options));
_options = mqttClientSessionsManager.Options;
_mqttClientSessionsManager = mqttClientSessionsManager ?? throw new ArgumentNullException(nameof(mqttClientSessionsManager));
_pendingMessagesQueue = new MqttClientPendingMessagesQueue(options, this, trace);
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_pendingMessagesQueue = new MqttClientPendingMessagesQueue(mqttClientSessionsManager.Options, this, msgQueueLogger);
}
public string ClientId { get; }
@@ -60,11 +61,11 @@ namespace MQTTnet.Core.Server
}
catch (MqttCommunicationException exception)
{
_trace.Warning(nameof(MqttClientSession), exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
_logger.LogWarning(new EventId(), exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
}
catch (Exception exception)
{
_trace.Error(nameof(MqttClientSession), exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
_logger.LogError(new EventId(), exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
}
}
@@ -81,7 +82,7 @@ namespace MQTTnet.Core.Server
_adapter = null;
_trace.Information(nameof(MqttClientSession), "Client '{0}': Disconnected.", ClientId);
_logger.LogInformation("Client '{0}': Disconnected.", ClientId);
}
public void EnqueuePublishPacket(MqttPublishPacket publishPacket)
@@ -94,7 +95,7 @@ namespace MQTTnet.Core.Server
}
_pendingMessagesQueue.Enqueue(publishPacket);
_trace.Verbose(nameof(MqttClientSession), "Client '{0}': Enqueued pending publish packet.", ClientId);
_logger.LogTrace("Client '{0}': Enqueued pending publish packet.", ClientId);
}
public void Dispose()
@@ -118,12 +119,12 @@ namespace MQTTnet.Core.Server
}
catch (MqttCommunicationException exception)
{
_trace.Warning(nameof(MqttClientSession), exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
_logger.LogWarning(new EventId(), exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
Stop();
}
catch (Exception exception)
{
_trace.Error(nameof(MqttClientSession), exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
_logger.LogError(new EventId(), exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
Stop();
}
}
@@ -165,7 +166,7 @@ namespace MQTTnet.Core.Server
}
else
{
_trace.Warning(nameof(MqttClientSession), "Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet);
_logger.LogWarning("Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet);
Stop();
}
}
MQTTnet.Core/Server/MqttClientSessionsManager.cs
+26 -20
View File
@@ -4,26 +4,28 @@ using System.Linq;
using System.Threading;
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;
using MQTTnet.Core.Serializer;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
namespace MQTTnet.Core.Server
{
public sealed class MqttClientSessionsManager
{
private readonly Dictionary<string, MqttClientSession> _clientSessions = new Dictionary<string, MqttClientSession>();
private readonly MqttServerOptions _options;
private readonly MqttNetTrace _trace;
private readonly ILogger<MqttClientSessionsManager> _logger;
private readonly IMqttClientSesssionFactory _mqttClientSesssionFactory;
public MqttClientSessionsManager(MqttServerOptions options, MqttNetTrace trace)
public MqttClientSessionsManager(IOptions<MqttServerOptions> options, ILogger<MqttClientSessionsManager> logger, MqttClientRetainedMessagesManager retainedMessagesManager, IMqttClientSesssionFactory mqttClientSesssionFactory)
{
_trace = trace ?? throw new ArgumentNullException(nameof(trace));
_options = options ?? throw new ArgumentNullException(nameof(options));
RetainedMessagesManager = new MqttClientRetainedMessagesManager(options, trace);
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
Options = options.Value ?? throw new ArgumentNullException(nameof(options));
RetainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(options));
_mqttClientSesssionFactory = mqttClientSesssionFactory ?? throw new ArgumentNullException(nameof(mqttClientSesssionFactory));
}
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
@@ -31,13 +33,14 @@ namespace MQTTnet.Core.Server
public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
public MqttClientRetainedMessagesManager RetainedMessagesManager { get; }
public MqttServerOptions Options { get; }
public async Task RunClientSessionAsync(IMqttCommunicationAdapter clientAdapter)
{
var clientId = string.Empty;
try
{
if (!(await clientAdapter.ReceivePacketAsync(_options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false) is MqttConnectPacket connectPacket))
if (!(await clientAdapter.ReceivePacketAsync(Options.DefaultCommunicationTimeout, CancellationToken.None).ConfigureAwait(false) is MqttConnectPacket connectPacket))
{
throw new MqttProtocolViolationException("The first packet from a client must be a 'CONNECT' packet [MQTT-3.1.0-1].");
}
@@ -50,7 +53,7 @@ namespace MQTTnet.Core.Server
var connectReturnCode = ValidateConnection(connectPacket);
if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted)
{
await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, CancellationToken.None, new MqttConnAckPacket
await clientAdapter.SendPacketsAsync(Options.DefaultCommunicationTimeout, CancellationToken.None, new MqttConnAckPacket
{
ConnectReturnCode = connectReturnCode
}).ConfigureAwait(false);
@@ -60,7 +63,7 @@ namespace MQTTnet.Core.Server
var clientSession = GetOrCreateClientSession(connectPacket);
await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, CancellationToken.None, new MqttConnAckPacket
await clientAdapter.SendPacketsAsync(Options.DefaultCommunicationTimeout, CancellationToken.None, new MqttConnAckPacket
{
ConnectReturnCode = connectReturnCode,
IsSessionPresent = clientSession.IsExistingSession
@@ -72,17 +75,20 @@ namespace MQTTnet.Core.Server
ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion
}));
await clientSession.Session.RunAsync(connectPacket.WillMessage, clientAdapter).ConfigureAwait(false);
using (_logger.BeginScope(clientId))
{
await clientSession.Session.RunAsync(connectPacket.WillMessage, clientAdapter).ConfigureAwait(false);
}
}
catch (Exception exception)
{
_trace.Error(nameof(MqttServer), exception, exception.Message);
_logger.LogError(new EventId(), exception, exception.Message);
}
finally
{
try
{
await clientAdapter.DisconnectAsync(_options.DefaultCommunicationTimeout).ConfigureAwait(false);
await clientAdapter.DisconnectAsync(Options.DefaultCommunicationTimeout).ConfigureAwait(false);
}
catch (Exception)
{
@@ -126,7 +132,7 @@ namespace MQTTnet.Core.Server
}
catch (Exception exception)
{
_trace.Error(nameof(MqttClientSessionsManager), exception, "Error while processing application message");
_logger.LogError(new EventId(), exception, "Error while processing application message");
}
lock (_clientSessions)
@@ -140,9 +146,9 @@ namespace MQTTnet.Core.Server
private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket)
{
if (_options.ConnectionValidator != null)
if (Options.ConnectionValidator != null)
{
return _options.ConnectionValidator(connectPacket);
return Options.ConnectionValidator(connectPacket);
}
return MqttConnectReturnCode.ConnectionAccepted;
@@ -160,11 +166,11 @@ namespace MQTTnet.Core.Server
_clientSessions.Remove(connectPacket.ClientId);
clientSession.Dispose();
clientSession = null;
_trace.Verbose(nameof(MqttClientSessionsManager), "Disposed existing session of client '{0}'.", connectPacket.ClientId);
_logger.LogTrace("Disposed existing session of client '{0}'.", connectPacket.ClientId);
}
else
{
_trace.Verbose(nameof(MqttClientSessionsManager), "Reusing existing session of client '{0}'.", connectPacket.ClientId);
_logger.LogTrace("Reusing existing session of client '{0}'.", connectPacket.ClientId);
}
}
@@ -173,10 +179,10 @@ namespace MQTTnet.Core.Server
{
isExistingSession = false;
clientSession = new MqttClientSession(connectPacket.ClientId, _options, this, _trace);
clientSession = _mqttClientSesssionFactory.CreateClientSession(connectPacket.ClientId, this);
_clientSessions[connectPacket.ClientId] = clientSession;
_trace.Verbose(nameof(MqttClientSessionsManager), "Created a new session for client '{0}'.", connectPacket.ClientId);
_logger.LogTrace("Created a new session for client '{0}'.", connectPacket.ClientId);
}
return new GetOrCreateClientSessionResult { IsExistingSession = isExistingSession, Session = clientSession };
MQTTnet.Core/Server/MqttServer.cs
+18 -11
View File
@@ -3,26 +3,33 @@ using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Core.Adapter;
using MQTTnet.Core.Diagnostics;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using System.Linq;
namespace MQTTnet.Core.Server
{
public sealed class MqttServer : IMqttServer
{
private readonly MqttNetTrace _trace;
private readonly ILogger<MqttServer> _logger;
private readonly MqttClientSessionsManager _clientSessionsManager;
private readonly ICollection<IMqttServerAdapter> _adapters;
private readonly MqttServerOptions _options;
private CancellationTokenSource _cancellationTokenSource;
public MqttServer(MqttServerOptions options, ICollection<IMqttServerAdapter> adapters, MqttNetTrace trace)
public MqttServer(IOptions<MqttServerOptions> options, IEnumerable<IMqttServerAdapter> adapters, ILogger<MqttServer> logger, MqttClientSessionsManager clientSessionsManager)
{
_trace = trace ?? throw new ArgumentNullException(nameof(trace));
_options = options ?? throw new ArgumentNullException(nameof(options));
_adapters = adapters ?? throw new ArgumentNullException(nameof(adapters));
_options = options.Value ?? throw new ArgumentNullException(nameof(options));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_clientSessionsManager = clientSessionsManager ?? throw new ArgumentNullException(nameof(clientSessionsManager));
if (adapters == null)
{
throw new ArgumentNullException(nameof(adapters));
}
_adapters = adapters.ToList();
_clientSessionsManager = new MqttClientSessionsManager(options, trace);
_clientSessionsManager.ApplicationMessageReceived += (s, e) => ApplicationMessageReceived?.Invoke(s, e);
_clientSessionsManager.ClientConnected += OnClientConnected;
_clientSessionsManager.ClientDisconnected += OnClientDisconnected;
@@ -59,7 +66,7 @@ namespace MQTTnet.Core.Server
await adapter.StartAsync(_options);
}
_trace.Information(nameof(MqttServer), "Started.");
_logger.LogInformation("Started.");
}
public async Task StopAsync()
@@ -76,7 +83,7 @@ namespace MQTTnet.Core.Server
_clientSessionsManager.Clear();
_trace.Information(nameof(MqttServer), "Stopped.");
_logger.LogInformation("Stopped.");
}
private void OnClientAccepted(object sender, MqttServerAdapterClientAcceptedEventArgs eventArgs)
@@ -86,13 +93,13 @@ namespace MQTTnet.Core.Server
private void OnClientConnected(object sender, MqttClientConnectedEventArgs eventArgs)
{
_trace.Information(nameof(MqttServer), "Client '{0}': Connected.", eventArgs.Client.ClientId);
_logger.LogInformation("Client '{0}': Connected.", eventArgs.Client.ClientId);
ClientConnected?.Invoke(this, eventArgs);
}
private void OnClientDisconnected(object sender, MqttClientDisconnectedEventArgs eventArgs)
{
_trace.Information(nameof(MqttServer), "Client '{0}': Disconnected.", eventArgs.Client.ClientId);
_logger.LogInformation("Client '{0}': Disconnected.", eventArgs.Client.ClientId);
ClientDisconnected?.Invoke(this, eventArgs);
}
}