diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttCommunicationAdapterFactory.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttCommunicationAdapterFactory.cs index 05e6fb63..ceaac98b 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttCommunicationAdapterFactory.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttCommunicationAdapterFactory.cs @@ -21,6 +21,11 @@ namespace MQTTnet.Implementations return new MqttChannelCommunicationAdapter(new MqttWebSocketChannel(webSocketOptions), new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion }); } + if (options is MqttClientManagedOptions queuedOptions) + { + return new MqttChannelCommunicationAdapter(new MqttTcpChannel(queuedOptions), new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion }); + } + throw new NotSupportedException(); } } diff --git a/Frameworks/MQTTnet.NetStandard/MqttClientFactory.cs b/Frameworks/MQTTnet.NetStandard/MqttClientFactory.cs index 15da033d..53ce2765 100644 --- a/Frameworks/MQTTnet.NetStandard/MqttClientFactory.cs +++ b/Frameworks/MQTTnet.NetStandard/MqttClientFactory.cs @@ -9,5 +9,10 @@ namespace MQTTnet { return new MqttClient(new MqttCommunicationAdapterFactory()); } + + public IMqttClientManaged CreateMqttManagedClient() + { + return new MqttClientManaged(new MqttCommunicationAdapterFactory()); + } } } \ No newline at end of file diff --git a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttCommunicationAdapterFactory.cs b/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttCommunicationAdapterFactory.cs index 05e6fb63..9e429fd1 100644 --- a/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttCommunicationAdapterFactory.cs +++ b/Frameworks/MQTTnet.UniversalWindows/Implementations/MqttCommunicationAdapterFactory.cs @@ -21,6 +21,12 @@ namespace MQTTnet.Implementations return new MqttChannelCommunicationAdapter(new MqttWebSocketChannel(webSocketOptions), new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion }); } + if (options is MqttClientManagedOptions queuedOptions) + { + return new MqttChannelCommunicationAdapter(new MqttTcpChannel(queuedOptions), new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion }); + } + + throw new NotSupportedException(); } } diff --git a/Frameworks/MQTTnet.UniversalWindows/MqttClientFactory.cs b/Frameworks/MQTTnet.UniversalWindows/MqttClientFactory.cs index 15da033d..53ce2765 100644 --- a/Frameworks/MQTTnet.UniversalWindows/MqttClientFactory.cs +++ b/Frameworks/MQTTnet.UniversalWindows/MqttClientFactory.cs @@ -9,5 +9,10 @@ namespace MQTTnet { return new MqttClient(new MqttCommunicationAdapterFactory()); } + + public IMqttClientManaged CreateMqttManagedClient() + { + return new MqttClientManaged(new MqttCommunicationAdapterFactory()); + } } } \ No newline at end of file diff --git a/MQTTnet.Core/Client/IMqttClientFactory.cs b/MQTTnet.Core/Client/IMqttClientFactory.cs index 033eb993..25e03b39 100644 --- a/MQTTnet.Core/Client/IMqttClientFactory.cs +++ b/MQTTnet.Core/Client/IMqttClientFactory.cs @@ -3,5 +3,7 @@ public interface IMqttClientFactory { IMqttClient CreateMqttClient(); + + IMqttClientManaged CreateMqttManagedClient(); } } \ No newline at end of file diff --git a/MQTTnet.Core/Client/IMqttClientManaged.cs b/MQTTnet.Core/Client/IMqttClientManaged.cs new file mode 100644 index 00000000..1fc42f09 --- /dev/null +++ b/MQTTnet.Core/Client/IMqttClientManaged.cs @@ -0,0 +1,9 @@ +using System.Threading.Tasks; + +namespace MQTTnet.Core.Client +{ + public interface IMqttClientManaged : IMqttClient + { + //Task ConnectAsync(MqttClientManagedOptions options); + } +} diff --git a/MQTTnet.Core/Client/IMqttClientQueuedStorage.cs b/MQTTnet.Core/Client/IMqttClientQueuedStorage.cs new file mode 100644 index 00000000..7909a563 --- /dev/null +++ b/MQTTnet.Core/Client/IMqttClientQueuedStorage.cs @@ -0,0 +1,12 @@ +using System.Collections.Generic; +using System.Threading.Tasks; + +namespace MQTTnet.Core.Client +{ + public interface IMqttClientQueuedStorage + { + Task SaveQueuedMessagesAsync(IList messages); + + Task> LoadQueuedMessagesAsync(); + } +} diff --git a/MQTTnet.Core/Client/MqttClientManaged .cs b/MQTTnet.Core/Client/MqttClientManaged .cs new file mode 100644 index 00000000..03afc926 --- /dev/null +++ b/MQTTnet.Core/Client/MqttClientManaged .cs @@ -0,0 +1,163 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +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.Packets; +using MQTTnet.Core.Protocol; +using MQTTnet.Core.Internal; + +namespace MQTTnet.Core.Client +{ + public class MqttClientManaged: IMqttClientManaged + { + private MqttClientManagedOptions _options; + private int _latestPacketIdentifier; + private readonly BlockingCollection _inflightQueue; + private bool _usePersistance = false; + private MqttClientQueuedPersistentMessagesManager _persistentMessagesManager; + private readonly MqttClient _baseMqttClient; + + public MqttClientManaged(IMqttCommunicationAdapterFactory communicationChannelFactory) + { + _baseMqttClient = new MqttClient(communicationChannelFactory); + _baseMqttClient.Connected += BaseMqttClient_Connected; + _baseMqttClient.Disconnected += BaseMqttClient_Disconnected; + _baseMqttClient.ApplicationMessageReceived += BaseMqttClient_ApplicationMessageReceived; + _inflightQueue = new BlockingCollection(); + } + + private void BaseMqttClient_ApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e) + { + ApplicationMessageReceived?.Invoke(this, e); + } + + private void BaseMqttClient_Disconnected(object sender, EventArgs e) + { + Disconnected?.Invoke(this, e); + } + + private void BaseMqttClient_Connected(object sender, EventArgs e) + { + Connected?.Invoke(this, e); + } + + public event EventHandler Connected; + public event EventHandler Disconnected; + public event EventHandler ApplicationMessageReceived; + + public bool IsConnected => _baseMqttClient.IsConnected; + + + public async Task ConnectAsync(IMqttClientOptions options) + { + //TODO VERY BAD + _options = options as MqttClientManagedOptions; + this._usePersistance = _options.Storage != null; + await _baseMqttClient.ConnectAsync(options); + SetupOutgoingPacketProcessingAsync(); + + //load persistentMessages + if (_usePersistance) + { + if (_persistentMessagesManager == null) + _persistentMessagesManager = new MqttClientQueuedPersistentMessagesManager(_options); + await _persistentMessagesManager.LoadMessagesAsync(); + await InternalPublishAsync(_persistentMessagesManager.GetMessages(), false); + } + } + + public async Task DisconnectAsync() + { + await _baseMqttClient.DisconnectAsync(); + } + + public async Task UnsubscribeAsync(IEnumerable topicFilters) + { + await _baseMqttClient.UnsubscribeAsync(topicFilters); + } + + public async Task PublishAsync(IEnumerable applicationMessages) + { + await InternalPublishAsync(applicationMessages, true); + } + + private async Task InternalPublishAsync(IEnumerable applicationMessages, bool appendIfUsePersistance) + { + ThrowIfNotConnected(); + + foreach (var applicationMessage in applicationMessages) + { + if (_usePersistance && appendIfUsePersistance) + await _persistentMessagesManager.SaveMessageAsync(applicationMessage); + + _inflightQueue.Add(applicationMessage); + } + } + + public async Task> SubscribeAsync(IEnumerable topicFilters) + { + return await _baseMqttClient.SubscribeAsync(topicFilters); + } + + private void ThrowIfNotConnected() + { + if (!IsConnected) throw new MqttCommunicationException("The client is not connected."); + } + + private ushort GetNewPacketIdentifier() + { + return (ushort)Interlocked.Increment(ref _latestPacketIdentifier); + } + + + private void SetupOutgoingPacketProcessingAsync() + { + Task.Factory.StartNew( + () => SendPackets(_baseMqttClient._cancellationTokenSource.Token), + _baseMqttClient._cancellationTokenSource.Token, + TaskCreationOptions.LongRunning, + TaskScheduler.Default).ConfigureAwait(false); + } + + private async Task SendPackets(CancellationToken cancellationToken) + { + MqttNetTrace.Information(nameof(MqttClientManaged), "Start sending packets."); + MqttApplicationMessage messageInQueue = null; + + try + { + while (!cancellationToken.IsCancellationRequested) + { + messageInQueue = _inflightQueue.Take(); + await _baseMqttClient.PublishAsync(new List() { messageInQueue }); + if (_usePersistance) + await _persistentMessagesManager.Remove(messageInQueue); + } + } + catch (OperationCanceledException) + { + } + catch (MqttCommunicationException exception) + { + MqttNetTrace.Warning(nameof(MqttClient), exception, "MQTT communication exception while sending packets."); + //message not send, equeue it again + if (messageInQueue != null) + _inflightQueue.Add(messageInQueue); + } + catch (Exception exception) + { + MqttNetTrace.Error(nameof(MqttClient), exception, "Unhandled exception while sending packets."); + await DisconnectAsync().ConfigureAwait(false); + } + finally + { + MqttNetTrace.Information(nameof(MqttClient), "Stopped sending packets."); + } + } + } +} diff --git a/MQTTnet.Core/Client/MqttClientManagedOptions.cs b/MQTTnet.Core/Client/MqttClientManagedOptions.cs new file mode 100644 index 00000000..77feabed --- /dev/null +++ b/MQTTnet.Core/Client/MqttClientManagedOptions.cs @@ -0,0 +1,12 @@ + +using System; + +namespace MQTTnet.Core.Client +{ + public class MqttClientManagedOptions: MqttClientTcpOptions + { + public bool UseAutoReconnect { get; set; } + public TimeSpan AutoReconnectDelay { get; set; } + public IMqttClientQueuedStorage Storage { get; set; } + } +} diff --git a/MQTTnet.Core/Client/MqttClientQueuedPersistentMessagesManager.cs b/MQTTnet.Core/Client/MqttClientQueuedPersistentMessagesManager.cs new file mode 100644 index 00000000..b012e215 --- /dev/null +++ b/MQTTnet.Core/Client/MqttClientQueuedPersistentMessagesManager.cs @@ -0,0 +1,84 @@ +using MQTTnet.Core.Diagnostics; +using MQTTnet.Core.Packets; +using System; +using System.Linq; +using System.Collections.Generic; +using System.Threading.Tasks; + +namespace MQTTnet.Core.Client +{ + public class MqttClientQueuedPersistentMessagesManager + { + private readonly IList _persistedMessages = new List(); + private readonly MqttClientManagedOptions _options; + + public MqttClientQueuedPersistentMessagesManager(MqttClientManagedOptions options) + { + _options = options ?? throw new ArgumentNullException(nameof(options)); + } + + public async Task LoadMessagesAsync() + { + try + { + var persistentMessages = await _options.Storage.LoadQueuedMessagesAsync(); + lock (_persistedMessages) + { + _persistedMessages.Clear(); + foreach (var persistentMessage in persistentMessages) + { + _persistedMessages.Add(persistentMessage); + } + } + } + catch (Exception exception) + { + MqttNetTrace.Error(nameof(MqttClientQueuedPersistentMessagesManager), exception, "Unhandled exception while loading persistent messages."); + } + } + + public async Task SaveMessageAsync(MqttApplicationMessage applicationMessage) + { + if (applicationMessage != null) + { + lock (_persistedMessages) + { + _persistedMessages.Add(applicationMessage); + } + } + try + { + if (_options.Storage != null) + { + await _options.Storage.SaveQueuedMessagesAsync(_persistedMessages); + } + } + catch (Exception exception) + { + MqttNetTrace.Error(nameof(MqttClientQueuedPersistentMessagesManager), exception, "Unhandled exception while saving persistent messages."); + } + } + + public List GetMessages() + { + var persistedMessages = new List(); + lock (_persistedMessages) + { + foreach (var persistedMessage in _persistedMessages) + { + persistedMessages.Add(persistedMessage); + } + } + + return persistedMessages; + } + + public async Task Remove(MqttApplicationMessage message) + { + lock (_persistedMessages) + _persistedMessages.Remove(message); + + await SaveMessageAsync(null); + } + } +} diff --git a/MQTTnet.Core/MQTTnet.Core.csproj b/MQTTnet.Core/MQTTnet.Core.csproj index 679e4a85..89fdc9c0 100644 --- a/MQTTnet.Core/MQTTnet.Core.csproj +++ b/MQTTnet.Core/MQTTnet.Core.csproj @@ -22,15 +22,7 @@ - - - - - - - - diff --git a/Tests/MQTTnet.TestApp.NetCore/messages.bin b/Tests/MQTTnet.TestApp.NetCore/messages.bin new file mode 100644 index 00000000..59505fa8 Binary files /dev/null and b/Tests/MQTTnet.TestApp.NetCore/messages.bin differ