diff --git a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs index ff9f5303..e4eaa687 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/IManagedMqttClient.cs @@ -21,6 +21,6 @@ namespace MQTTnet.Extensions.ManagedClient Task SubscribeAsync(IEnumerable topicFilters); Task UnsubscribeAsync(IEnumerable topics); - Task PublishAsync(IEnumerable applicationMessages); + Task PublishAsync(ManagedMqttApplicationMessage applicationMessages); } } \ No newline at end of file diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index a9693a80..9bd14b21 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -99,27 +99,23 @@ namespace MQTTnet.Extensions.ManagedClient return Task.FromResult(0); } - public Task PublishAsync(IEnumerable applicationMessages) + public Task PublishAsync(MqttApplicationMessage applicationMessage) { - if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages)); + if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - return PublishAsync(applicationMessages.Select(m => - new ManagedMqttApplicationMessageBuilder().WithApplicationMessage(m).Build())); + return PublishAsync(new ManagedMqttApplicationMessageBuilder().WithApplicationMessage(applicationMessage).Build()); } - public async Task PublishAsync(IEnumerable applicationMessages) + public async Task PublishAsync(ManagedMqttApplicationMessage applicationMessage) { - if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages)); + if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - foreach (var applicationMessage in applicationMessages) + if (_storageManager != null) { - if (_storageManager != null) - { - await _storageManager.AddAsync(applicationMessage).ConfigureAwait(false); - } - - _messageQueue.Add(applicationMessage); + await _storageManager.AddAsync(applicationMessage).ConfigureAwait(false); } + + _messageQueue.Add(applicationMessage); } public async Task SubscribeAsync(IEnumerable topicFilters) diff --git a/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs b/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs index 3a99add2..e026e7ba 100644 --- a/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/IMqttChannelAdapter.cs @@ -1,5 +1,4 @@ using System; -using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using MQTTnet.Packets; @@ -21,7 +20,7 @@ namespace MQTTnet.Adapter Task DisconnectAsync(TimeSpan timeout, CancellationToken cancellationToken); - Task SendPacketsAsync(TimeSpan timeout, IEnumerable packets, CancellationToken cancellationToken); + Task SendPacketAsync(TimeSpan timeout, MqttBasePacket packet, CancellationToken cancellationToken); Task ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken); } diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index 605997a3..4fc681ad 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -1,5 +1,4 @@ using System; -using System.Collections.Generic; using System.IO; using System.Net.Sockets; using System.Runtime.InteropServices; @@ -59,17 +58,7 @@ namespace MQTTnet.Adapter Internal.TaskExtensions.TimeoutAfter(ct => _channel.DisconnectAsync(), timeout, cancellationToken)); } - public async Task SendPacketsAsync(TimeSpan timeout, IEnumerable packets, CancellationToken cancellationToken) - { - ThrowIfDisposed(); - - foreach (var packet in packets) - { - await SendPacketAsync(timeout, cancellationToken, packet).ConfigureAwait(false); - } - } - - private Task SendPacketAsync(TimeSpan timeout, CancellationToken cancellationToken, MqttBasePacket packet) + public Task SendPacketAsync(TimeSpan timeout, MqttBasePacket packet, CancellationToken cancellationToken) { return ExecuteAndWrapExceptionAsync(() => { diff --git a/Source/MQTTnet/ApplicationMessagePublisherExtensions.cs b/Source/MQTTnet/ApplicationMessagePublisherExtensions.cs index 4479f22d..dd166627 100644 --- a/Source/MQTTnet/ApplicationMessagePublisherExtensions.cs +++ b/Source/MQTTnet/ApplicationMessagePublisherExtensions.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.Threading.Tasks; using MQTTnet.Protocol; @@ -6,19 +7,33 @@ namespace MQTTnet { public static class ApplicationMessagePublisherExtensions { - public static Task PublishAsync(this IApplicationMessagePublisher publisher, params MqttApplicationMessage[] applicationMessages) + public static async Task PublishAsync(this IApplicationMessagePublisher publisher, IEnumerable applicationMessages) { if (publisher == null) throw new ArgumentNullException(nameof(publisher)); if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages)); - return publisher.PublishAsync(applicationMessages); + foreach (var applicationMessage in applicationMessages) + { + await publisher.PublishAsync(applicationMessage); + } + } + + public static async Task PublishAsync(this IApplicationMessagePublisher publisher, params MqttApplicationMessage[] applicationMessages) + { + if (publisher == null) throw new ArgumentNullException(nameof(publisher)); + if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages)); + + foreach (var applicationMessage in applicationMessages) + { + await publisher.PublishAsync(applicationMessage); + } } public static Task PublishAsync(this IApplicationMessagePublisher publisher, string topic) { if (publisher == null) throw new ArgumentNullException(nameof(publisher)); if (topic == null) throw new ArgumentNullException(nameof(topic)); - + return publisher.PublishAsync(builder => builder .WithTopic(topic) ); diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index d93a356d..79a0ecff 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -19,7 +19,7 @@ namespace MQTTnet.Client private readonly Stopwatch _sendTracker = new Stopwatch(); private readonly SemaphoreSlim _disconnectLock = new SemaphoreSlim(1, 1); private readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher(); - + private readonly IMqttClientAdapterFactory _adapterFactory; private readonly IMqttNetChildLogger _logger; @@ -145,55 +145,32 @@ namespace MQTTnet.Client await SendAndReceiveAsync(unsubscribePacket, _cancellationTokenSource.Token).ConfigureAwait(false); } - public async Task PublishAsync(IEnumerable applicationMessages) + public Task PublishAsync(MqttApplicationMessage applicationMessage) { ThrowIfNotConnected(); - var publishPackets = applicationMessages.Select(m => m.ToPublishPacket()); - var packetGroups = publishPackets.GroupBy(p => p.QualityOfServiceLevel).OrderBy(g => g.Key); + var publishPacket = applicationMessage.ToPublishPacket(); - foreach (var qosGroup in packetGroups) + switch (applicationMessage.QualityOfServiceLevel) { - switch (qosGroup.Key) - { - case MqttQualityOfServiceLevel.AtMostOnce: - { - // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier] - await SendAsync(qosGroup, _cancellationTokenSource.Token).ConfigureAwait(false); - break; - } - case MqttQualityOfServiceLevel.AtLeastOnce: - { - foreach (var publishPacket in qosGroup) - { - publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(); - await SendAndReceiveAsync(publishPacket, _cancellationTokenSource.Token).ConfigureAwait(false); - } - - break; - } - case MqttQualityOfServiceLevel.ExactlyOnce: - { - foreach (var publishPacket in qosGroup) - { - publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(); - - var pubRecPacket = await SendAndReceiveAsync(publishPacket, _cancellationTokenSource.Token).ConfigureAwait(false); - var pubRelPacket = new MqttPubRelPacket - { - PacketIdentifier = pubRecPacket.PacketIdentifier - }; - - await SendAndReceiveAsync(pubRelPacket, _cancellationTokenSource.Token).ConfigureAwait(false); - } - - break; - } - default: - { - throw new InvalidOperationException(); - } - } + case MqttQualityOfServiceLevel.AtMostOnce: + { + // No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier] + return SendAsync(publishPacket, _cancellationTokenSource.Token); + } + case MqttQualityOfServiceLevel.AtLeastOnce: + { + publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(); + return SendAndReceiveAsync(publishPacket, _cancellationTokenSource.Token); + } + case MqttQualityOfServiceLevel.ExactlyOnce: + { + return PublishExactlyOnce(publishPacket, _cancellationTokenSource.Token); + } + default: + { + throw new NotSupportedException(); + } } } @@ -295,11 +272,6 @@ namespace MQTTnet.Client } private Task SendAsync(MqttBasePacket packet, CancellationToken cancellationToken) - { - return SendAsync(new[] { packet }, cancellationToken); - } - - private Task SendAsync(IEnumerable packets, CancellationToken cancellationToken) { if (cancellationToken.IsCancellationRequested) { @@ -307,7 +279,7 @@ namespace MQTTnet.Client } _sendTracker.Restart(); - return _adapter.SendPacketsAsync(_options.CommunicationTimeout, packets, cancellationToken); + return _adapter.SendPacketAsync(_options.CommunicationTimeout, packet, cancellationToken); } private async Task SendAndReceiveAsync(MqttBasePacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttBasePacket @@ -328,7 +300,7 @@ namespace MQTTnet.Client var packetAwaiter = _packetDispatcher.AddPacketAwaiter(identifier); try { - await _adapter.SendPacketsAsync(_options.CommunicationTimeout, new[] { requestPacket }, cancellationToken).ConfigureAwait(false); + await _adapter.SendPacketAsync(_options.CommunicationTimeout, requestPacket, cancellationToken).ConfigureAwait(false); var respone = await Internal.TaskExtensions.TimeoutAfter(ct => packetAwaiter.Task, _options.CommunicationTimeout, cancellationToken).ConfigureAwait(false); return (TResponsePacket)respone; @@ -498,6 +470,19 @@ namespace MQTTnet.Client return SendAsync(response, cancellationToken); } + private async Task PublishExactlyOnce(MqttPublishPacket publishPacket, CancellationToken cancellationToken) + { + publishPacket.PacketIdentifier = _packetIdentifierProvider.GetNewPacketIdentifier(); + + var pubRecPacket = await SendAndReceiveAsync(publishPacket, cancellationToken).ConfigureAwait(false); + var pubRelPacket = new MqttPubRelPacket + { + PacketIdentifier = pubRecPacket.PacketIdentifier + }; + + await SendAndReceiveAsync(pubRelPacket, cancellationToken).ConfigureAwait(false); + } + private void StartReceivingPackets(CancellationToken cancellationToken) { _packetReceiverTask = Task.Run( diff --git a/Source/MQTTnet/IApplicationMessagePublisher.cs b/Source/MQTTnet/IApplicationMessagePublisher.cs index e137669e..c47893af 100644 --- a/Source/MQTTnet/IApplicationMessagePublisher.cs +++ b/Source/MQTTnet/IApplicationMessagePublisher.cs @@ -1,10 +1,9 @@ -using System.Collections.Generic; -using System.Threading.Tasks; +using System.Threading.Tasks; namespace MQTTnet { public interface IApplicationMessagePublisher { - Task PublishAsync(IEnumerable applicationMessages); + Task PublishAsync(MqttApplicationMessage applicationMessage); } } diff --git a/Source/MQTTnet/Internal/AsyncAutoResetEvent.cs b/Source/MQTTnet/Internal/AsyncAutoResetEvent.cs index f20733d8..132e7e05 100644 --- a/Source/MQTTnet/Internal/AsyncAutoResetEvent.cs +++ b/Source/MQTTnet/Internal/AsyncAutoResetEvent.cs @@ -47,18 +47,17 @@ namespace MQTTnet.Internal _isSignaled = false; return true; } - else if (timeout == TimeSpan.Zero) + + if (timeout == TimeSpan.Zero) { return _isSignaled; } - else - { - tcs = new TaskCompletionSource(); - _waiters.AddLast(tcs); - } + + tcs = new TaskCompletionSource(); + _waiters.AddLast(tcs); } - Task winner = await Task.WhenAny(tcs.Task, Task.Delay(timeout, cancellationToken)).ConfigureAwait(false); + var winner = await Task.WhenAny(tcs.Task, Task.Delay(timeout, cancellationToken)).ConfigureAwait(false); if (winner == tcs.Task) { // The task was signaled. diff --git a/Source/MQTTnet/Server/MqttClientPendingMessagesQueue.cs b/Source/MQTTnet/Server/MqttClientPendingMessagesQueue.cs index 15d4af90..ff43e6fc 100644 --- a/Source/MQTTnet/Server/MqttClientPendingMessagesQueue.cs +++ b/Source/MQTTnet/Server/MqttClientPendingMessagesQueue.cs @@ -115,7 +115,7 @@ namespace MQTTnet.Server return; } - await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { packet }, cancellationToken).ConfigureAwait(false); + await adapter.SendPacketAsync(_options.DefaultCommunicationTimeout, packet, cancellationToken).ConfigureAwait(false); _logger.Verbose("Enqueued packet sent (ClientId: {0}).", _clientSession.ClientId); } diff --git a/Source/MQTTnet/Server/MqttClientSession.cs b/Source/MQTTnet/Server/MqttClientSession.cs index ed029432..28e97f02 100644 --- a/Source/MQTTnet/Server/MqttClientSession.cs +++ b/Source/MQTTnet/Server/MqttClientSession.cs @@ -245,7 +245,7 @@ namespace MQTTnet.Server if (packet is MqttPingReqPacket) { - return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { new MqttPingRespPacket() }, cancellationToken); + return adapter.SendPacketAsync(_options.DefaultCommunicationTimeout, new MqttPingRespPacket(), cancellationToken); } if (packet is MqttPubRelPacket pubRelPacket) @@ -260,7 +260,7 @@ namespace MQTTnet.Server PacketIdentifier = pubRecPacket.PacketIdentifier }; - return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { responsePacket }, cancellationToken); + return adapter.SendPacketAsync(_options.DefaultCommunicationTimeout, responsePacket, cancellationToken); } if (packet is MqttPubAckPacket || packet is MqttPubCompPacket) @@ -308,7 +308,7 @@ namespace MQTTnet.Server private async Task HandleIncomingSubscribePacketAsync(IMqttChannelAdapter adapter, MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) { var subscribeResult = _subscriptionsManager.Subscribe(subscribePacket); - await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { subscribeResult.ResponsePacket }, cancellationToken).ConfigureAwait(false); + await adapter.SendPacketAsync(_options.DefaultCommunicationTimeout, subscribeResult.ResponsePacket, cancellationToken).ConfigureAwait(false); if (subscribeResult.CloseConnection) { @@ -322,7 +322,7 @@ namespace MQTTnet.Server private Task HandleIncomingUnsubscribePacketAsync(IMqttChannelAdapter adapter, MqttUnsubscribePacket unsubscribePacket, CancellationToken cancellationToken) { var unsubscribeResult = _subscriptionsManager.Unsubscribe(unsubscribePacket); - return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { unsubscribeResult }, cancellationToken); + return adapter.SendPacketAsync(_options.DefaultCommunicationTimeout, unsubscribeResult, cancellationToken); } private Task HandleIncomingPublishPacketAsync(IMqttChannelAdapter adapter, MqttPublishPacket publishPacket, CancellationToken cancellationToken) @@ -356,7 +356,7 @@ namespace MQTTnet.Server _sessionsManager.StartDispatchApplicationMessage(this, applicationMessage); var response = new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }; - return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { response }, cancellationToken); + return adapter.SendPacketAsync(_options.DefaultCommunicationTimeout, response, cancellationToken); } private Task HandleIncomingPublishPacketWithQoS2(IMqttChannelAdapter adapter, MqttApplicationMessage applicationMessage, MqttPublishPacket publishPacket, CancellationToken cancellationToken) @@ -365,13 +365,13 @@ namespace MQTTnet.Server _sessionsManager.StartDispatchApplicationMessage(this, applicationMessage); var response = new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }; - return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { response }, cancellationToken); + return adapter.SendPacketAsync(_options.DefaultCommunicationTimeout, response, cancellationToken); } private Task HandleIncomingPubRelPacketAsync(IMqttChannelAdapter adapter, MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken) { var response = new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier }; - return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] { response }, cancellationToken); + return adapter.SendPacketAsync(_options.DefaultCommunicationTimeout, response, cancellationToken); } private void OnAdapterReadingPacketCompleted(object sender, EventArgs e) diff --git a/Source/MQTTnet/Server/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/MqttClientSessionsManager.cs index dadf32a2..c52c706c 100644 --- a/Source/MQTTnet/Server/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/MqttClientSessionsManager.cs @@ -60,13 +60,12 @@ namespace MQTTnet.Server var connectReturnCode = ValidateConnection(connectPacket); if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted) { - await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] - { + await clientAdapter.SendPacketAsync(_options.DefaultCommunicationTimeout, new MqttConnAckPacket { ConnectReturnCode = connectReturnCode - } - }, cancellationToken).ConfigureAwait(false); + }, + cancellationToken).ConfigureAwait(false); return; } @@ -74,14 +73,13 @@ namespace MQTTnet.Server var result = await PrepareClientSessionAsync(connectPacket).ConfigureAwait(false); var clientSession = result.Session; - await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new[] - { + await clientAdapter.SendPacketAsync(_options.DefaultCommunicationTimeout, new MqttConnAckPacket { ConnectReturnCode = connectReturnCode, IsSessionPresent = result.IsExistingSession - } - }, cancellationToken).ConfigureAwait(false); + }, + cancellationToken).ConfigureAwait(false); Server.OnClientConnected(clientId); @@ -242,14 +240,19 @@ namespace MQTTnet.Server try { var interceptorContext = InterceptApplicationMessage(senderClientSession, applicationMessage); - if (interceptorContext.CloseConnection) + if (interceptorContext != null) { - senderClientSession.Stop(MqttClientDisconnectType.NotClean); - } + if (interceptorContext.CloseConnection) + { + senderClientSession.Stop(MqttClientDisconnectType.NotClean); + } - if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish) - { - return; + if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish) + { + return; + } + + applicationMessage = interceptorContext.ApplicationMessage; } Server.OnApplicationMessageReceived(senderClientSession?.ClientId, applicationMessage); diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index b13fdf73..a38dfd21 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -59,16 +59,13 @@ namespace MQTTnet.Server return _clientSessionsManager.UnsubscribeAsync(clientId, topicFilters); } - public Task PublishAsync(IEnumerable applicationMessages) + public Task PublishAsync(MqttApplicationMessage applicationMessage) { - if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages)); + if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); if (_cancellationTokenSource == null) throw new InvalidOperationException("The server is not started."); - foreach (var applicationMessage in applicationMessages) - { - _clientSessionsManager.StartDispatchApplicationMessage(null, applicationMessage); - } + _clientSessionsManager.StartDispatchApplicationMessage(null, applicationMessage); return Task.FromResult(0); } diff --git a/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs b/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs index 577d6960..20b581c0 100644 --- a/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs @@ -65,7 +65,7 @@ namespace MQTTnet.Benchmarks for (var i = 0; i < 10000; i++) { - _channelAdapter.SendPacketsAsync(TimeSpan.FromSeconds(15), new[] { _packet }, CancellationToken.None).GetAwaiter().GetResult(); + _channelAdapter.SendPacketAsync(TimeSpan.FromSeconds(15), _packet, CancellationToken.None).GetAwaiter().GetResult(); } _stream.Position = 0; diff --git a/Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs b/Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs index 8143728f..ffe2a77a 100644 --- a/Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs +++ b/Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs @@ -36,14 +36,11 @@ namespace MQTTnet.Core.Tests return Task.FromResult(0); } - public Task SendPacketsAsync(TimeSpan timeout, IEnumerable packets, CancellationToken cancellationToken) + public Task SendPacketAsync(TimeSpan timeout, MqttBasePacket packet, CancellationToken cancellationToken) { ThrowIfPartnerIsNull(); - foreach (var packet in packets) - { - Partner.EnqueuePacketInternal(packet); - } + Partner.EnqueuePacketInternal(packet); return Task.FromResult(0); }