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

Add new properties and events for the managed client.

Browse Source
This commit is contained in:
Christian Kratky
2018-06-21 20:42:44 +02:00
parent 8b376874c0
commit e8d5bd7bfb
7 changed files with 58 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Build/MQTTnet.AspNetCore.nuspec
+1 -1
View File
@@ -15,7 +15,7 @@
<copyright>Copyright Christian Kratky 2016-2018</copyright>
<tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin</tags>
<dependencies>
<dependency id="MQTTnet" version="2.8.0-alpha5" />
<dependency id="MQTTnet" version="2.8.0" />
</dependencies>
</metadata>
Build/MQTTnet.Extensions.ManagedClient.nuspec
+1 -1
View File
@@ -15,7 +15,7 @@
<copyright>Copyright Christian Kratky 2016-2018</copyright>
<tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin</tags>
<dependencies>
<dependency id="MQTTnet" version="2.8.0-beta1" />
<dependency id="MQTTnet" version="2.8.0" />
</dependencies>
</metadata>
Build/MQTTnet.Extensions.Rpc.nuspec
+1 -1
View File
@@ -15,7 +15,7 @@
<copyright>Copyright Christian Kratky 2016-2018</copyright>
<tags>MQTT Message Queue Telemetry Transport MQTTClient MQTTServer Server MQTTBroker Broker NETStandard IoT InternetOfThings Messaging Hardware Arduino Sensor Actuator M2M ESP Smart Home Cities Automation Xamarin</tags>
<dependencies>
<dependency id="MQTTnet" version="2.8.0-beta1" />
<dependency id="MQTTnet" version="2.8.0" />
</dependencies>
</metadata>
Build/MQTTnet.nuspec
+3
View File
@@ -21,6 +21,9 @@
* [ManagedClient] The managed client is moved to a separate nuget package.
* [ManagedClient] Added an own message format with extended properties like ID (BREAKING CHANGE).
* [ManagedClient] Fixed a loading issue of stored application messages (thanks to @JTrotta).
* [ManagedClient] Added a new event which is fired when a synchronization of the subscriptions has failed.
* [ManagedClient] Added a new event which is fired when a connection attempt has failed.
* [ManagedClient] Exposed a new property which provides the count of not published messages (pending messages count).
* [Server] Added support for other WebSocket sub protocol formats like mqttv-3.1.1 (thanks to @israellot).
* [Server] The takeover of an existing client sessions is now treated as a _clean_ disconnect of the previous client.
* [Server] The pending messages queue per client is now limited to 250 messages. Overflow strategy and count can be changed via options (thanks to @VladimirAkopyan)
Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs
+4
View File
@@ -9,12 +9,16 @@ namespace MQTTnet.Extensions.ManagedClient
{
bool IsStarted { get; }
bool IsConnected { get; }
int PendingApplicationMessagesCount { get; }
event EventHandler<MqttClientConnectedEventArgs> Connected;
event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
event EventHandler<ApplicationMessageProcessedEventArgs> ApplicationMessageProcessed;
event EventHandler<MqttManagedProcessFailedEventArgs> ConnectingFailed;
event EventHandler<MqttManagedProcessFailedEventArgs> SynchronizingSubscriptionsFailed;
Task StartAsync(IManagedMqttClientOptions options);
Task StopAsync();
Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs
+34 -30
View File
@@ -14,8 +14,8 @@ namespace MQTTnet.Extensions.ManagedClient
public class ManagedMqttClient : IManagedMqttClient
{
private readonly BlockingCollection<ManagedMqttApplicationMessage> _messageQueue = new BlockingCollection<ManagedMqttApplicationMessage>();
private readonly ConcurrentDictionary<string, MqttQualityOfServiceLevel> _subscriptions = new ConcurrentDictionary<string, MqttQualityOfServiceLevel>();
private readonly List<string> _unsubscriptions = new List<string>();
private readonly Dictionary<string, MqttQualityOfServiceLevel> _subscriptions = new Dictionary<string, MqttQualityOfServiceLevel>();
private readonly HashSet<string> _unsubscriptions = new HashSet<string>();
private readonly IMqttClient _mqttClient;
private readonly IMqttNetChildLogger _logger;
@@ -43,12 +43,16 @@ namespace MQTTnet.Extensions.ManagedClient
public bool IsConnected => _mqttClient.IsConnected;
public bool IsStarted => _connectionCancellationToken != null;
public int PendingApplicationMessagesCount => _messageQueue.Count;
public event EventHandler<MqttClientConnectedEventArgs> Connected;
public event EventHandler<MqttClientDisconnectedEventArgs> Disconnected;
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived;
public event EventHandler<ApplicationMessageProcessedEventArgs> ApplicationMessageProcessed;
public event EventHandler SynchronizingSubscriptionsFailed;
public event EventHandler<MqttManagedProcessFailedEventArgs> ConnectingFailed;
public event EventHandler<MqttManagedProcessFailedEventArgs> SynchronizingSubscriptionsFailed;
public async Task StartAsync(IManagedMqttClientOptions options)
{
@@ -120,10 +124,13 @@ namespace MQTTnet.Extensions.ManagedClient
{
if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters));
foreach (var topicFilter in topicFilters)
lock (_subscriptions)
{
_subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
_subscriptionsNotPushed = true;
foreach (var topicFilter in topicFilters)
{
_subscriptions[topicFilter.Topic] = topicFilter.QualityOfServiceLevel;
_subscriptionsNotPushed = true;
}
}
return Task.FromResult(0);
@@ -131,12 +138,17 @@ namespace MQTTnet.Extensions.ManagedClient
public Task UnsubscribeAsync(IEnumerable<string> topics)
{
foreach (var topic in topics)
if (topics == null) throw new ArgumentNullException(nameof(topics));
lock (_subscriptions)
{
if (_subscriptions.TryRemove(topic, out _))
foreach (var topic in topics)
{
_unsubscriptions.Add(topic);
_subscriptionsNotPushed = true;
if (_subscriptions.Remove(topic))
{
_unsubscriptions.Add(topic);
_subscriptionsNotPushed = true;
}
}
}
@@ -210,7 +222,7 @@ namespace MQTTnet.Extensions.ManagedClient
}
}
private async Task PublishQueuedMessagesAsync(CancellationToken cancellationToken)
private void PublishQueuedMessages(CancellationToken cancellationToken)
{
try
{
@@ -222,12 +234,9 @@ namespace MQTTnet.Extensions.ManagedClient
continue;
}
if (cancellationToken.IsCancellationRequested)
{
continue;
}
cancellationToken.ThrowIfCancellationRequested();
await TryPublishQueuedMessageAsync(message).ConfigureAwait(false);
TryPublishQueuedMessage(message);
}
}
catch (OperationCanceledException)
@@ -243,17 +252,13 @@ namespace MQTTnet.Extensions.ManagedClient
}
}
private async Task TryPublishQueuedMessageAsync(ManagedMqttApplicationMessage message)
private void TryPublishQueuedMessage(ManagedMqttApplicationMessage message)
{
Exception transmitException = null;
try
{
await _mqttClient.PublishAsync(message.ApplicationMessage).ConfigureAwait(false);
if (_storageManager != null)
{
await _storageManager.RemoveAsync(message).ConfigureAwait(false);
}
_mqttClient.PublishAsync(message.ApplicationMessage).GetAwaiter().GetResult();
_storageManager?.RemoveAsync(message).GetAwaiter().GetResult();
}
catch (MqttCommunicationException exception)
{
@@ -282,13 +287,13 @@ namespace MQTTnet.Extensions.ManagedClient
_logger.Info(nameof(ManagedMqttClient), "Synchronizing subscriptions");
List<TopicFilter> subscriptions;
List<string> unsubscriptions;
HashSet<string> unsubscriptions;
lock (_subscriptions)
{
subscriptions = _subscriptions.Select(i => new TopicFilter(i.Key, i.Value)).ToList();
unsubscriptions = new List<string>(_unsubscriptions);
unsubscriptions = new HashSet<string>(_unsubscriptions);
_unsubscriptions.Clear();
_subscriptionsNotPushed = false;
@@ -316,7 +321,7 @@ namespace MQTTnet.Extensions.ManagedClient
_logger.Warning(exception, "Synchronizing subscriptions failed.");
_subscriptionsNotPushed = true;
SynchronizingSubscriptionsFailed?.Invoke(this, EventArgs.Empty);
SynchronizingSubscriptionsFailed?.Invoke(this, new MqttManagedProcessFailedEventArgs(exception));
}
}
@@ -332,8 +337,9 @@ namespace MQTTnet.Extensions.ManagedClient
await _mqttClient.ConnectAsync(_options.ClientOptions).ConfigureAwait(false);
return ReconnectionResult.Reconnected;
}
catch (Exception)
catch (Exception exception)
{
ConnectingFailed?.Invoke(this, new MqttManagedProcessFailedEventArgs(exception));
return ReconnectionResult.NotConnected;
}
}
@@ -364,9 +370,7 @@ namespace MQTTnet.Extensions.ManagedClient
_publishingCancellationToken = cts;
#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
Task.Run(() => PublishQueuedMessagesAsync(cts.Token), cts.Token);
#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
Task.Factory.StartNew(() => PublishQueuedMessages(cts.Token), cts.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
}
private void StopPublishing()
Source/MQTTnet.Extensions.ManagedClient/MqttManagedProcessFailedEventArgs.cs
+14
View File
@@ -0,0 +1,14 @@
using System;
namespace MQTTnet.Extensions.ManagedClient
{
public class MqttManagedProcessFailedEventArgs : EventArgs
{
public MqttManagedProcessFailedEventArgs(Exception exception)
{
Exception = exception;
}
public Exception Exception { get; }
}
}