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

Merge from develop.

Browse Source
This commit is contained in:
Christian Kratky
2018-11-16 22:16:21 +01:00
parent 5a59cd278c fcff3a2c8d
commit fed16d7710
50 changed files with 845 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet.AspnetCore/AspNetMqttServerOptionsBuilder.cs
+15
View File
@@ -0,0 +1,15 @@
using MQTTnet.Server;
using System;
namespace MQTTnet.AspNetCore
{
public class AspNetMqttServerOptionsBuilder : MqttServerOptionsBuilder
{
public AspNetMqttServerOptionsBuilder(IServiceProvider serviceProvider)
{
ServiceProvider = serviceProvider;
}
public IServiceProvider ServiceProvider { get; }
}
}
Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj
+3 -3
View File
@@ -14,9 +14,9 @@
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.AspNetCore.Connections.Abstractions" Version="2.1.0" />
<PackageReference Include="Microsoft.AspNetCore.WebSockets" Version="2.0.1" />
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="2.0.1" />
<PackageReference Include="Microsoft.AspNetCore.Connections.Abstractions" Version="2.1.3" />
<PackageReference Include="Microsoft.AspNetCore.WebSockets" Version="2.1.1" />
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="2.1.1" />
</ItemGroup>
<ItemGroup>
Source/MQTTnet.AspnetCore/MqttConnectionContext.cs
+18 -5
View File
@@ -1,6 +1,7 @@
using Microsoft.AspNetCore.Connections;
using MQTTnet.Adapter;
using MQTTnet.AspNetCore.Client.Tcp;
using MQTTnet.Exceptions;
using MQTTnet.Packets;
using MQTTnet.Serializer;
using System;
@@ -24,6 +25,8 @@ namespace MQTTnet.AspNetCore
public event EventHandler ReadingPacketStarted;
public event EventHandler ReadingPacketCompleted;
private readonly SemaphoreSlim _writerSemaphore = new SemaphoreSlim(1, 1);
public Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
{
if (Connection is TcpConnection tcp && !tcp.IsConnected)
@@ -57,7 +60,7 @@ namespace MQTTnet.AspNetCore
}
else
{
readResult = await readTask;
readResult = await readTask.ConfigureAwait(false);
}
var buffer = readResult.Buffer;
@@ -81,7 +84,7 @@ namespace MQTTnet.AspNetCore
}
else if (readResult.IsCompleted)
{
break;
throw new MqttCommunicationException("Connection Aborted");
}
}
finally
@@ -102,10 +105,20 @@ namespace MQTTnet.AspNetCore
return null;
}
public Task SendPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken)
public async Task SendPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken)
{
var buffer = PacketSerializerAdapter.Serialize(packet);
return Connection.Transport.Output.WriteAsync(buffer.AsMemory(), cancellationToken).AsTask();
var buffer = PacketSerializer.Serialize(packet).AsMemory();
var output = Connection.Transport.Output;
await _writerSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
await output.WriteAsync(buffer, cancellationToken).ConfigureAwait(false);
}
finally
{
_writerSemaphore.Release();
}
}
public void Dispose()
Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs
+18 -5
View File
@@ -20,14 +20,27 @@ namespace MQTTnet.AspNetCore
return services;
}
public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, Action<MqttServerOptionsBuilder> configure)
{
var builder = new MqttServerOptionsBuilder();
services.AddSingleton<IMqttServerOptions>(s => {
var builder = new MqttServerOptionsBuilder();
configure(builder);
return builder.Build();
});
configure(builder);
services.AddSingleton<IMqttServerOptions>(builder.Build());
services.AddHostedMqttServer();
return services;
}
public static IServiceCollection AddHostedMqttServerWithServices(this IServiceCollection services, Action<AspNetMqttServerOptionsBuilder> configure)
{
services.AddSingleton<IMqttServerOptions>(s => {
var builder = new AspNetMqttServerOptionsBuilder(s);
configure(builder);
return builder.Build();
});
services.AddHostedMqttServer();
Source/MQTTnet.Extensions.ManagedClient/ApplicationMessageSkippedEventArgs.cs
+14
View File
@@ -0,0 +1,14 @@
using System;
namespace MQTTnet.Extensions.ManagedClient
{
public class ApplicationMessageSkippedEventArgs : EventArgs
{
public ApplicationMessageSkippedEventArgs(ManagedMqttApplicationMessage applicationMessage)
{
ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage));
}
public ManagedMqttApplicationMessage ApplicationMessage { get; }
}
}
Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs
+1
View File
@@ -15,6 +15,7 @@ namespace MQTTnet.Extensions.ManagedClient
event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
event EventHandler<ApplicationMessageProcessedEventArgs> ApplicationMessageProcessed;
event EventHandler<ApplicationMessageSkippedEventArgs> ApplicationMessageSkipped;
event EventHandler<MqttManagedProcessFailedEventArgs> ConnectingFailed;
event EventHandler<MqttManagedProcessFailedEventArgs> SynchronizingSubscriptionsFailed;
Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClientOptions.cs
+5
View File
@@ -1,5 +1,6 @@
using System;
using MQTTnet.Client;
using MQTTnet.Server;
namespace MQTTnet.Extensions.ManagedClient
{
@@ -12,5 +13,9 @@ namespace MQTTnet.Extensions.ManagedClient
TimeSpan ConnectionCheckInterval { get; }
IManagedMqttClientStorage Storage { get; }
int MaxPendingMessages { get; }
MqttPendingMessagesOverflowStrategy PendingMessagesOverflowStrategy { get; }
}
}
Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs
+40 -15
View File
@@ -1,5 +1,4 @@
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
@@ -7,13 +6,17 @@ using System.Threading.Tasks;
using MQTTnet.Client;
using MQTTnet.Diagnostics;
using MQTTnet.Exceptions;
using MQTTnet.Internal;
using MQTTnet.Protocol;
using MQTTnet.Server;
using MqttClientConnectedEventArgs = MQTTnet.Client.MqttClientConnectedEventArgs;
using MqttClientDisconnectedEventArgs = MQTTnet.Client.MqttClientDisconnectedEventArgs;
namespace MQTTnet.Extensions.ManagedClient
{
public class ManagedMqttClient : IManagedMqttClient
{
private readonly BlockingCollection<ManagedMqttApplicationMessage> _messageQueue = new BlockingCollection<ManagedMqttApplicationMessage>();
private readonly BlockingQueue<ManagedMqttApplicationMessage> _messageQueue = new BlockingQueue<ManagedMqttApplicationMessage>();
private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
private readonly HashSet<string> _unsubscriptions = new HashSet<string>();
@@ -50,6 +53,7 @@ namespace MQTTnet.Extensions.ManagedClient
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public event EventHandler<ApplicationMessageProcessedEventArgs> ApplicationMessageProcessed;
public event EventHandler<ApplicationMessageSkippedEventArgs> ApplicationMessageSkipped;
public event EventHandler<MqttManagedProcessFailedEventArgs> ConnectingFailed;
public event EventHandler<MqttManagedProcessFailedEventArgs> SynchronizingSubscriptionsFailed;
@@ -75,7 +79,7 @@ namespace MQTTnet.Extensions.ManagedClient
foreach (var message in messages)
{
_messageQueue.Add(message);
_messageQueue.Enqueue(message);
}
}
@@ -93,10 +97,7 @@ namespace MQTTnet.Extensions.ManagedClient
StopPublishing();
StopMaintainingConnection();
while (_messageQueue.Any())
{
_messageQueue.Take();
}
_messageQueue.Clear();
return Task.FromResult(0);
}
@@ -112,12 +113,38 @@ namespace MQTTnet.Extensions.ManagedClient
{
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
if (_storageManager != null)
ManagedMqttApplicationMessage removedMessage = null;
lock (_messageQueue)
{
await _storageManager.AddAsync(applicationMessage).ConfigureAwait(false);
if (_messageQueue.Count >= _options.MaxPendingMessages)
{
if (_options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage)
{
_logger.Verbose("Skipping publish of new application message because internal queue is full.");
ApplicationMessageSkipped?.Invoke(this, new ApplicationMessageSkippedEventArgs(applicationMessage));
return;
}
if (_options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage)
{
removedMessage = _messageQueue.RemoveFirst();
_logger.Verbose("Removed oldest application message from internal queue because it is full.");
ApplicationMessageSkipped?.Invoke(this, new ApplicationMessageSkippedEventArgs(removedMessage));
}
}
_messageQueue.Enqueue(applicationMessage);
}
_messageQueue.Add(applicationMessage);
if (_storageManager != null)
{
if (removedMessage != null)
{
await _storageManager.RemoveAsync(removedMessage).ConfigureAwait(false);
}
await _storageManager.AddAsync(applicationMessage).ConfigureAwait(false);
}
}
public Task SubscribeAsync(IEnumerable<TopicFilter> topicFilters)
@@ -157,7 +184,6 @@ namespace MQTTnet.Extensions.ManagedClient
public void Dispose()
{
_messageQueue?.Dispose();
_connectionCancellationToken?.Dispose();
_publishingCancellationToken?.Dispose();
}
@@ -226,9 +252,9 @@ namespace MQTTnet.Extensions.ManagedClient
{
try
{
while (!cancellationToken.IsCancellationRequested)
while (!cancellationToken.IsCancellationRequested && _mqttClient.IsConnected)
{
var message = _messageQueue.Take(cancellationToken);
var message = _messageQueue.Dequeue();
if (message == null)
{
continue;
@@ -268,7 +294,7 @@ namespace MQTTnet.Extensions.ManagedClient
if (message.ApplicationMessage.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
{
_messageQueue.Add(message);
_messageQueue.Enqueue(message);
}
}
catch (Exception exception)
@@ -367,7 +393,6 @@ namespace MQTTnet.Extensions.ManagedClient
}
var cts = new CancellationTokenSource();
_publishingCancellationToken = cts;
Task.Factory.StartNew(() => PublishQueuedMessages(cts.Token), cts.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptions.cs
+5
View File
@@ -1,5 +1,6 @@
using System;
using MQTTnet.Client;
using MQTTnet.Server;
namespace MQTTnet.Extensions.ManagedClient
{
@@ -12,5 +13,9 @@ namespace MQTTnet.Extensions.ManagedClient
public TimeSpan ConnectionCheckInterval { get; set; } = TimeSpan.FromSeconds(1);
public IManagedMqttClientStorage Storage { get; set; }
public int MaxPendingMessages { get; set; } = int.MaxValue;
public MqttPendingMessagesOverflowStrategy PendingMessagesOverflowStrategy { get; set; } = MqttPendingMessagesOverflowStrategy.DropNewMessage;
}
}
Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientOptionsBuilder.cs
+13
View File
@@ -1,5 +1,6 @@
using System;
using MQTTnet.Client;
using MQTTnet.Server;
namespace MQTTnet.Extensions.ManagedClient
{
@@ -8,6 +9,18 @@ namespace MQTTnet.Extensions.ManagedClient
private readonly ManagedMqttClientOptions _options = new ManagedMqttClientOptions();
private MqttClientOptionsBuilder _clientOptionsBuilder;
public ManagedMqttClientOptionsBuilder WithMaxPendingMessages(int value)
{
_options.MaxPendingMessages = value;
return this;
}
public ManagedMqttClientOptionsBuilder WithPendingMessagesOverflowStrategy(MqttPendingMessagesOverflowStrategy value)
{
_options.PendingMessagesOverflowStrategy = value;
return this;
}
public ManagedMqttClientOptionsBuilder WithAutoReconnectDelay(TimeSpan value)
{
_options.AutoReconnectDelay = value;
Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClientStorageManager.cs
+4
View File
@@ -28,6 +28,8 @@ namespace MQTTnet.Extensions.ManagedClient
public async Task AddAsync(ManagedMqttApplicationMessage applicationMessage)
{
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
using (await _messagesLock.LockAsync(CancellationToken.None).ConfigureAwait(false))
{
_messages.Add(applicationMessage);
@@ -37,6 +39,8 @@ namespace MQTTnet.Extensions.ManagedClient
public async Task RemoveAsync(ManagedMqttApplicationMessage applicationMessage)
{
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
using (await _messagesLock.LockAsync(CancellationToken.None).ConfigureAwait(false))
{
var index = _messages.IndexOf(applicationMessage);
Source/MQTTnet/Adapter/MqttChannelAdapter.cs
+5 -1
View File
@@ -154,6 +154,9 @@ namespace MQTTnet.Adapter
return packet;
}
catch (OperationCanceledException)
{
}
catch (Exception exception)
{
if (IsWrappedException(exception))
@@ -241,7 +244,8 @@ namespace MQTTnet.Adapter
{
if (exception is IOException && exception.InnerException is SocketException socketException)
{
if (socketException.SocketErrorCode == SocketError.ConnectionAborted)
if (socketException.SocketErrorCode == SocketError.ConnectionAborted ||
socketException.SocketErrorCode == SocketError.OperationAborted)
{
throw new OperationCanceledException();
}
Source/MQTTnet/Client/IMqttClientFactory.cs
+4
View File
@@ -7,5 +7,9 @@ namespace MQTTnet.Client
IMqttClient CreateMqttClient();
IMqttClient CreateMqttClient(IMqttNetLogger logger);
IMqttClient CreateMqttClient(IMqttClientAdapterFactory adapterFactory);
IMqttClient CreateMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory adapterFactory);
}
}
Source/MQTTnet/Client/MqttClient.cs
+38 -25
View File
@@ -30,7 +30,7 @@ namespace MQTTnet.Client
internal Task _keepAliveMessageSenderTask;
private IMqttChannelAdapter _adapter;
private bool _cleanDisconnectInitiated;
private TaskCompletionSource<bool> _disconnectReason;
private long _disconnectGate;
public MqttClient(IMqttClientAdapterFactory channelFactory, IMqttNetLogger logger)
{
@@ -55,12 +55,12 @@ namespace MQTTnet.Client
try
{
_cancellationTokenSource = new CancellationTokenSource();
_disconnectReason = new TaskCompletionSource<bool>();
_options = options;
_packetIdentifierProvider.Reset();
_packetDispatcher.Reset();
_cancellationTokenSource = new CancellationTokenSource();
_disconnectGate = 0;
_adapter = _adapterFactory.CreateClientAdapter(options, _logger);
_logger.Verbose($"Trying to connect with server ({_options.ChannelOptions}).");
@@ -88,10 +88,12 @@ namespace MQTTnet.Client
catch (Exception exception)
{
_logger.Error(exception, "Error while connecting with server.");
if (_disconnectReason.TrySetException(exception))
if (!DisconnectIsPending())
{
await DisconnectInternalAsync(null, exception).ConfigureAwait(false);
}
throw;
}
}
@@ -109,7 +111,7 @@ namespace MQTTnet.Client
}
finally
{
if (_disconnectReason.TrySetCanceled())
if (!DisconnectIsPending())
{
await DisconnectInternalAsync(null, null).ConfigureAwait(false);
}
@@ -184,10 +186,12 @@ namespace MQTTnet.Client
public void Dispose()
{
_cancellationTokenSource?.Cancel(false);
_cancellationTokenSource?.Dispose();
_cancellationTokenSource = null;
_adapter?.Dispose();
_adapter = null;
}
private async Task<MqttV3ConnAckPacket> AuthenticateAsync(MqttApplicationMessage willApplicationMessage, CancellationToken cancellationToken)
@@ -213,7 +217,7 @@ namespace MQTTnet.Client
private void ThrowIfNotConnected()
{
if (!IsConnected) throw new MqttCommunicationException("The client is not connected.");
if (!IsConnected || Interlocked.Read(ref _disconnectGate) == 1) throw new MqttCommunicationException("The client is not connected.");
}
private void ThrowIfConnected(string message)
@@ -223,9 +227,10 @@ namespace MQTTnet.Client
private async Task DisconnectInternalAsync(Task sender, Exception exception)
{
InitiateDisconnect();
var clientWasConnected = IsConnected;
InitiateDisconnect();
IsConnected = false;
try
@@ -246,10 +251,7 @@ namespace MQTTnet.Client
}
finally
{
_adapter?.Dispose();
_adapter = null;
_cancellationTokenSource?.Dispose();
_cancellationTokenSource = null;
Dispose();
_cleanDisconnectInitiated = false;
_logger.Info("Disconnected.");
@@ -263,16 +265,16 @@ namespace MQTTnet.Client
{
try
{
if (_cancellationTokenSource == null || _cancellationTokenSource.IsCancellationRequested)
if (_cancellationTokenSource?.IsCancellationRequested == true)
{
return;
}
_cancellationTokenSource.Cancel(false);
_cancellationTokenSource?.Cancel(false);
}
catch (Exception adapterException)
catch (Exception exception)
{
_logger.Warning(adapterException, "Error while initiating disconnect.");
_logger.Warning(exception, "Error while initiating disconnect.");
}
}
}
@@ -360,7 +362,7 @@ namespace MQTTnet.Client
_logger.Error(exception, "Unhandled exception while sending/receiving keep alive packets.");
}
if (_disconnectReason.TrySetException(exception))
if (!DisconnectIsPending())
{
await DisconnectInternalAsync(_keepAliveMessageSenderTask, exception).ConfigureAwait(false);
}
@@ -379,8 +381,10 @@ namespace MQTTnet.Client
{
while (!cancellationToken.IsCancellationRequested)
{
var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false);
if (packet != null)
var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken)
.ConfigureAwait(false);
if (packet != null && !cancellationToken.IsCancellationRequested)
{
await ProcessReceivedPacketAsync(packet, cancellationToken).ConfigureAwait(false);
}
@@ -407,7 +411,7 @@ namespace MQTTnet.Client
_packetDispatcher.Dispatch(exception);
if (_disconnectReason.TrySetException(exception))
if (!DisconnectIsPending())
{
await DisconnectInternalAsync(_packetReceiverTask, exception).ConfigureAwait(false);
}
@@ -493,16 +497,20 @@ namespace MQTTnet.Client
private void StartReceivingPackets(CancellationToken cancellationToken)
{
_packetReceiverTask = Task.Run(
_packetReceiverTask = Task.Factory.StartNew(
() => ReceivePacketsAsync(cancellationToken),
cancellationToken);
cancellationToken,
TaskCreationOptions.LongRunning,
TaskScheduler.Default);
}
private void StartSendingKeepAliveMessages(CancellationToken cancellationToken)
{
_keepAliveMessageSenderTask = Task.Run(
_keepAliveMessageSenderTask = Task.Factory.StartNew(
() => SendKeepAliveMessagesAsync(cancellationToken),
cancellationToken);
cancellationToken,
TaskCreationOptions.LongRunning,
TaskScheduler.Default);
}
private void FireApplicationMessageReceivedEvent(MqttPublishPacket publishPacket)
@@ -538,5 +546,10 @@ namespace MQTTnet.Client
{
}
}
private bool DisconnectIsPending()
{
return Interlocked.CompareExchange(ref _disconnectGate, 1, 0) != 0;
}
}
}
}
Source/MQTTnet/Implementations/MqttTcpChannel.cs
+1 -1
View File
@@ -116,7 +116,7 @@ namespace MQTTnet.Implementations
return true;
}
if (chain.ChainStatus.Any(c => c.Status == X509ChainStatusFlags.RevocationStatusUnknown || c.Status == X509ChainStatusFlags.Revoked || c.Status == X509ChainStatusFlags.RevocationStatusUnknown))
if (chain.ChainStatus.Any(c => c.Status == X509ChainStatusFlags.RevocationStatusUnknown || c.Status == X509ChainStatusFlags.Revoked || c.Status == X509ChainStatusFlags.OfflineRevocation))
{
if (!_options.TlsOptions.IgnoreCertificateRevocationErrors)
{
Source/MQTTnet/Implementations/MqttTcpServerListener.cs
+7 -2
View File
@@ -20,8 +20,8 @@ namespace MQTTnet.Implementations
private readonly CancellationToken _cancellationToken;
private readonly AddressFamily _addressFamily;
private readonly MqttServerTcpEndpointBaseOptions _options;
private readonly MqttServerTlsTcpEndpointOptions _tlsOptions;
private readonly X509Certificate2 _tlsCertificate;
private Socket _socket;
public MqttTcpServerListener(
@@ -36,6 +36,11 @@ namespace MQTTnet.Implementations
_tlsCertificate = tlsCertificate;
_cancellationToken = cancellationToken;
_logger = logger.CreateChildLogger(nameof(MqttTcpServerListener));
if (_options is MqttServerTlsTcpEndpointOptions tlsOptions)
{
_tlsOptions = tlsOptions;
}
}
public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
@@ -75,7 +80,7 @@ namespace MQTTnet.Implementations
if (_tlsCertificate != null)
{
sslStream = new SslStream(new NetworkStream(clientSocket), false);
await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, SslProtocols.Tls12, false).ConfigureAwait(false);
await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, _tlsOptions.SslProtocol, false).ConfigureAwait(false);
}
_logger.Verbose("Client '{0}' accepted by TCP listener '{1}, {2}'.",
Source/MQTTnet/Internal/BlockingQueue.cs
+78
View File
@@ -0,0 +1,78 @@
using System;
using System.Collections.Generic;
using System.Threading;
namespace MQTTnet.Internal
{
public class BlockingQueue<TItem>
{
private readonly object _syncRoot = new object();
private readonly LinkedList<TItem> _items = new LinkedList<TItem>();
private readonly ManualResetEvent _gate = new ManualResetEvent(false);
public int Count
{
get
{
lock (_syncRoot)
{
return _items.Count;
}
}
}
public void Enqueue(TItem item)
{
if (item == null) throw new ArgumentNullException(nameof(item));
lock (_syncRoot)
{
_items.AddLast(item);
_gate.Set();
}
}
public TItem Dequeue()
{
while (true)
{
lock (_syncRoot)
{
if (_items.Count > 0)
{
var item = _items.First.Value;
_items.RemoveFirst();
return item;
}
if (_items.Count == 0)
{
_gate.Reset();
}
}
_gate.WaitOne();
}
}
public TItem RemoveFirst()
{
lock (_syncRoot)
{
var item = _items.First;
_items.RemoveFirst();
return item.Value;
}
}
public void Clear()
{
lock (_syncRoot)
{
_items.Clear();
}
}
}
}
Source/MQTTnet/MQTTnet.csproj
+1 -1
View File
@@ -54,7 +54,7 @@
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)'=='uap10.0'">
<PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="6.1.5" />
<PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform" Version="6.1.9" />
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)'=='net452'">
Source/MQTTnet/Server/IMqttServer.cs
+6
View File
@@ -16,8 +16,14 @@ namespace MQTTnet.Server
IMqttServerOptions Options { get; }
[Obsolete("This method is no longer async. Use the not async method.")]
Task<IList<IMqttClientSessionStatus>> GetClientSessionsStatusAsync();
IList<IMqttClientSessionStatus> GetClientSessionsStatus();
IList<MqttApplicationMessage> GetRetainedMessages();
Task ClearRetainedMessagesAsync();
Task SubscribeAsync(string clientId, IList<TopicFilter> topicFilters);
Task UnsubscribeAsync(string clientId, IList<string> topicFilters);
Source/MQTTnet/Server/IMqttServerFactory.cs
+2
View File
@@ -8,6 +8,8 @@ namespace MQTTnet.Server
{
IMqttServer CreateMqttServer();
IMqttServer CreateMqttServer(IMqttNetLogger logger);
IMqttServer CreateMqttServer(IEnumerable<IMqttServerAdapter> adapters, IMqttNetLogger logger);
}
}
Source/MQTTnet/Server/IMqttServerOptions.cs
+1 -1
View File
@@ -14,7 +14,7 @@ namespace MQTTnet.Server
Action<MqttConnectionValidatorContext> ConnectionValidator { get; }
Action<MqttSubscriptionInterceptorContext> SubscriptionInterceptor { get; }
Action<MqttApplicationMessageInterceptorContext> ApplicationMessageInterceptor { get; }
Action<MqttClientMessageQueueInterceptorContext> ClientMessageQueueInterceptor { get; set; }
Action<MqttClientMessageQueueInterceptorContext> ClientMessageQueueInterceptor { get; }
MqttServerTcpEndpointOptions DefaultEndpointOptions { get; }
MqttServerTlsTcpEndpointOptions TlsEndpointOptions { get; }
Source/MQTTnet/Server/MqttClientSession.cs
+5 -1
View File
@@ -137,7 +137,9 @@ namespace MQTTnet.Server
_cleanupHandle?.Dispose();
_cleanupHandle = null;
_adapter = null;
_cancellationTokenSource?.Cancel(false);
_cancellationTokenSource?.Dispose();
_cancellationTokenSource = null;
}
@@ -289,7 +291,9 @@ namespace MQTTnet.Server
{
_pendingPacketsQueue?.Dispose();
_cancellationTokenSource?.Cancel ();
_cancellationTokenSource?.Dispose();
_cancellationTokenSource = null;
}
private void ProcessReceivedPacket(IMqttChannelAdapter adapter, MqttBasePacket packet, CancellationToken cancellationToken)
Source/MQTTnet/Server/MqttClientSessionsManager.cs
+2 -2
View File
@@ -66,7 +66,7 @@ namespace MQTTnet.Server
return Task.Run(() => RunSession(clientAdapter, _cancellationToken), _cancellationToken);
}
public Task<IList<IMqttClientSessionStatus>> GetClientStatusAsync()
public IList<IMqttClientSessionStatus> GetClientStatus()
{
var result = new List<IMqttClientSessionStatus>();
@@ -78,7 +78,7 @@ namespace MQTTnet.Server
result.Add(status);
}
return Task.FromResult((IList<IMqttClientSessionStatus>)result);
return result;
}
public void EnqueueApplicationMessage(MqttClientSession senderClientSession, MqttPublishPacket publishPacket)
Source/MQTTnet/Server/MqttRetainedMessagesManager.cs
+56 -37
View File
@@ -52,7 +52,51 @@ namespace MQTTnet.Server
try
{
await HandleMessageInternalAsync(clientId, applicationMessage).ConfigureAwait(false);
List<MqttApplicationMessage> messagesForSave = null;
lock (_messages)
{
var saveIsRequired = false;
if (applicationMessage.Payload?.Length == 0)
{
saveIsRequired = _messages.Remove(applicationMessage.Topic);
_logger.Verbose("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);
}
else
{
if (!_messages.TryGetValue(applicationMessage.Topic, out var existingMessage))
{
_messages[applicationMessage.Topic] = applicationMessage;
saveIsRequired = true;
}
else
{
if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? new byte[0]))
{
_messages[applicationMessage.Topic] = applicationMessage;
saveIsRequired = true;
}
}
_logger.Verbose("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic);
}
if (saveIsRequired)
{
messagesForSave = new List<MqttApplicationMessage>(_messages.Values);
}
}
if (messagesForSave == null)
{
_logger.Verbose("Skipped saving retained messages because no changes were detected.");
return;
}
if (_options.Storage != null)
{
await _options.Storage.SaveRetainedMessagesAsync(messagesForSave).ConfigureAwait(false);
}
}
catch (Exception exception)
{
@@ -84,52 +128,27 @@ namespace MQTTnet.Server
return retainedMessages;
}
private async Task HandleMessageInternalAsync(string clientId, MqttApplicationMessage applicationMessage)
public IList<MqttApplicationMessage> GetMessages()
{
var saveIsRequired = false;
lock (_messages)
{
if (applicationMessage.Payload?.Length == 0)
{
saveIsRequired = _messages.Remove(applicationMessage.Topic);
_logger.Info("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);
}
else
{
if (!_messages.TryGetValue(applicationMessage.Topic, out var existingMessage))
{
_messages[applicationMessage.Topic] = applicationMessage;
saveIsRequired = true;
}
else
{
if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel || !existingMessage.Payload.SequenceEqual(applicationMessage.Payload ?? new byte[0]))
{
_messages[applicationMessage.Topic] = applicationMessage;
saveIsRequired = true;
}
}
_logger.Info("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic);
}
return _messages.Values.ToList();
}
}
if (!saveIsRequired)
public Task ClearMessagesAsync()
{
lock (_messages)
{
_logger.Verbose("Skipped saving retained messages because no changes were detected.");
_messages.Clear();
}
if (saveIsRequired && _options.Storage != null)
if (_options.Storage != null)
{
List<MqttApplicationMessage> messages;
lock (_messages)
{
messages = _messages.Values.ToList();
}
await _options.Storage.SaveRetainedMessagesAsync(messages).ConfigureAwait(false);
return _options.Storage.SaveRetainedMessagesAsync(new List<MqttApplicationMessage>());
}
return Task.FromResult((object)null);
}
}
}
Source/MQTTnet/Server/MqttServer.cs
+16 -1
View File
@@ -41,7 +41,17 @@ namespace MQTTnet.Server
public Task<IList<IMqttClientSessionStatus>> GetClientSessionsStatusAsync()
{
return _clientSessionsManager.GetClientStatusAsync();
return Task.FromResult(_clientSessionsManager.GetClientStatus());
}
public IList<IMqttClientSessionStatus> GetClientSessionsStatus()
{
return _clientSessionsManager.GetClientStatus();
}
public IList<MqttApplicationMessage> GetRetainedMessages()
{
return _retainedMessagesManager.GetMessages();
}
public Task SubscribeAsync(string clientId, IList<TopicFilter> topicFilters)
@@ -129,6 +139,11 @@ namespace MQTTnet.Server
}
}
public Task ClearRetainedMessagesAsync()
{
return _retainedMessagesManager?.ClearMessagesAsync();
}
internal void OnClientConnected(string clientId)
{
_logger.Info("Client '{0}': Connected.", clientId);
Source/MQTTnet/Server/MqttServerOptionsBuilder.cs
+7
View File
@@ -1,5 +1,6 @@
using System;
using System.Net;
using System.Security.Authentication;
namespace MQTTnet.Server
{
@@ -80,6 +81,12 @@ namespace MQTTnet.Server
return this;
}
public MqttServerOptionsBuilder WithEncryptionSslProtocol(SslProtocols value)
{
_options.TlsEndpointOptions.SslProtocol = value;
return this;
}
public MqttServerOptionsBuilder WithoutEncryptedEndpoint()
{
_options.TlsEndpointOptions.IsEnabled = false;
Source/MQTTnet/Server/MqttServerTlsTcpEndpointOptions.cs
+6 -1
View File
@@ -1,4 +1,6 @@
namespace MQTTnet.Server
using System.Security.Authentication;
namespace MQTTnet.Server
{
public class MqttServerTlsTcpEndpointOptions : MqttServerTcpEndpointBaseOptions
{
@@ -8,5 +10,8 @@
}
public byte[] Certificate { get; set; }
public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12;
}
}