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

Merge branch 'master' into feature/Single_Connection_Monitoring_Task

Browse Source
This commit is contained in:
Christian Kratky
2020-11-09 21:29:38 +01:00
parent e1bf9de347 88b0af9b21
commit f894622884
13 changed files with 92 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Build/MQTTnet.nuspec
+2 -1
View File
@@ -13,11 +13,12 @@
<description>MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports v3.1.0, v3.1.1 and v5.0.0 of the MQTT protocol.</description>
<releaseNotes>
* [Core] Added support for TLS 1.3 (requires .NET Core 3.1+) (thanks to @Dvergatal).
* [Core] Added support for _MaximumQoS_ transmit when using MQTTv5 (thanks to @nayato).
* [RpcClient] Moved the RPC topic validation to the topic generation strategy.
* [RcpClient] Added interface for _MqttRpcClient_.
* [Server] Reduced async tasks count by moving dedicated keep alive tasks per connection to shared one.
* [Server] Session takeover and keep alive timeout are now properly set in DISCONNECT packet.
* [Server] Performance improvements.
* [Server] Fixed bug in PubRel packet generation (MQTTv5 only).
</releaseNotes>
<copyright>Copyright Christian Kratky 2016-2020</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 Blazor</tags>
Source/MQTTnet/Client/Connecting/MqttClientAuthenticateResult.cs
+3
View File
@@ -1,5 +1,6 @@
using System.Collections.Generic;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Client.Connecting
{
@@ -25,6 +26,8 @@ namespace MQTTnet.Client.Connecting
public string ReasonString { get; set; }
public ushort? ReceiveMaximum { get; set; }
public MqttQualityOfServiceLevel MaximumQoS { get; set; }
public string ResponseInformation { get; set; }
Source/MQTTnet/Formatter/V5/MqttV500DataConverter.cs
+1
View File
@@ -102,6 +102,7 @@ namespace MQTTnet.Formatter.V5
MaximumPacketSize = connAckPacket.Properties?.MaximumPacketSize,
ReasonString = connAckPacket.Properties?.ReasonString,
ReceiveMaximum = connAckPacket.Properties?.ReceiveMaximum,
MaximumQoS = connAckPacket.Properties?.MaximumQoS ?? MqttQualityOfServiceLevel.ExactlyOnce,
ResponseInformation = connAckPacket.Properties?.ResponseInformation,
TopicAliasMaximum = connAckPacket.Properties?.TopicAliasMaximum,
ServerReference = connAckPacket.Properties?.ServerReference,
Source/MQTTnet/Formatter/V5/MqttV500PacketDecoder.cs
+4
View File
@@ -254,6 +254,10 @@ namespace MQTTnet.Formatter.V5
{
packet.Properties.ReceiveMaximum = propertiesReader.ReadReceiveMaximum();
}
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.MaximumQoS)
{
packet.Properties.MaximumQoS = propertiesReader.ReadMaximumQoS();
}
else if (propertiesReader.CurrentPropertyId == MqttPropertyId.AssignedClientIdentifier)
{
packet.Properties.AssignedClientIdentifier = propertiesReader.ReadAssignedClientIdentifier();
Source/MQTTnet/Formatter/V5/MqttV500PacketEncoder.cs
+2 -6
View File
@@ -10,14 +10,9 @@ namespace MQTTnet.Formatter.V5
{
readonly IMqttPacketWriter _packetWriter;
public MqttV500PacketEncoder()
: this(new MqttPacketWriter())
{
}
public MqttV500PacketEncoder(IMqttPacketWriter packetWriter)
{
_packetWriter = packetWriter;
_packetWriter = packetWriter ?? throw new ArgumentNullException(nameof(packetWriter));
}
public ArraySegment<byte> Encode(MqttBasePacket packet)
@@ -201,6 +196,7 @@ namespace MQTTnet.Formatter.V5
propertiesWriter.WriteAuthenticationData(packet.Properties.AuthenticationData);
propertiesWriter.WriteRetainAvailable(packet.Properties.RetainAvailable);
propertiesWriter.WriteReceiveMaximum(packet.Properties.ReceiveMaximum);
propertiesWriter.WriteMaximumQoS(packet.Properties.MaximumQoS);
propertiesWriter.WriteAssignedClientIdentifier(packet.Properties.AssignedClientIdentifier);
propertiesWriter.WriteTopicAliasMaximum(packet.Properties.TopicAliasMaximum);
propertiesWriter.WriteReasonString(packet.Properties.ReasonString);
Source/MQTTnet/Formatter/V5/MqttV500PacketFormatter.cs
+3 -8
View File
@@ -4,15 +4,10 @@ using MQTTnet.Packets;
namespace MQTTnet.Formatter.V5
{
public class MqttV500PacketFormatter : IMqttPacketFormatter
public sealed class MqttV500PacketFormatter : IMqttPacketFormatter
{
private readonly MqttV500PacketEncoder _encoder;
private readonly MqttV500PacketDecoder _decoder = new MqttV500PacketDecoder();
public MqttV500PacketFormatter()
{
_encoder = new MqttV500PacketEncoder();
}
readonly MqttV500PacketDecoder _decoder = new MqttV500PacketDecoder();
readonly MqttV500PacketEncoder _encoder;
public MqttV500PacketFormatter(IMqttPacketWriter writer)
{
Source/MQTTnet/Formatter/V5/MqttV500PropertiesReader.cs
+11
View File
@@ -82,6 +82,17 @@ namespace MQTTnet.Formatter.V5
return _body.ReadTwoByteInteger();
}
public MqttQualityOfServiceLevel ReadMaximumQoS()
{
byte value = _body.ReadByte();
if (value > 1)
{
throw new MqttProtocolViolationException($"Unexpected Maximum QoS value: {value}");
}
return (MqttQualityOfServiceLevel)value;
}
public string ReadAssignedClientIdentifier()
{
return _body.ReadStringWithLengthPrefix();
Source/MQTTnet/Formatter/V5/MqttV500PropertiesWriter.cs
+11
View File
@@ -132,6 +132,17 @@ namespace MQTTnet.Formatter.V5
Write(MqttPropertyId.ReceiveMaximum, value);
}
public void WriteMaximumQoS(MqttQualityOfServiceLevel? value)
{
if (!value.HasValue || value.Value > MqttQualityOfServiceLevel.AtLeastOnce)
{
return;
}
_packetWriter.Write((byte)MqttPropertyId.MaximumQoS);
_packetWriter.Write((byte)value.Value);
}
public void WriteMaximumPacketSize(uint? value)
{
WriteAsFourByteInteger(MqttPropertyId.MaximumPacketSize, value);
Source/MQTTnet/Implementations/MqttTcpChannel.cs
-4
View File
@@ -191,10 +191,6 @@ namespace MQTTnet.Implementations
}
await stream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
// This subsequent call is required to check whether the socket is still connected.
// Without this call a broken connection is only recognized at the next call.
await stream.WriteAsync(PlatformAbstractionLayer.EmptyByteArray, 0, 0, cancellationToken);
}
}
catch (ObjectDisposedException)
Source/MQTTnet/Packets/MqttConnAckPacketProperties.cs
+3
View File
@@ -1,4 +1,5 @@
using System.Collections.Generic;
using MQTTnet.Protocol;
namespace MQTTnet.Packets
{
@@ -8,6 +9,8 @@ namespace MQTTnet.Packets
public ushort? ReceiveMaximum { get; set; }
public MqttQualityOfServiceLevel? MaximumQoS { get; set; }
public bool? RetainAvailable { get; set; }
public uint? MaximumPacketSize { get; set; }
Source/MQTTnet/Server/MqttClientConnection.cs
+6 -1
View File
@@ -446,7 +446,12 @@ namespace MQTTnet.Server
await SendAsync(publishPacket, cancellationToken).ConfigureAwait(false);
await awaiter1.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false);
await SendAsync(new MqttPubRelPacket { PacketIdentifier = publishPacket.PacketIdentifier }, cancellationToken).ConfigureAwait(false);
await SendAsync(new MqttPubRelPacket
{
PacketIdentifier = publishPacket.PacketIdentifier,
ReasonCode = MqttPubRelReasonCode.Success
}, cancellationToken).ConfigureAwait(false);
await awaiter2.WaitOneAsync(_serverOptions.DefaultCommunicationTimeout).ConfigureAwait(false);
}
}
Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs
+6 -1
View File
@@ -5,6 +5,7 @@ using MQTTnet.Diagnostics;
using MQTTnet.Server;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
@@ -45,6 +46,8 @@ namespace MQTTnet.Tests.Mockups
ServerLogger.LogMessagePublished += (s, e) =>
{
Debug.WriteLine(e.LogMessage.ToString());
if (e.LogMessage.Level == MqttNetLogLevel.Error)
{
lock (_serverErrors)
@@ -56,6 +59,8 @@ namespace MQTTnet.Tests.Mockups
ClientLogger.LogMessagePublished += (s, e) =>
{
Debug.WriteLine(e.LogMessage.ToString());
if (e.LogMessage.Level == MqttNetLogLevel.Error)
{
lock (_clientErrors)
@@ -159,7 +164,7 @@ namespace MQTTnet.Tests.Mockups
return client;
}
public void ThrowIfLogErrors()
{
lock (_serverErrors)
Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs
+40
View File
@@ -15,6 +15,10 @@ using System.Net.Sockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Client.Receiving;
using MQTTnet.Diagnostics;
using MQTTnet.Extensions.ManagedClient;
using MQTTnet.Formatter;
namespace MQTTnet.Tests
{
@@ -743,5 +747,41 @@ namespace MQTTnet.Tests
Assert.AreEqual(null, receivedMessage.Payload);
}
}
[TestMethod]
public async Task Subscribe_With_QoS2()
{
using (var testEnvironment = new TestEnvironment())
{
await testEnvironment.StartServerAsync();
var client1 = await testEnvironment.ConnectClientAsync(o => o.WithProtocolVersion(MqttProtocolVersion.V500));
var client2 = await testEnvironment.ConnectClientAsync(o => o.WithProtocolVersion(MqttProtocolVersion.V500));
var disconnectedFired = false;
client1.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(c =>
{
disconnectedFired = true;
});
var messageReceived = false;
client1.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(c =>
{
messageReceived = true;
});
await client1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("topic1").WithExactlyOnceQoS().Build());
await Task.Delay(500);
var message = new MqttApplicationMessageBuilder().WithTopic("topic1").WithPayload("Hello World").WithExactlyOnceQoS().WithRetainFlag().Build();
await client2.PublishAsync(message);
await Task.Delay(500);
Assert.IsTrue(messageReceived);
Assert.IsTrue(client1.IsConnected);
Assert.IsFalse(disconnectedFired);
}
}
}
}