Fixed an issue with queued client message in the server.

This commit is contained in:
Christian Kratky
2017-11-07 23:30:15 +01:00
parent 6e5b991406
commit f481c704c9
4 changed files with 56 additions and 6 deletions
@@ -0,0 +1,49 @@
using System;
using MQTTnet.Core.Client;
namespace MQTTnet.Core.ManagedClient
{
public class ManagedMqttClientOptionsBuilder
{
private readonly ManagedMqttClientOptions _options = new ManagedMqttClientOptions();
public ManagedMqttClientOptionsBuilder WithAutoReconnectDelay(TimeSpan value)
{
_options.AutoReconnectDelay = value;
return this;
}
public ManagedMqttClientOptionsBuilder WithStorage(IManagedMqttClientStorage value)
{
_options.Storage = value;
return this;
}
public ManagedMqttClientOptionsBuilder WithClientOptions(IMqttClientOptions value)
{
_options.ClientOptions = value ?? throw new ArgumentNullException(nameof(value));
return this;
}
public ManagedMqttClientOptionsBuilder WithClientOptions(Action<MqttClientOptionsBuilder> options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
var builder = new MqttClientOptionsBuilder();
options(builder);
_options.ClientOptions = builder.Build();
return this;
}
public ManagedMqttClientOptions Build()
{
if (_options.ClientOptions == null)
{
throw new InvalidOperationException("The ClientOptions cannot be null.");
}
return _options;
}
}
}
@@ -59,11 +59,12 @@ namespace MQTTnet.Core.Server
private async Task SendPendingPublishPacketAsync(IMqttCommunicationAdapter adapter, CancellationToken cancellationToken)
{
var packet = _pendingPublishPackets.Take(cancellationToken);
MqttPublishPacket packet = null;
try
{
packet = _pendingPublishPackets.Take(cancellationToken);
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, packet).ConfigureAwait(false);
_logger.LogTrace("Enqueued packet sent (ClientId: {0}).", _session.ClientId);
}
catch (Exception exception)
@@ -84,10 +85,10 @@ namespace MQTTnet.Core.Server
_logger.LogError(new EventId(), exception, "Sending publish packet failed (ClientId: {0}).", _session.ClientId);
}
if (packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
if (packet != null && packet.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce)
{
packet.Dup = true;
_pendingPublishPackets.Add(packet, cancellationToken);
_pendingPublishPackets.Add(packet, CancellationToken.None);
}
_session.Stop();
+1 -1
View File
@@ -4,7 +4,7 @@ namespace MQTTnet.Core
{
public sealed class TopicFilter
{
public TopicFilter(string topic, MqttQualityOfServiceLevel qualityOfServiceLevel)
public TopicFilter(string topic, MqttQualityOfServiceLevel qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce)
{
Topic = topic;
QualityOfServiceLevel = qualityOfServiceLevel;
+1 -1
View File
@@ -5,8 +5,8 @@ namespace MQTTnet.Core
{
public class TopicFilterBuilder
{
private string _topic;
private MqttQualityOfServiceLevel _qualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce;
private string _topic;
public TopicFilterBuilder WithTopic(string topic)
{