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

Removed 3rd-Party libraries.

Browse Source
This commit is contained in:
Christian Kratky
2017-11-22 22:30:55 +01:00
parent e0235aedff
commit 77b80dc53e
66 changed files with 626 additions and 1005 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MQTTnet.Core/Adapter/IMqttCommunicationAdapter.cs → MQTTnet.Core/Adapter/IMqttChannelAdapter.cs
+1 -1
View File
@@ -7,7 +7,7 @@ using MQTTnet.Core.Serializer;
namespace MQTTnet.Core.Adapter
{
public interface IMqttCommunicationAdapter
public interface IMqttChannelAdapter
{
IMqttPacketSerializer PacketSerializer { get; }
MQTTnet.Core/Adapter/IMqttCommunicationAdapterFactory.cs
-12
View File
@@ -1,12 +0,0 @@
using MQTTnet.Core.Client;
using MQTTnet.Core.Channel;
namespace MQTTnet.Core.Adapter
{
public interface IMqttCommunicationAdapterFactory
{
IMqttCommunicationAdapter CreateClientCommunicationAdapter(IMqttClientOptions options);
IMqttCommunicationAdapter CreateServerCommunicationAdapter(IMqttCommunicationChannel channel);
}
}
MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs → MQTTnet.Core/Adapter/MqttChannelAdapter.cs
+9 -9
View File
@@ -5,23 +5,23 @@ using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Core.Channel;
using MQTTnet.Core.Diagnostics;
using MQTTnet.Core.Exceptions;
using MQTTnet.Core.Internal;
using MQTTnet.Core.Packets;
using MQTTnet.Core.Serializer;
using Microsoft.Extensions.Logging;
namespace MQTTnet.Core.Adapter
{
public class MqttChannelCommunicationAdapter : IMqttCommunicationAdapter
public class MqttChannelAdapter : IMqttChannelAdapter
{
private const uint ErrorOperationAborted = 0x800703E3;
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
private readonly ILogger<MqttChannelCommunicationAdapter> _logger;
private readonly IMqttCommunicationChannel _channel;
private readonly IMqttNetLogger _logger;
private readonly IMqttChannel _channel;
public MqttChannelCommunicationAdapter(IMqttCommunicationChannel channel, IMqttPacketSerializer serializer, ILogger<MqttChannelCommunicationAdapter> logger)
public MqttChannelAdapter(IMqttChannel channel, IMqttPacketSerializer serializer, IMqttNetLogger logger)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_channel = channel ?? throw new ArgumentNullException(nameof(channel));
@@ -32,14 +32,14 @@ namespace MQTTnet.Core.Adapter
public async Task ConnectAsync(TimeSpan timeout)
{
_logger.LogInformation("Connecting [Timeout={0}]", timeout);
_logger.Info<MqttChannelAdapter>("Connecting [Timeout={0}]", timeout);
await ExecuteAndWrapExceptionAsync(() => _channel.ConnectAsync().TimeoutAfter(timeout));
}
public async Task DisconnectAsync(TimeSpan timeout)
{
_logger.LogInformation("Disconnecting [Timeout={0}]", timeout);
_logger.Info<MqttChannelAdapter>("Disconnecting [Timeout={0}]", timeout);
await ExecuteAndWrapExceptionAsync(() => _channel.DisconnectAsync().TimeoutAfter(timeout));
}
@@ -58,7 +58,7 @@ namespace MQTTnet.Core.Adapter
continue;
}
_logger.LogInformation("TX >>> {0} [Timeout={1}]", packet, timeout);
_logger.Trace<MqttChannelAdapter>("TX >>> {0} [Timeout={1}]", packet, timeout);
var writeBuffer = PacketSerializer.Serialize(packet);
await _channel.SendStream.WriteAsync(writeBuffer, 0, writeBuffer.Length, cancellationToken).ConfigureAwait(false);
@@ -108,7 +108,7 @@ namespace MQTTnet.Core.Adapter
throw new MqttProtocolViolationException("Received malformed packet.");
}
_logger.LogInformation("RX <<< {0}", packet);
_logger.Trace<MqttChannelAdapter>("RX <<< {0}", packet);
}
finally
{
MQTTnet.Core/Adapter/MqttCommunicationAdapterExtensions.cs → MQTTnet.Core/Adapter/MqttChannelAdapterExtensions.cs
+2 -2
View File
@@ -5,9 +5,9 @@ using MQTTnet.Core.Packets;
namespace MQTTnet.Core.Adapter
{
public static class MqttCommunicationAdapterExtensions
public static class MqttChannelAdapterExtensions
{
public static Task SendPacketsAsync(this IMqttCommunicationAdapter adapter, TimeSpan timeout, CancellationToken cancellationToken, params MqttBasePacket[] packets)
public static Task SendPacketsAsync(this IMqttChannelAdapter adapter, TimeSpan timeout, CancellationToken cancellationToken, params MqttBasePacket[] packets)
{
if (adapter == null) throw new ArgumentNullException(nameof(adapter));
MQTTnet.Core/Adapter/MqttServerAdapterClientAcceptedEventArgs.cs
+2 -2
View File
@@ -5,12 +5,12 @@ namespace MQTTnet.Core.Adapter
{
public class MqttServerAdapterClientAcceptedEventArgs : EventArgs
{
public MqttServerAdapterClientAcceptedEventArgs(IMqttCommunicationAdapter client)
public MqttServerAdapterClientAcceptedEventArgs(IMqttChannelAdapter client)
{
Client = client ?? throw new ArgumentNullException(nameof(client));
}
public IMqttCommunicationAdapter Client { get; }
public IMqttChannelAdapter Client { get; }
public Task SessionTask { get; set; }
}
MQTTnet.Core/Channel/IMqttCommunicationChannel.cs → MQTTnet.Core/Channel/IMqttChannel.cs
+1 -1
View File
@@ -3,7 +3,7 @@ using System.IO;
namespace MQTTnet.Core.Channel
{
public interface IMqttCommunicationChannel
public interface IMqttChannel
{
Stream SendStream { get; }
Stream ReceiveStream { get; }
MQTTnet.Core/Client/IMqttClientAdapterFactory.cs
+10
View File
@@ -0,0 +1,10 @@
using MQTTnet.Core.Adapter;
using MQTTnet.Core.Diagnostics;
namespace MQTTnet.Core.Client
{
public interface IMqttClientAdapterFactory
{
IMqttChannelAdapter CreateClientAdapter(IMqttClientChannelOptions options, IMqttNetLogger logger);
}
}
MQTTnet.Core/Client/IMqttClientFactory.cs
+6 -1
View File
@@ -1,4 +1,5 @@
using MQTTnet.Core.ManagedClient;
using MQTTnet.Core.Diagnostics;
using MQTTnet.Core.ManagedClient;
namespace MQTTnet.Core.Client
{
@@ -6,6 +7,10 @@ namespace MQTTnet.Core.Client
{
IMqttClient CreateMqttClient();
IMqttClient CreateMqttClient(IMqttNetLogger logger);
IManagedMqttClient CreateManagedMqttClient();
IManagedMqttClient CreateManagedMqttClient(IMqttNetLogger logger);
}
}
MQTTnet.Core/Client/IMqttClientOptions.cs
-4
View File
@@ -7,10 +7,6 @@ namespace MQTTnet.Core.Client
{
string ClientId { get; }
/// <summary>
/// The LogId is used to create a scope to correlate logging. If no value is provided the ClientId is used instead
/// </summary>
string LogId { get; }
IMqttClientCredentials Credentials { get; }
bool CleanSession { get; }
MqttApplicationMessage WillMessage { get; }
MQTTnet.Core/Client/MqttClient.cs
+29 -31
View File
@@ -4,33 +4,34 @@ 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 Microsoft.Extensions.Logging;
namespace MQTTnet.Core.Client
{
public class MqttClient : IMqttClient
{
private readonly HashSet<ushort> _unacknowledgedPublishPackets = new HashSet<ushort>();
private readonly IMqttClientAdapterFactory _adapterFactory;
private readonly MqttPacketDispatcher _packetDispatcher;
private readonly IMqttCommunicationAdapterFactory _communicationAdapterFactory;
private readonly ILogger<MqttClient> _logger;
private readonly IMqttNetLogger _logger;
private IMqttClientOptions _options;
private bool _isReceivingPackets;
private int _latestPacketIdentifier;
private CancellationTokenSource _cancellationTokenSource;
private IMqttCommunicationAdapter _adapter;
private IDisposable _scopeHandle;
private IMqttChannelAdapter _adapter;
public MqttClient(IMqttCommunicationAdapterFactory communicationAdapterFactory, ILogger<MqttClient> logger, MqttPacketDispatcher packetDispatcher)
public MqttClient(IMqttClientAdapterFactory channelFactory, IMqttNetLogger logger)
{
_communicationAdapterFactory = communicationAdapterFactory ?? throw new ArgumentNullException(nameof(communicationAdapterFactory));
_adapterFactory = channelFactory ?? throw new ArgumentNullException(nameof(channelFactory));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_packetDispatcher = packetDispatcher ?? throw new ArgumentNullException(nameof(packetDispatcher));
_packetDispatcher = new MqttPacketDispatcher(logger);
}
public event EventHandler<MqttClientConnectedEventArgs> Connected;
@@ -53,17 +54,16 @@ namespace MQTTnet.Core.Client
_latestPacketIdentifier = 0;
_packetDispatcher.Reset();
_adapter = _communicationAdapterFactory.CreateClientCommunicationAdapter(options);
_scopeHandle = _logger.BeginScope(options.LogId ?? options.ClientId);
_logger.LogTrace("Trying to connect with server.");
_adapter = _adapterFactory.CreateClientAdapter(options.ChannelOptions, _logger);
_logger.Trace<MqttClient>("Trying to connect with server.");
await _adapter.ConnectAsync(_options.CommunicationTimeout).ConfigureAwait(false);
_logger.LogTrace("Connection with server established.");
_logger.Trace<MqttClient>("Connection with server established.");
await StartReceivingPacketsAsync(_cancellationTokenSource.Token).ConfigureAwait(false);
var connectResponse = await AuthenticateAsync(options.WillMessage).ConfigureAwait(false);
_logger.LogTrace("MQTT connection with server established.");
_logger.Trace<MqttClient>("MQTT connection with server established.");
if (_options.KeepAlivePeriod != TimeSpan.Zero)
{
@@ -76,7 +76,7 @@ namespace MQTTnet.Core.Client
}
catch (Exception exception)
{
_logger.LogError(new EventId(), exception, "Error while connecting with server.");
_logger.Error<MqttClient>(exception, "Error while connecting with server.");
await DisconnectInternalAsync().ConfigureAwait(false);
throw;
}
@@ -215,8 +215,6 @@ namespace MQTTnet.Core.Client
private async Task DisconnectInternalAsync()
{
_scopeHandle?.Dispose();
var clientWasConnected = IsConnected;
IsConnected = false;
@@ -233,15 +231,15 @@ namespace MQTTnet.Core.Client
try
{
await _adapter.DisconnectAsync(_options.CommunicationTimeout).ConfigureAwait(false);
_logger.LogInformation("Disconnected from adapter.");
_logger.Info<MqttClient>("Disconnected from adapter.");
}
catch (Exception exception)
{
_logger.LogWarning(new EventId(), exception, "Error while disconnecting from adapter.");
_logger.Warning<MqttClient>(exception, "Error while disconnecting from adapter.");
}
finally
{
_logger.LogInformation("Disconnected.");
_logger.Info<MqttClient>("Disconnected.");
Disconnected?.Invoke(this, new MqttClientDisconnectedEventArgs(clientWasConnected));
}
}
@@ -250,7 +248,7 @@ namespace MQTTnet.Core.Client
{
try
{
_logger.LogInformation("Received <<< {0}", packet);
_logger.Info<MqttClient>("Received <<< {0}", packet);
if (packet is MqttPingReqPacket)
{
@@ -280,7 +278,7 @@ namespace MQTTnet.Core.Client
}
catch (Exception exception)
{
_logger.LogError(new EventId(), exception, "Unhandled exception while processing received packet.");
_logger.Error<MqttClient>(exception, "Unhandled exception while processing received packet.");
}
}
@@ -293,7 +291,7 @@ namespace MQTTnet.Core.Client
}
catch (Exception exception)
{
_logger.LogError(new EventId(), exception, "Unhandled exception while handling application message.");
_logger.Error<MqttClient>(exception, "Unhandled exception while handling application message.");
}
}
@@ -357,7 +355,7 @@ namespace MQTTnet.Core.Client
private async Task SendKeepAliveMessagesAsync(CancellationToken cancellationToken)
{
_logger.LogInformation("Start sending keep alive packets.");
_logger.Info<MqttClient>("Start sending keep alive packets.");
try
{
@@ -388,23 +386,23 @@ namespace MQTTnet.Core.Client
return;
}
_logger.LogWarning(new EventId(), exception, "MQTT communication exception while sending/receiving keep alive packets.");
_logger.Warning<MqttClient>(exception, "MQTT communication exception while sending/receiving keep alive packets.");
await DisconnectInternalAsync().ConfigureAwait(false);
}
catch (Exception exception)
{
_logger.LogWarning(new EventId(), exception, "Unhandled exception while sending/receiving keep alive packets.");
_logger.Warning<MqttClient>(exception, "Unhandled exception while sending/receiving keep alive packets.");
await DisconnectInternalAsync().ConfigureAwait(false);
}
finally
{
_logger.LogInformation("Stopped sending keep alive packets.");
_logger.Info<MqttClient>("Stopped sending keep alive packets.");
}
}
private async Task ReceivePacketsAsync(CancellationToken cancellationToken)
{
_logger.LogInformation("Start receiving packets.");
_logger.Info<MqttClient>("Start receiving packets.");
try
{
@@ -438,17 +436,17 @@ namespace MQTTnet.Core.Client
return;
}
_logger.LogWarning(new EventId(), exception, "MQTT communication exception while receiving packets.");
_logger.Warning<MqttClient>(exception, "MQTT communication exception while receiving packets.");
await DisconnectInternalAsync().ConfigureAwait(false);
}
catch (Exception exception)
{
_logger.LogError(new EventId(), exception, "Unhandled exception while receiving packets.");
_logger.Error<MqttClient>(exception, "Unhandled exception while receiving packets.");
await DisconnectInternalAsync().ConfigureAwait(false);
}
finally
{
_logger.LogInformation(nameof(MqttClient), "Stopped receiving packets.");
_logger.Info<MqttClient>("Stopped receiving packets.");
}
}
MQTTnet.Core/Client/MqttClientOptions.cs
-3
View File
@@ -9,9 +9,6 @@ namespace MQTTnet.Core.Client
public string ClientId { get; set; } = Guid.NewGuid().ToString("N");
/// <inheritdoc />
public string LogId { get; set; }
public bool CleanSession { get; set; } = true;
public IMqttClientCredentials Credentials { get; set; } = new MqttClientCredentials();
MQTTnet.Core/Client/MqttClientOptionsBuilder.cs
-6
View File
@@ -12,12 +12,6 @@ namespace MQTTnet.Core.Client
private MqttClientTlsOptions _tlsOptions;
public MqttClientOptionsBuilder WithLogId(string value)
{
_options.LogId = value;
return this;
}
public MqttClientOptionsBuilder WithProtocolVersion(MqttProtocolVersion value)
{
_options.ProtocolVersion = value;
MQTTnet.Core/Client/MqttPacketDispatcher.cs
+4 -4
View File
@@ -1,10 +1,10 @@
using System;
using System.Collections.Concurrent;
using System.Threading.Tasks;
using MQTTnet.Core.Diagnostics;
using MQTTnet.Core.Exceptions;
using MQTTnet.Core.Internal;
using MQTTnet.Core.Packets;
using Microsoft.Extensions.Logging;
namespace MQTTnet.Core.Client
{
@@ -12,9 +12,9 @@ namespace MQTTnet.Core.Client
{
private readonly ConcurrentDictionary<Type, TaskCompletionSource<MqttBasePacket>> _packetByResponseType = new ConcurrentDictionary<Type, TaskCompletionSource<MqttBasePacket>>();
private readonly ConcurrentDictionary<Type, ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>> _packetByResponseTypeAndIdentifier = new ConcurrentDictionary<Type, ConcurrentDictionary<ushort, TaskCompletionSource<MqttBasePacket>>>();
private readonly ILogger<MqttPacketDispatcher> _logger;
private readonly IMqttNetLogger _logger;
public MqttPacketDispatcher(ILogger<MqttPacketDispatcher> logger)
public MqttPacketDispatcher(IMqttNetLogger logger)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
}
@@ -30,7 +30,7 @@ namespace MQTTnet.Core.Client
}
catch (MqttCommunicationTimedOutException)
{
_logger.LogWarning("Timeout while waiting for packet of type '{0}'.", responseType.Name);
_logger.Warning<MqttPacketDispatcher>("Timeout while waiting for packet of type '{0}'.", responseType.Name);
throw;
}
finally
MQTTnet.Core/Diagnostics/IMqttNetLogger.cs
+21
View File
@@ -0,0 +1,21 @@
using System;
namespace MQTTnet.Core.Diagnostics
{
public interface IMqttNetLogger
{
event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;
void Trace<TSource>(string message, params object[] parameters);
void Info<TSource>(string message, params object[] parameters);
void Warning<TSource>(Exception exception, string message, params object[] parameters);
void Warning<TSource>(string message, params object[] parameters);
void Error<TSource>(Exception exception, string message, params object[] parameters);
void Error<TSource>(string message, params object[] parameters);
}
}
MQTTnet.Core/Diagnostics/MqttNetGlobalLog.cs
+18
View File
@@ -0,0 +1,18 @@
using System;
namespace MQTTnet.Core.Diagnostics
{
public static class MqttNetGlobalLog
{
public static event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;
public static bool HasListeners => LogMessagePublished != null;
public static void Publish(MqttNetLogMessage logMessage)
{
if (logMessage == null) throw new ArgumentNullException(nameof(logMessage));
LogMessagePublished?.Invoke(null, new MqttNetLogMessagePublishedEventArgs(logMessage));
}
}
}
MQTTnet.Core/Diagnostics/MqttNetLogLevel.cs
+13
View File
@@ -0,0 +1,13 @@
namespace MQTTnet.Core.Diagnostics
{
public enum MqttNetLogLevel
{
Verbose,
Info,
Warning,
Error
}
}
MQTTnet.Core/Diagnostics/MqttTraceMessage.cs → MQTTnet.Core/Diagnostics/MqttNetLogMessage.cs
+6 -4
View File
@@ -1,12 +1,12 @@
using System;
using Microsoft.Extensions.Logging;
namespace MQTTnet.Core.Diagnostics
{
public sealed class MqttNetTraceMessage
public sealed class MqttNetLogMessage
{
public MqttNetTraceMessage(DateTime timestamp, int threadId, string source, LogLevel level, string message, Exception exception)
public MqttNetLogMessage(string logId, DateTime timestamp, int threadId, string source, MqttNetLogLevel level, string message, Exception exception)
{
LogId = logId;
Timestamp = timestamp;
ThreadId = threadId;
Source = source;
@@ -15,13 +15,15 @@ namespace MQTTnet.Core.Diagnostics
Exception = exception;
}
public string LogId { get; }
public DateTime Timestamp { get; }
public int ThreadId { get; }
public string Source { get; }
public LogLevel Level { get; }
public MqttNetLogLevel Level { get; }
public string Message { get; }
MQTTnet.Core/Diagnostics/MqttNetLogMessagePublishedEventArgs.cs
+14
View File
@@ -0,0 +1,14 @@
using System;
namespace MQTTnet.Core.Diagnostics
{
public sealed class MqttNetLogMessagePublishedEventArgs : EventArgs
{
public MqttNetLogMessagePublishedEventArgs(MqttNetLogMessage logMessage)
{
TraceMessage = logMessage ?? throw new ArgumentNullException(nameof(logMessage));
}
public MqttNetLogMessage TraceMessage { get; }
}
}
MQTTnet.Core/Diagnostics/MqttNetLogger.cs
+54 -33
View File
@@ -1,52 +1,73 @@
using System;
using Microsoft.Extensions.Logging;
namespace MQTTnet.Core.Diagnostics
{
public class MqttNetLogger : ILogger
public class MqttNetLogger : IMqttNetLogger
{
private readonly string _categoryName;
private readonly MqttNetTrace _mqttNetTrace;
private readonly string _logId;
public MqttNetLogger(string categoryName, MqttNetTrace mqttNetTrace)
public MqttNetLogger(string logId = null)
{
_categoryName = categoryName;
_mqttNetTrace = mqttNetTrace;
_logId = logId;
}
public event EventHandler<MqttNetLogMessagePublishedEventArgs> LogMessagePublished;
public void Trace<TSource>(string message, params object[] parameters)
{
Publish<TSource>(MqttNetLogLevel.Verbose, null, message, parameters);
}
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception exception, Func<TState, Exception, string> formatter)
public void Info<TSource>(string message, params object[] parameters)
{
if (formatter == null)
{
throw new ArgumentNullException(nameof(formatter));
}
Publish<TSource>(MqttNetLogLevel.Info, null, message, parameters);
}
if (!MqttNetTrace.HasListeners)
public void Warning<TSource>(Exception exception, string message, params object[] parameters)
{
Publish<TSource>(MqttNetLogLevel.Warning, exception, message, parameters);
}
public void Warning<TSource>(string message, params object[] parameters)
{
Warning<TSource>(null, message, parameters);
}
public void Error<TSource>(Exception exception, string message, params object[] parameters)
{
Publish<TSource>(MqttNetLogLevel.Error, exception, message, parameters);
}
public void Error<TSource>(string message, params object[] parameters)
{
Warning<TSource>(null, message, parameters);
}
private void Publish<TSource>(MqttNetLogLevel logLevel, Exception exception, string message, object[] parameters)
{
var hasLocalListeners = LogMessagePublished != null;
var hasGlobalListeners = MqttNetGlobalLog.HasListeners;
if (!hasLocalListeners && !hasGlobalListeners)
{
return;
}
var message = formatter(state, exception);
var traceMessage = new MqttNetTraceMessage(DateTime.Now, Environment.CurrentManagedThreadId, _categoryName, logLevel, message, exception);
_mqttNetTrace.Publish(traceMessage);
}
public bool IsEnabled(LogLevel logLevel)
{
return MqttNetTrace.HasListeners;
}
//not supported: async local requires netstandard1.3
//for implementation see https://github.com/aspnet/Logging/blob/dev/src/Microsoft.Extensions.Logging.Console/ConsoleLogScope.cs
public IDisposable BeginScope<TState>(TState state)
{
return new DisposableScope();
}
private class DisposableScope : IDisposable
{
public void Dispose()
if (parameters.Length > 0)
{
message = string.Format(message, parameters);
}
var traceMessage = new MqttNetLogMessage(_logId, DateTime.Now, Environment.CurrentManagedThreadId, typeof(TSource).Name, logLevel, message, exception);
if (hasGlobalListeners)
{
MqttNetGlobalLog.Publish(traceMessage);
}
if (hasLocalListeners)
{
LogMessagePublished?.Invoke(this, new MqttNetLogMessagePublishedEventArgs(traceMessage));
}
}
}
MQTTnet.Core/Diagnostics/MqttNetTrace.cs
-35
View File
@@ -1,35 +0,0 @@
using System;
using System.Collections.Concurrent;
using Microsoft.Extensions.Logging;
namespace MQTTnet.Core.Diagnostics
{
public class MqttNetTrace : ILoggerProvider
{
private readonly ConcurrentDictionary<string, MqttNetLogger> _loggers = new ConcurrentDictionary<string, MqttNetLogger>();
public static event EventHandler<MqttNetTraceMessagePublishedEventArgs> TraceMessagePublished;
public static bool HasListeners => TraceMessagePublished != null;
public void Publish(MqttNetTraceMessage traceMessage)
{
TraceMessagePublished?.Invoke(this, new MqttNetTraceMessagePublishedEventArgs(traceMessage));
}
public void Dispose()
{
TraceMessagePublished = null;
}
public ILogger CreateLogger(string categoryName)
{
return _loggers.GetOrAdd(categoryName, CreateLoggerImplementation);
}
private MqttNetLogger CreateLoggerImplementation(string categoryName)
{
return new MqttNetLogger(categoryName, this);
}
}
}
MQTTnet.Core/Diagnostics/MqttNetTraceMessagePublishedEventArgs.cs
-14
View File
@@ -1,14 +0,0 @@
using System;
namespace MQTTnet.Core.Diagnostics
{
public sealed class MqttNetTraceMessagePublishedEventArgs : EventArgs
{
public MqttNetTraceMessagePublishedEventArgs(MqttNetTraceMessage traceMessage)
{
TraceMessage = traceMessage ?? throw new ArgumentNullException(nameof(traceMessage));
}
public MqttNetTraceMessage TraceMessage { get; }
}
}
MQTTnet.Core/MQTTnet.Core.csproj
+2 -8
View File
@@ -17,15 +17,9 @@
<PackageIconUrl></PackageIconUrl>
<RepositoryUrl></RepositoryUrl>
<PackageTags></PackageTags>
<FileVersion>2.5.2.0</FileVersion>
<AssemblyVersion>2.5.2.0</AssemblyVersion>
<FileVersion>2.5.3.0</FileVersion>
<AssemblyVersion>2.5.3.0</AssemblyVersion>
<PackageLicenseUrl></PackageLicenseUrl>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="1.0.2" />
<PackageReference Include="Microsoft.Extensions.Options" Version="1.0.2" />
<PackageReference Include="System.Threading" Version="4.3.0" />
</ItemGroup>
</Project>
MQTTnet.Core/ManagedClient/ManagedMqttClient.cs
+19 -19
View File
@@ -5,9 +5,9 @@ using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Core.Client;
using MQTTnet.Core.Diagnostics;
using MQTTnet.Core.Exceptions;
using MQTTnet.Core.Protocol;
using Microsoft.Extensions.Logging;
namespace MQTTnet.Core.ManagedClient
{
@@ -18,7 +18,7 @@ namespace MQTTnet.Core.ManagedClient
private readonly HashSet<TopicFilter> _subscriptions = new HashSet<TopicFilter>();
private readonly IMqttClient _mqttClient;
private readonly ILogger<ManagedMqttClient> _logger;
private readonly IMqttNetLogger _logger;
private CancellationTokenSource _connectionCancellationToken;
private CancellationTokenSource _publishingCancellationToken;
@@ -26,7 +26,7 @@ namespace MQTTnet.Core.ManagedClient
private IManagedMqttClientOptions _options;
private bool _subscriptionsNotPushed;
public ManagedMqttClient(ILogger<ManagedMqttClient> logger, IMqttClient mqttClient)
public ManagedMqttClient(IMqttClient mqttClient, IMqttNetLogger logger)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_mqttClient = mqttClient ?? throw new ArgumentNullException(nameof(mqttClient));
@@ -52,10 +52,7 @@ namespace MQTTnet.Core.ManagedClient
throw new NotSupportedException("The managed client does not support existing sessions.");
}
if (_connectionCancellationToken != null)
{
throw new InvalidOperationException("The managed client is already started.");
}
if (_connectionCancellationToken != null) throw new InvalidOperationException("The managed client is already started.");
_options = options;
await _storageManager.SetStorageAsync(_options.Storage).ConfigureAwait(false);
@@ -72,10 +69,10 @@ namespace MQTTnet.Core.ManagedClient
_connectionCancellationToken = new CancellationTokenSource();
#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
Task.Factory.StartNew(() => MaintainConnectionAsync(_connectionCancellationToken.Token), _connectionCancellationToken.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default).ConfigureAwait(false);
Task.Run(async () => await MaintainConnectionAsync(_connectionCancellationToken.Token), _connectionCancellationToken.Token).ConfigureAwait(false);
#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
_logger.LogInformation("Started");
_logger.Info<ManagedMqttClient>("Started");
}
public Task StopAsync()
@@ -83,6 +80,9 @@ namespace MQTTnet.Core.ManagedClient
_connectionCancellationToken?.Cancel(false);
_connectionCancellationToken = null;
_publishingCancellationToken?.Cancel(false);
_publishingCancellationToken = null;
while (_messageQueue.Any())
{
_messageQueue.Take();
@@ -159,7 +159,7 @@ namespace MQTTnet.Core.ManagedClient
_publishingCancellationToken = new CancellationTokenSource();
#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
Task.Factory.StartNew(() => PublishQueuedMessagesAsync(_publishingCancellationToken.Token), _publishingCancellationToken.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default).ConfigureAwait(false);
Task.Run(async () => await PublishQueuedMessagesAsync(_publishingCancellationToken.Token), _publishingCancellationToken.Token).ConfigureAwait(false);
#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
continue;
@@ -167,7 +167,7 @@ namespace MQTTnet.Core.ManagedClient
if (connectionState == ReconnectionResult.StillConnected)
{
await Task.Delay(100, _connectionCancellationToken.Token).ConfigureAwait(false); // Consider using the _Disconnected_ event here. (TaskCompletionSource)
await Task.Delay(TimeSpan.FromSeconds(1), _connectionCancellationToken.Token).ConfigureAwait(false);
}
}
}
@@ -176,16 +176,16 @@ namespace MQTTnet.Core.ManagedClient
}
catch (MqttCommunicationException exception)
{
_logger.LogWarning(new EventId(), exception, "Communication exception while maintaining connection.");
_logger.Warning<ManagedMqttClient>(exception, "Communication exception while maintaining connection.");
}
catch (Exception exception)
{
_logger.LogError(new EventId(), exception, "Unhandled exception while maintaining connection.");
_logger.Error<ManagedMqttClient>(exception, "Unhandled exception while maintaining connection.");
}
finally
{
await _mqttClient.DisconnectAsync().ConfigureAwait(false);
_logger.LogInformation("Stopped");
_logger.Info<ManagedMqttClient>("Stopped");
}
}
@@ -215,7 +215,7 @@ namespace MQTTnet.Core.ManagedClient
}
finally
{
_logger.LogInformation("Stopped publishing messages");
_logger.Info<ManagedMqttClient>("Stopped publishing messages");
}
}
@@ -227,7 +227,7 @@ namespace MQTTnet.Core.ManagedClient
}
catch (MqttCommunicationException exception)
{
_logger.LogWarning(new EventId(), exception, "Publishing application message failed.");
_logger.Warning<ManagedMqttClient>(exception, "Publishing application message failed.");
if (message.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
{
@@ -236,13 +236,13 @@ namespace MQTTnet.Core.ManagedClient
}
catch (Exception exception)
{
_logger.LogError(new EventId(), exception, "Unhandled exception while publishing queued application message.");
_logger.Error<ManagedMqttClient>(exception, "Unhandled exception while publishing queued application message.");
}
}
private async Task PushSubscriptionsAsync()
{
_logger.LogInformation(nameof(ManagedMqttClient), "Synchronizing subscriptions");
_logger.Info<ManagedMqttClient>(nameof(ManagedMqttClient), "Synchronizing subscriptions");
List<TopicFilter> subscriptions;
lock (_subscriptions)
@@ -262,7 +262,7 @@ namespace MQTTnet.Core.ManagedClient
}
catch (Exception exception)
{
_logger.LogWarning(new EventId(), exception, "Synchronizing subscriptions failed");
_logger.Warning<ManagedMqttClient>(exception, "Synchronizing subscriptions failed");
_subscriptionsNotPushed = true;
}
}
MQTTnet.Core/Server/IMqttClientRetainedMessageManager.cs
-15
View File
@@ -1,15 +0,0 @@
using System.Collections.Generic;
using System.Threading.Tasks;
using MQTTnet.Core.Packets;
namespace MQTTnet.Core.Server
{
public interface IMqttClientRetainedMessageManager
{
Task LoadMessagesAsync();
Task HandleMessageAsync(string clientId, MqttApplicationMessage applicationMessage);
Task<List<MqttApplicationMessage>> GetSubscribedMessagesAsync(MqttSubscribePacket subscribePacket);
}
}
MQTTnet.Core/Server/IMqttClientSesssionFactory.cs
-7
View File
@@ -1,7 +0,0 @@
namespace MQTTnet.Core.Server
{
public interface IMqttClientSesssionFactory
{
MqttClientSession CreateClientSession(string sessionId, MqttClientSessionsManager mqttClientSessionsManager);
}
}
MQTTnet.Core/Server/IMqttServer.cs
+1 -1
View File
@@ -12,7 +12,7 @@ namespace MQTTnet.Core.Server
Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync();
Task StartAsync();
Task StartAsync(MqttServerOptions options);
Task StopAsync();
}
}
MQTTnet.Core/Server/IMqttServerFactory.cs
+4 -2
View File
@@ -1,4 +1,6 @@
using System;
using System.Collections.Generic;
using MQTTnet.Core.Adapter;
using MQTTnet.Core.Diagnostics;
namespace MQTTnet.Core.Server
{
@@ -6,6 +8,6 @@ namespace MQTTnet.Core.Server
{
IMqttServer CreateMqttServer();
IMqttServer CreateMqttServer(Action<MqttServerOptions> configure);
IMqttServer CreateMqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger);
}
}
MQTTnet.Core/Server/MqttClientPendingMessagesQueue.cs
+12 -12
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,16 +15,16 @@ namespace MQTTnet.Core.Server
private readonly BlockingCollection<MqttPublishPacket> _pendingPublishPackets = new BlockingCollection<MqttPublishPacket>();
private readonly MqttServerOptions _options;
private readonly MqttClientSession _session;
private readonly ILogger<MqttClientPendingMessagesQueue> _logger;
private readonly IMqttNetLogger _logger;
public MqttClientPendingMessagesQueue(MqttServerOptions options, MqttClientSession session, ILogger<MqttClientPendingMessagesQueue> logger)
public MqttClientPendingMessagesQueue(MqttServerOptions options, MqttClientSession session, IMqttNetLogger logger)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_session = session ?? throw new ArgumentNullException(nameof(session));
_options = options ?? throw new ArgumentNullException(nameof(options));
}
public void Start(IMqttCommunicationAdapter adapter, CancellationToken cancellationToken)
public void Start(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
{
if (adapter == null) throw new ArgumentNullException(nameof(adapter));
@@ -41,10 +41,10 @@ namespace MQTTnet.Core.Server
if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
_pendingPublishPackets.Add(publishPacket);
_logger.LogTrace("Enqueued packet (ClientId: {0}).", _session.ClientId);
_logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet (ClientId: {0}).", _session.ClientId);
}
private async Task SendPendingPublishPacketsAsync(IMqttCommunicationAdapter adapter, CancellationToken cancellationToken)
private async Task SendPendingPublishPacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
{
try
{
@@ -58,11 +58,11 @@ namespace MQTTnet.Core.Server
}
catch (Exception exception)
{
_logger.LogError(new EventId(), exception, "Unhandled exception while sending enqueued packet (ClientId: {0}).", _session.ClientId);
_logger.Error<MqttClientPendingMessagesQueue>(exception, "Unhandled exception while sending enqueued packet (ClientId: {0}).", _session.ClientId);
}
}
private async Task SendPendingPublishPacketAsync(IMqttCommunicationAdapter adapter, CancellationToken cancellationToken)
private async Task SendPendingPublishPacketAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
{
MqttPublishPacket packet = null;
try
@@ -70,24 +70,24 @@ namespace MQTTnet.Core.Server
packet = _pendingPublishPackets.Take(cancellationToken);
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, packet).ConfigureAwait(false);
_logger.LogTrace("Enqueued packet sent (ClientId: {0}).", _session.ClientId);
_logger.Trace<MqttClientPendingMessagesQueue>("Enqueued packet sent (ClientId: {0}).", _session.ClientId);
}
catch (Exception exception)
{
if (exception is MqttCommunicationTimedOutException)
{
_logger.LogWarning(new EventId(), exception, "Sending publish packet failed due to timeout (ClientId: {0}).", _session.ClientId);
_logger.Warning<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed due to timeout (ClientId: {0}).", _session.ClientId);
}
else if (exception is MqttCommunicationException)
{
_logger.LogWarning(new EventId(), exception, "Sending publish packet failed due to communication exception (ClientId: {0}).", _session.ClientId);
_logger.Warning<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed due to communication exception (ClientId: {0}).", _session.ClientId);
}
else if (exception is OperationCanceledException)
{
}
else
{
_logger.LogError(new EventId(), exception, "Sending publish packet failed (ClientId: {0}).", _session.ClientId);
_logger.Error<MqttClientPendingMessagesQueue>(exception, "Sending publish packet failed (ClientId: {0}).", _session.ClientId);
}
if (packet != null && packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
MQTTnet.Core/Server/MqttClientSession.cs
+21 -23
View File
@@ -3,13 +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;
using Microsoft.Extensions.Options;
namespace MQTTnet.Core.Server
{
@@ -21,28 +20,27 @@ namespace MQTTnet.Core.Server
private readonly MqttClientSessionsManager _sessionsManager;
private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue;
private readonly MqttServerOptions _options;
private readonly ILogger<MqttClientSession> _logger;
private readonly IMqttNetLogger _logger;
private IMqttCommunicationAdapter _adapter;
private IMqttChannelAdapter _adapter;
private CancellationTokenSource _cancellationTokenSource;
private MqttApplicationMessage _willMessage;
public MqttClientSession(
string clientId,
IOptions<MqttServerOptions> options,
MqttServerOptions options,
MqttClientSessionsManager sessionsManager,
MqttClientSubscriptionsManager subscriptionsManager,
ILogger<MqttClientSession> logger,
ILogger<MqttClientPendingMessagesQueue> messageQueueLogger)
IMqttNetLogger logger)
{
_sessionsManager = sessionsManager ?? throw new ArgumentNullException(nameof(sessionsManager));
_subscriptionsManager = subscriptionsManager ?? throw new ArgumentNullException(nameof(subscriptionsManager));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
ClientId = clientId;
_options = options.Value;
_pendingMessagesQueue = new MqttClientPendingMessagesQueue(_options, this, messageQueueLogger);
_options = options;
_subscriptionsManager = new MqttClientSubscriptionsManager(_options);
_pendingMessagesQueue = new MqttClientPendingMessagesQueue(_options, this, _logger);
}
public string ClientId { get; }
@@ -51,7 +49,7 @@ namespace MQTTnet.Core.Server
public bool IsConnected => _adapter != null;
public async Task RunAsync(MqttApplicationMessage willMessage, IMqttCommunicationAdapter adapter)
public async Task RunAsync(MqttApplicationMessage willMessage, IMqttChannelAdapter adapter)
{
if (adapter == null) throw new ArgumentNullException(nameof(adapter));
@@ -71,11 +69,11 @@ namespace MQTTnet.Core.Server
}
catch (MqttCommunicationException exception)
{
_logger.LogWarning(new EventId(), exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
_logger.Warning<MqttClientSession>(exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
}
catch (Exception exception)
{
_logger.LogError(new EventId(), exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
_logger.Error<MqttClientSession>(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
}
}
@@ -93,7 +91,7 @@ namespace MQTTnet.Core.Server
_adapter = null;
}
_logger.LogInformation("Client '{0}': Session stopped.", ClientId);
_logger.Info<MqttClientSession>("Client '{0}': Session stopped.", ClientId);
}
finally
{
@@ -120,7 +118,7 @@ namespace MQTTnet.Core.Server
_pendingMessagesQueue.Enqueue(publishPacket);
}
private async Task ReceivePacketsAsync(IMqttCommunicationAdapter adapter, CancellationToken cancellationToken)
private async Task ReceivePacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken)
{
try
{
@@ -135,17 +133,17 @@ namespace MQTTnet.Core.Server
}
catch (MqttCommunicationException exception)
{
_logger.LogWarning(new EventId(), exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
_logger.Warning<MqttClientSession>(exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
await StopAsync();
}
catch (Exception exception)
{
_logger.LogError(new EventId(), exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
_logger.Error<MqttClientSession>(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
await StopAsync();
}
}
private Task ProcessReceivedPacketAsync(IMqttCommunicationAdapter adapter, MqttBasePacket packet, CancellationToken cancellationToken)
private Task ProcessReceivedPacketAsync(IMqttChannelAdapter adapter, MqttBasePacket packet, CancellationToken cancellationToken)
{
if (packet is MqttPingReqPacket)
{
@@ -188,11 +186,11 @@ namespace MQTTnet.Core.Server
return StopAsync();
}
_logger.LogWarning("Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet);
_logger.Warning<MqttClientSession>("Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet);
return StopAsync();
}
private async Task HandleIncomingSubscribePacketAsync(IMqttCommunicationAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
{
var subscribeResult = _subscriptionsManager.Subscribe(subscribePacket, ClientId);
@@ -215,7 +213,7 @@ namespace MQTTnet.Core.Server
}
}
private async Task HandleIncomingPublishPacketAsync(IMqttCommunicationAdapter adapter, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
private async Task HandleIncomingPublishPacketAsync(IMqttChannelAdapter adapter, MqttPublishPacket publishPacket, CancellationToken cancellationToken)
{
var applicationMessage = publishPacket.ToApplicationMessage();
@@ -255,7 +253,7 @@ namespace MQTTnet.Core.Server
}
}
private Task HandleIncomingPubRelPacketAsync(IMqttCommunicationAdapter adapter, MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
private Task HandleIncomingPubRelPacketAsync(IMqttChannelAdapter adapter, MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
{
lock (_unacknowledgedPublishPackets)
{
MQTTnet.Core/Server/MqttClientSessionsManager.cs
+16 -26
View File
@@ -4,13 +4,12 @@ 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
{
@@ -20,27 +19,21 @@ namespace MQTTnet.Core.Server
private readonly SemaphoreSlim _sessionsSemaphore = new SemaphoreSlim(1, 1);
private readonly MqttServerOptions _options;
private readonly ILogger<MqttClientSessionsManager> _logger;
private readonly IMqttClientSesssionFactory _clientSesssionFactory;
private readonly IMqttClientRetainedMessageManager _clientRetainedMessageManager;
private readonly MqttRetainedMessagesManager _retainedMessagesManager;
private readonly IMqttNetLogger _logger;
public MqttClientSessionsManager(
IOptions<MqttServerOptions> options,
ILogger<MqttClientSessionsManager> logger,
IMqttClientSesssionFactory clientSesssionFactory,
IMqttClientRetainedMessageManager clientRetainedMessageManager)
public MqttClientSessionsManager(MqttServerOptions options, MqttRetainedMessagesManager retainedMessagesManager, IMqttNetLogger logger)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_options = options.Value ?? throw new ArgumentNullException(nameof(options));
_clientSesssionFactory = clientSesssionFactory ?? throw new ArgumentNullException(nameof(clientSesssionFactory));
_clientRetainedMessageManager = clientRetainedMessageManager ?? throw new ArgumentNullException(nameof(clientRetainedMessageManager));
_options = options ?? throw new ArgumentNullException(nameof(options));
_retainedMessagesManager = retainedMessagesManager ?? throw new ArgumentNullException(nameof(retainedMessagesManager));
}
public event EventHandler<MqttClientConnectedEventArgs> ClientConnected;
public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public async Task RunClientSessionAsync(IMqttCommunicationAdapter clientAdapter, CancellationToken cancellationToken)
public async Task RunClientSessionAsync(IMqttChannelAdapter clientAdapter, CancellationToken cancellationToken)
{
var clientId = string.Empty;
try
@@ -80,14 +73,11 @@ namespace MQTTnet.Core.Server
ProtocolVersion = clientAdapter.PacketSerializer.ProtocolVersion
}));
using (_logger.BeginScope(clientId))
{
await clientSession.Session.RunAsync(connectPacket.WillMessage, clientAdapter).ConfigureAwait(false);
}
await clientSession.Session.RunAsync(connectPacket.WillMessage, clientAdapter).ConfigureAwait(false);
}
catch (Exception exception)
{
_logger.LogError(new EventId(), exception, exception.Message);
_logger.Error<MqttClientSessionsManager>(exception, exception.Message);
}
finally
{
@@ -157,7 +147,7 @@ namespace MQTTnet.Core.Server
if (applicationMessage.Retain)
{
await _clientRetainedMessageManager.HandleMessageAsync(senderClientSession?.ClientId, applicationMessage).ConfigureAwait(false);
await _retainedMessagesManager.HandleMessageAsync(senderClientSession?.ClientId, applicationMessage).ConfigureAwait(false);
}
var eventArgs = new MqttApplicationMessageReceivedEventArgs(senderClientSession?.ClientId, applicationMessage);
@@ -165,7 +155,7 @@ namespace MQTTnet.Core.Server
}
catch (Exception exception)
{
_logger.LogError(new EventId(), exception, "Error while processing application message");
_logger.Error<MqttClientSessionsManager>(exception, "Error while processing application message");
}
lock (_sessions)
@@ -179,7 +169,7 @@ namespace MQTTnet.Core.Server
public Task<List<MqttApplicationMessage>> GetRetainedMessagesAsync(MqttSubscribePacket subscribePacket)
{
return _clientRetainedMessageManager.GetSubscribedMessagesAsync(subscribePacket);
return _retainedMessagesManager.GetSubscribedMessagesAsync(subscribePacket);
}
private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket)
@@ -206,11 +196,11 @@ namespace MQTTnet.Core.Server
await clientSession.StopAsync();
clientSession = null;
_logger.LogTrace("Stopped existing session of client '{0}'.", connectPacket.ClientId);
_logger.Trace<MqttClientSessionsManager>("Stopped existing session of client '{0}'.", connectPacket.ClientId);
}
else
{
_logger.LogTrace("Reusing existing session of client '{0}'.", connectPacket.ClientId);
_logger.Trace<MqttClientSessionsManager>("Reusing existing session of client '{0}'.", connectPacket.ClientId);
}
}
@@ -219,10 +209,10 @@ namespace MQTTnet.Core.Server
{
isExistingSession = false;
clientSession = _clientSesssionFactory.CreateClientSession(connectPacket.ClientId, this);
clientSession = new MqttClientSession(connectPacket.ClientId, _options, this, _logger);
_sessions[connectPacket.ClientId] = clientSession;
_logger.LogTrace("Created a new session for client '{0}'.", connectPacket.ClientId);
_logger.Trace<MqttClientSessionsManager>("Created a new session for client '{0}'.", connectPacket.ClientId);
}
return new GetOrCreateClientSessionResult { IsExistingSession = isExistingSession, Session = clientSession };
MQTTnet.Core/Server/MqttClientSubscriptionsManager.cs
+2 -3
View File
@@ -1,6 +1,5 @@
using System;
using System.Collections.Generic;
using Microsoft.Extensions.Options;
using MQTTnet.Core.Packets;
using MQTTnet.Core.Protocol;
@@ -11,9 +10,9 @@ namespace MQTTnet.Core.Server
private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
private readonly MqttServerOptions _options;
public MqttClientSubscriptionsManager(IOptions<MqttServerOptions> options)
public MqttClientSubscriptionsManager(MqttServerOptions options)
{
_options = options?.Value ?? throw new ArgumentNullException(nameof(options));
_options = options ?? throw new ArgumentNullException(nameof(options));
}
public MqttClientSubscribeResult Subscribe(MqttSubscribePacket subscribePacket, string clientId)
MQTTnet.Core/Server/MqttClientRetainedMessagesManager.cs → MQTTnet.Core/Server/MqttRetainedMessagesManager.cs
+10 -11
View File
@@ -4,22 +4,21 @@ using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Core.Packets;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using MQTTnet.Core.Diagnostics;
namespace MQTTnet.Core.Server
{
public sealed class MqttClientRetainedMessagesManager : IMqttClientRetainedMessageManager
public sealed class MqttRetainedMessagesManager
{
private readonly Dictionary<string, MqttApplicationMessage> _retainedMessages = new Dictionary<string, MqttApplicationMessage>();
private readonly SemaphoreSlim _gate = new SemaphoreSlim(1, 1);
private readonly ILogger<MqttClientRetainedMessagesManager> _logger;
private readonly IMqttNetLogger _logger;
private readonly MqttServerOptions _options;
public MqttClientRetainedMessagesManager(IOptions<MqttServerOptions> options, ILogger<MqttClientRetainedMessagesManager> logger)
public MqttRetainedMessagesManager(MqttServerOptions options, IMqttNetLogger logger)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_options = options.Value ?? throw new ArgumentNullException(nameof(options));
_options = options ?? throw new ArgumentNullException(nameof(options));
}
public async Task LoadMessagesAsync()
@@ -42,7 +41,7 @@ namespace MQTTnet.Core.Server
}
catch (Exception exception)
{
_logger.LogError(new EventId(), exception, "Unhandled exception while loading retained messages.");
_logger.Error<MqttRetainedMessagesManager>(exception, "Unhandled exception while loading retained messages.");
}
finally
{
@@ -61,7 +60,7 @@ namespace MQTTnet.Core.Server
}
catch (Exception exception)
{
_logger.LogError(new EventId(), exception, "Unhandled exception while handling retained messages.");
_logger.Error<MqttRetainedMessagesManager>(exception, "Unhandled exception while handling retained messages.");
}
finally
{
@@ -110,7 +109,7 @@ namespace MQTTnet.Core.Server
if (applicationMessage.Payload?.Any() == false)
{
saveIsRequired = _retainedMessages.Remove(applicationMessage.Topic);
_logger.LogInformation("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);
_logger.Info<MqttRetainedMessagesManager>("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);
}
else
{
@@ -129,12 +128,12 @@ namespace MQTTnet.Core.Server
}
}
_logger.LogInformation("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic);
_logger.Info<MqttRetainedMessagesManager>("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic);
}
if (!saveIsRequired)
{
_logger.LogTrace("Skipped saving retained messages because no changes were detected.");
_logger.Trace<MqttRetainedMessagesManager>("Skipped saving retained messages because no changes were detected.");
}
if (saveIsRequired && _options.Storage != null)
MQTTnet.Core/Server/MqttServer.cs
+62 -43
View File
@@ -3,33 +3,24 @@ using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Core.Adapter;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using System.Linq;
using MQTTnet.Core.Diagnostics;
namespace MQTTnet.Core.Server
{
public class MqttServer : IMqttServer
{
private readonly IMqttClientRetainedMessageManager _clientRetainedMessageManager;
private readonly ILogger<MqttServer> _logger;
private readonly MqttClientSessionsManager _clientSessionsManager;
private readonly ICollection<IMqttServerAdapter> _adapters;
private readonly MqttServerOptions _options;
private readonly IMqttNetLogger _logger;
private MqttClientSessionsManager _clientSessionsManager;
private MqttRetainedMessagesManager _retainedMessagesManager;
private CancellationTokenSource _cancellationTokenSource;
private MqttServerOptions _options;
public MqttServer(
IOptions<MqttServerOptions> options,
IEnumerable<IMqttServerAdapter> adapters,
ILogger<MqttServer> logger,
MqttClientSessionsManager clientSessionsManager,
IMqttClientRetainedMessageManager clientRetainedMessageManager)
public MqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger)
{
_options = options.Value ?? throw new ArgumentNullException(nameof(options));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_clientSessionsManager = clientSessionsManager ?? throw new ArgumentNullException(nameof(clientSessionsManager));
_clientRetainedMessageManager = clientRetainedMessageManager ?? throw new ArgumentNullException(nameof(clientRetainedMessageManager));
if (adapters == null)
{
@@ -37,15 +28,6 @@ namespace MQTTnet.Core.Server
}
_adapters = adapters.ToList();
_clientSessionsManager.ApplicationMessageReceived += (s, e) => ApplicationMessageReceived?.Invoke(s, e);
_clientSessionsManager.ClientConnected += OnClientConnected;
_clientSessionsManager.ClientDisconnected += OnClientDisconnected;
}
public Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync()
{
return _clientSessionsManager.GetConnectedClientsAsync();
}
public event EventHandler<MqttServerStartedEventArgs> Started;
@@ -53,14 +35,16 @@ namespace MQTTnet.Core.Server
public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected;
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public Task<IList<ConnectedMqttClient>> GetConnectedClientsAsync()
{
return _clientSessionsManager.GetConnectedClientsAsync();
}
public async Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
{
if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages));
if (_cancellationTokenSource == null)
{
throw new InvalidOperationException("The server is not started.");
}
if (_cancellationTokenSource == null) throw new InvalidOperationException("The server is not started.");
foreach (var applicationMessage in applicationMessages)
{
@@ -68,13 +52,21 @@ namespace MQTTnet.Core.Server
}
}
public async Task StartAsync()
public async Task StartAsync(MqttServerOptions options)
{
if (_cancellationTokenSource != null) throw new InvalidOperationException("The MQTT server is already started.");
_options = options ?? throw new ArgumentNullException(nameof(options));
if (_cancellationTokenSource != null) throw new InvalidOperationException("The server is already started.");
_cancellationTokenSource = new CancellationTokenSource();
_retainedMessagesManager = new MqttRetainedMessagesManager(_options, _logger);
await _clientRetainedMessageManager.LoadMessagesAsync();
_clientSessionsManager = new MqttClientSessionsManager(_options, _retainedMessagesManager, _logger);
_clientSessionsManager.ApplicationMessageReceived += OnApplicationMessageReceived;
_clientSessionsManager.ClientConnected += OnClientConnected;
_clientSessionsManager.ClientDisconnected += OnClientDisconnected;
await _retainedMessagesManager.LoadMessagesAsync();
foreach (var adapter in _adapters)
{
@@ -82,26 +74,48 @@ namespace MQTTnet.Core.Server
await adapter.StartAsync(_options);
}
_logger.LogInformation("Started.");
_logger.Info<MqttServer>("Started.");
Started?.Invoke(this, new MqttServerStartedEventArgs());
}
public async Task StopAsync()
{
_cancellationTokenSource?.Cancel(false);
_cancellationTokenSource?.Dispose();
_cancellationTokenSource = null;
foreach (var adapter in _adapters)
try
{
adapter.ClientAccepted -= OnClientAccepted;
await adapter.StopAsync();
if (_cancellationTokenSource == null)
{
return;
}
_cancellationTokenSource.Cancel(false);
_cancellationTokenSource.Dispose();
foreach (var adapter in _adapters)
{
adapter.ClientAccepted -= OnClientAccepted;
await adapter.StopAsync();
}
await _clientSessionsManager.StopAsync();
_logger.Info<MqttServer>("Stopped.");
}
finally
{
_cancellationTokenSource = null;
await _clientSessionsManager.StopAsync();
_retainedMessagesManager = null;
_logger.LogInformation("Stopped.");
if (_clientSessionsManager != null)
{
_clientSessionsManager.ApplicationMessageReceived -= OnApplicationMessageReceived;
_clientSessionsManager.ClientConnected -= OnClientConnected;
_clientSessionsManager.ClientDisconnected -= OnClientDisconnected;
}
_clientSessionsManager = null;
}
}
private void OnClientAccepted(object sender, MqttServerAdapterClientAcceptedEventArgs eventArgs)
@@ -111,14 +125,19 @@ namespace MQTTnet.Core.Server
private void OnClientConnected(object sender, MqttClientConnectedEventArgs eventArgs)
{
_logger.LogInformation("Client '{0}': Connected.", eventArgs.Client.ClientId);
_logger.Info<MqttServer>("Client '{0}': Connected.", eventArgs.Client.ClientId);
ClientConnected?.Invoke(this, eventArgs);
}
private void OnClientDisconnected(object sender, MqttClientDisconnectedEventArgs eventArgs)
{
_logger.LogInformation("Client '{0}': Disconnected.", eventArgs.Client.ClientId);
_logger.Info<MqttServer>("Client '{0}': Disconnected.", eventArgs.Client.ClientId);
ClientDisconnected?.Invoke(this, eventArgs);
}
private void OnApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e)
{
ApplicationMessageReceived?.Invoke(this, e);
}
}
}