Add readonly property for client options.

This commit is contained in:
Christian Kratky
2018-12-03 21:20:11 +01:00
parent 86c348df58
commit 619eaf284d
5 changed files with 32 additions and 27 deletions
+2
View File
@@ -12,9 +12,11 @@
<description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker).</description> <description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker).</description>
<releaseNotes> * [Core] Updated nuget packages due to security issues. <releaseNotes> * [Core] Updated nuget packages due to security issues.
* [Client] Fixed wrong behavior of publish method when client is disconnecting (thanks to @PaulFake). * [Client] Fixed wrong behavior of publish method when client is disconnecting (thanks to @PaulFake).
* [Client] Added readonly property for accessing options.
* [ManagedClient] Added max pending messages count option. * [ManagedClient] Added max pending messages count option.
* [ManagedClient] Add pending messages overflow strategy option. * [ManagedClient] Add pending messages overflow strategy option.
* [ManagedClient] Fixed an issue which deletes the wrong message from the internal queue (thanks to @PaulFake). * [ManagedClient] Fixed an issue which deletes the wrong message from the internal queue (thanks to @PaulFake).
* [ManagedClient] Added readonly property for accessing options.
* [Server] Added new method which exposes all retained messages. * [Server] Added new method which exposes all retained messages.
* [Server] Removed (wrong) setter from the server options interface. * [Server] Removed (wrong) setter from the server options interface.
* [Server] Fixed cpu spike in case a client disconnects (issue 421). * [Server] Fixed cpu spike in case a client disconnects (issue 421).
@@ -10,6 +10,7 @@ namespace MQTTnet.Extensions.ManagedClient
bool IsStarted { get; } bool IsStarted { get; }
bool IsConnected { get; } bool IsConnected { get; }
int PendingApplicationMessagesCount { get; } int PendingApplicationMessagesCount { get; }
IManagedMqttClientOptions Options { get; }
event EventHandler<MqttClientConnectedEventArgs> Connected; event EventHandler<MqttClientConnectedEventArgs> Connected;
event EventHandler<MqttClientDisconnectedEventArgs> Disconnected; event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
@@ -27,8 +27,7 @@ namespace MQTTnet.Extensions.ManagedClient
private CancellationTokenSource _publishingCancellationToken; private CancellationTokenSource _publishingCancellationToken;
private ManagedMqttClientStorageManager _storageManager; private ManagedMqttClientStorageManager _storageManager;
private IManagedMqttClientOptions _options;
private bool _subscriptionsNotPushed; private bool _subscriptionsNotPushed;
public ManagedMqttClient(IMqttClient mqttClient, IMqttNetChildLogger logger) public ManagedMqttClient(IMqttClient mqttClient, IMqttNetChildLogger logger)
@@ -47,6 +46,7 @@ namespace MQTTnet.Extensions.ManagedClient
public bool IsConnected => _mqttClient.IsConnected; public bool IsConnected => _mqttClient.IsConnected;
public bool IsStarted => _connectionCancellationToken != null; public bool IsStarted => _connectionCancellationToken != null;
public int PendingApplicationMessagesCount => _messageQueue.Count; public int PendingApplicationMessagesCount => _messageQueue.Count;
public IManagedMqttClientOptions Options { get; private set; }
public event EventHandler<MqttClientConnectedEventArgs> Connected; public event EventHandler<MqttClientConnectedEventArgs> Connected;
public event EventHandler<MqttClientDisconnectedEventArgs> Disconnected; public event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
@@ -70,11 +70,11 @@ namespace MQTTnet.Extensions.ManagedClient
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; Options = options;
if (_options.Storage != null) if (Options.Storage != null)
{ {
_storageManager = new ManagedMqttClientStorageManager(_options.Storage); _storageManager = new ManagedMqttClientStorageManager(Options.Storage);
var messages = await _storageManager.LoadQueuedMessagesAsync().ConfigureAwait(false); var messages = await _storageManager.LoadQueuedMessagesAsync().ConfigureAwait(false);
foreach (var message in messages) foreach (var message in messages)
@@ -116,16 +116,16 @@ namespace MQTTnet.Extensions.ManagedClient
ManagedMqttApplicationMessage removedMessage = null; ManagedMqttApplicationMessage removedMessage = null;
lock (_messageQueue) lock (_messageQueue)
{ {
if (_messageQueue.Count >= _options.MaxPendingMessages) if (_messageQueue.Count >= Options.MaxPendingMessages)
{ {
if (_options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage) if (Options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage)
{ {
_logger.Verbose("Skipping publish of new application message because internal queue is full."); _logger.Verbose("Skipping publish of new application message because internal queue is full.");
ApplicationMessageSkipped?.Invoke(this, new ApplicationMessageSkippedEventArgs(applicationMessage)); ApplicationMessageSkipped?.Invoke(this, new ApplicationMessageSkippedEventArgs(applicationMessage));
return; return;
} }
if (_options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage) if (Options.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage)
{ {
removedMessage = _messageQueue.RemoveFirst(); removedMessage = _messageQueue.RemoveFirst();
_logger.Verbose("Removed oldest application message from internal queue because it is full."); _logger.Verbose("Removed oldest application message from internal queue because it is full.");
@@ -219,7 +219,7 @@ namespace MQTTnet.Extensions.ManagedClient
if (connectionState == ReconnectionResult.NotConnected) if (connectionState == ReconnectionResult.NotConnected)
{ {
StopPublishing(); StopPublishing();
await Task.Delay(_options.AutoReconnectDelay, cancellationToken).ConfigureAwait(false); await Task.Delay(Options.AutoReconnectDelay, cancellationToken).ConfigureAwait(false);
return; return;
} }
@@ -232,7 +232,7 @@ namespace MQTTnet.Extensions.ManagedClient
if (connectionState == ReconnectionResult.StillConnected) if (connectionState == ReconnectionResult.StillConnected)
{ {
await Task.Delay(_options.ConnectionCheckInterval, cancellationToken).ConfigureAwait(false); await Task.Delay(Options.ConnectionCheckInterval, cancellationToken).ConfigureAwait(false);
} }
} }
catch (OperationCanceledException) catch (OperationCanceledException)
@@ -388,7 +388,7 @@ namespace MQTTnet.Extensions.ManagedClient
try try
{ {
await _mqttClient.ConnectAsync(_options.ClientOptions).ConfigureAwait(false); await _mqttClient.ConnectAsync(Options.ClientOptions).ConfigureAwait(false);
return ReconnectionResult.Reconnected; return ReconnectionResult.Reconnected;
} }
catch (Exception exception) catch (Exception exception)
+1
View File
@@ -7,6 +7,7 @@ namespace MQTTnet.Client
public interface IMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher, IDisposable public interface IMqttClient : IApplicationMessageReceiver, IApplicationMessagePublisher, IDisposable
{ {
bool IsConnected { get; } bool IsConnected { get; }
IMqttClientOptions Options { get; }
event EventHandler<MqttClientConnectedEventArgs> Connected; event EventHandler<MqttClientConnectedEventArgs> Connected;
event EventHandler<MqttClientDisconnectedEventArgs> Disconnected; event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
+17 -16
View File
@@ -23,7 +23,6 @@ namespace MQTTnet.Client
private readonly IMqttClientAdapterFactory _adapterFactory; private readonly IMqttClientAdapterFactory _adapterFactory;
private readonly IMqttNetChildLogger _logger; private readonly IMqttNetChildLogger _logger;
private IMqttClientOptions _options;
private CancellationTokenSource _cancellationTokenSource; private CancellationTokenSource _cancellationTokenSource;
internal Task _packetReceiverTask; internal Task _packetReceiverTask;
internal Task _keepAliveMessageSenderTask; internal Task _keepAliveMessageSenderTask;
@@ -44,6 +43,7 @@ namespace MQTTnet.Client
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived; public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public bool IsConnected { get; private set; } public bool IsConnected { get; private set; }
public IMqttClientOptions Options { get; private set; }
public async Task<MqttClientConnectResult> ConnectAsync(IMqttClientOptions options) public async Task<MqttClientConnectResult> ConnectAsync(IMqttClientOptions options)
{ {
@@ -54,7 +54,8 @@ namespace MQTTnet.Client
try try
{ {
_options = options; Options = options;
_packetIdentifierProvider.Reset(); _packetIdentifierProvider.Reset();
_packetDispatcher.Reset(); _packetDispatcher.Reset();
@@ -62,8 +63,8 @@ namespace MQTTnet.Client
_disconnectGate = 0; _disconnectGate = 0;
_adapter = _adapterFactory.CreateClientAdapter(options, _logger); _adapter = _adapterFactory.CreateClientAdapter(options, _logger);
_logger.Verbose($"Trying to connect with server ({_options.ChannelOptions})."); _logger.Verbose($"Trying to connect with server ({Options.ChannelOptions}).");
await _adapter.ConnectAsync(_options.CommunicationTimeout, _cancellationTokenSource.Token).ConfigureAwait(false); await _adapter.ConnectAsync(Options.CommunicationTimeout, _cancellationTokenSource.Token).ConfigureAwait(false);
_logger.Verbose("Connection with server established."); _logger.Verbose("Connection with server established.");
StartReceivingPackets(_cancellationTokenSource.Token); StartReceivingPackets(_cancellationTokenSource.Token);
@@ -73,7 +74,7 @@ namespace MQTTnet.Client
_sendTracker.Restart(); _sendTracker.Restart();
if (_options.KeepAlivePeriod != TimeSpan.Zero) if (Options.KeepAlivePeriod != TimeSpan.Zero)
{ {
StartSendingKeepAliveMessages(_cancellationTokenSource.Token); StartSendingKeepAliveMessages(_cancellationTokenSource.Token);
} }
@@ -197,11 +198,11 @@ namespace MQTTnet.Client
{ {
var connectPacket = new MqttConnectPacket var connectPacket = new MqttConnectPacket
{ {
ClientId = _options.ClientId, ClientId = Options.ClientId,
Username = _options.Credentials?.Username, Username = Options.Credentials?.Username,
Password = _options.Credentials?.Password, Password = Options.Credentials?.Password,
CleanSession = _options.CleanSession, CleanSession = Options.CleanSession,
KeepAlivePeriod = (ushort)_options.KeepAlivePeriod.TotalSeconds, KeepAlivePeriod = (ushort)Options.KeepAlivePeriod.TotalSeconds,
WillMessage = willApplicationMessage WillMessage = willApplicationMessage
}; };
@@ -236,7 +237,7 @@ namespace MQTTnet.Client
{ {
if (_adapter != null) if (_adapter != null)
{ {
await _adapter.DisconnectAsync(_options.CommunicationTimeout, CancellationToken.None).ConfigureAwait(false); await _adapter.DisconnectAsync(Options.CommunicationTimeout, CancellationToken.None).ConfigureAwait(false);
} }
await WaitForTaskAsync(_packetReceiverTask, sender).ConfigureAwait(false); await WaitForTaskAsync(_packetReceiverTask, sender).ConfigureAwait(false);
@@ -303,7 +304,7 @@ namespace MQTTnet.Client
try try
{ {
await _adapter.SendPacketAsync(requestPacket, cancellationToken).ConfigureAwait(false); await _adapter.SendPacketAsync(requestPacket, cancellationToken).ConfigureAwait(false);
var respone = await Internal.TaskExtensions.TimeoutAfterAsync(ct => packetAwaiter.Task, _options.CommunicationTimeout, cancellationToken).ConfigureAwait(false); var respone = await Internal.TaskExtensions.TimeoutAfterAsync(ct => packetAwaiter.Task, Options.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
return (TResponsePacket)respone; return (TResponsePacket)respone;
} }
@@ -326,10 +327,10 @@ namespace MQTTnet.Client
while (!cancellationToken.IsCancellationRequested) while (!cancellationToken.IsCancellationRequested)
{ {
var keepAliveSendInterval = TimeSpan.FromSeconds(_options.KeepAlivePeriod.TotalSeconds * 0.75); var keepAliveSendInterval = TimeSpan.FromSeconds(Options.KeepAlivePeriod.TotalSeconds * 0.75);
if (_options.KeepAliveSendInterval.HasValue) if (Options.KeepAliveSendInterval.HasValue)
{ {
keepAliveSendInterval = _options.KeepAliveSendInterval.Value; keepAliveSendInterval = Options.KeepAliveSendInterval.Value;
} }
var waitTime = keepAliveSendInterval - _sendTracker.Elapsed; var waitTime = keepAliveSendInterval - _sendTracker.Elapsed;
@@ -517,7 +518,7 @@ namespace MQTTnet.Client
try try
{ {
var applicationMessage = publishPacket.ToApplicationMessage(); var applicationMessage = publishPacket.ToApplicationMessage();
ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(_options.ClientId, applicationMessage)); ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(Options.ClientId, applicationMessage));
} }
catch (Exception exception) catch (Exception exception)
{ {