YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-16 09:13:02 +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-10-31 11:33:54 +01:00
parent 42c1155bee 34ae54d542
commit e800901b4c
12 changed files with 73 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Build/MQTTnet.nuspec
+1 -1
View File
@@ -12,7 +12,7 @@
<requireLicenseAcceptance>true</requireLicenseAcceptance>
<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>
* [Server] ..
* [Core] Added support for TLS 1.3 (requires .NET Core 3.1+) (thanks to @Dvergatal).
</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.Extensions.WebSocket4Net/WebSocket4NetMqttChannel.cs
+8 -8
View File
@@ -16,14 +16,14 @@ using WebSocket4Net;
namespace MQTTnet.Extensions.WebSocket4Net
{
public class WebSocket4NetMqttChannel : IMqttChannel
public sealed class WebSocket4NetMqttChannel : IMqttChannel
{
private readonly BlockingCollection<byte> _receiveBuffer = new BlockingCollection<byte>();
readonly BlockingCollection<byte> _receiveBuffer = new BlockingCollection<byte>();
private readonly IMqttClientOptions _clientOptions;
private readonly MqttClientWebSocketOptions _webSocketOptions;
readonly IMqttClientOptions _clientOptions;
readonly MqttClientWebSocketOptions _webSocketOptions;
private WebSocket _webSocket;
WebSocket _webSocket;
public WebSocket4NetMqttChannel(IMqttClientOptions clientOptions, MqttClientWebSocketOptions webSocketOptions)
{
@@ -166,12 +166,12 @@ namespace MQTTnet.Extensions.WebSocket4Net
_webSocket = null;
}
private void OnError(object sender, ErrorEventArgs e)
static void OnError(object sender, ErrorEventArgs e)
{
System.Diagnostics.Debug.Write(e.Exception.ToString());
}
private void OnDataReceived(object sender, DataReceivedEventArgs e)
void OnDataReceived(object sender, DataReceivedEventArgs e)
{
foreach (var @byte in e.Data)
{
@@ -179,7 +179,7 @@ namespace MQTTnet.Extensions.WebSocket4Net
}
}
private async Task ConnectInternalAsync(CancellationToken cancellationToken)
async Task ConnectInternalAsync(CancellationToken cancellationToken)
{
_webSocket.Error += OnError;
_webSocket.DataReceived += OnDataReceived;
Source/MQTTnet.Extensions.Wrappers.WebSocket4Net/WebSocket4NetMqttClientAdapterFactory.cs
-134
View File
@@ -1,134 +0,0 @@
using MQTTnet.Client;
using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Security.Authentication;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Adapter;
using MQTTnet.Channel;
using MQTTnet.Diagnostics;
using MQTTnet.Serializer;
using WebSocket4Net;
namespace MQTTnet.TestApp.NetCore
{
public class WebSocket4NetMqttClientAdapterFactory : IMqttClientAdapterFactory
{
public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger)
{
if (options == null) throw new ArgumentNullException(nameof(options));
if (logger == null) throw new ArgumentNullException(nameof(logger));
if (!(options.ChannelOptions is MqttClientWebSocketOptions))
{
throw new NotSupportedException("Only WebSocket connections are supported.");
}
return new MqttChannelAdapter(new WebSocket4NetMqttChannel(options), new MqttPacketSerializer(), logger);
}
private class WebSocket4NetMqttChannel : IMqttChannel
{
private readonly BlockingCollection<byte> _receiveBuffer = new BlockingCollection<byte>();
private readonly IMqttClientOptions _clientOptions;
private WebSocket4Net.WebSocket _webSocket;
public WebSocket4NetMqttChannel(IMqttClientOptions clientOptions)
{
_clientOptions = clientOptions ?? throw new ArgumentNullException(nameof(clientOptions));
}
public string Endpoint { get; } = "";
public Task ConnectAsync(CancellationToken cancellationToken)
{
var channelOptions = (MqttClientWebSocketOptions)_clientOptions.ChannelOptions;
var uri = "ws://" + channelOptions.Uri;
var sslProtocols = SslProtocols.None;
if (channelOptions.TlsOptions.UseTls)
{
uri = "wss://" + channelOptions.Uri;
sslProtocols = SslProtocols.Tls12;
}
var subProtocol = channelOptions.SubProtocols.FirstOrDefault() ?? string.Empty;
_webSocket = new WebSocket4Net.WebSocket(uri, subProtocol, sslProtocols: sslProtocols);
_webSocket.DataReceived += OnDataReceived;
_webSocket.Open();
SpinWait.SpinUntil(() => _webSocket.State == WebSocketState.Open, _clientOptions.CommunicationTimeout);
return Task.FromResult(0);
}
public Task DisconnectAsync()
{
if (_webSocket != null)
{
_webSocket.DataReceived -= OnDataReceived;
_webSocket.Close();
SpinWait.SpinUntil(() => _webSocket.State == WebSocketState.Closed, _clientOptions.CommunicationTimeout);
}
_webSocket = null;
return Task.FromResult(0);
}
public Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
var readBytes = 0;
while (count > 0 && !cancellationToken.IsCancellationRequested)
{
byte @byte;
if (readBytes == 0)
{
// Block until at lease one byte was received.
@byte = _receiveBuffer.Take(cancellationToken);
}
else
{
if (!_receiveBuffer.TryTake(out @byte))
{
return Task.FromResult(readBytes);
}
}
buffer[offset] = @byte;
offset++;
count--;
readBytes++;
}
return Task.FromResult(readBytes);
}
public Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
_webSocket.Send(buffer, offset, count);
return Task.FromResult(0);
}
public void Dispose()
{
if (_webSocket != null)
{
_webSocket.DataReceived -= OnDataReceived;
_webSocket.Dispose();
}
}
private void OnDataReceived(object sender, WebSocket4Net.DataReceivedEventArgs e)
{
foreach (var @byte in e.Data)
{
_receiveBuffer.Add(@byte);
}
}
}
}
}
Source/MQTTnet.Server/Mqtt/MqttServerService.cs
+6
View File
@@ -212,9 +212,15 @@ namespace MQTTnet.Server.Mqtt
// Configure encrypted connections
if (_settings.EncryptedTcpEndPoint.Enabled)
{
#if NETCOREAPP3_1 || NET5_0
options
.WithEncryptedEndpoint()
.WithEncryptionSslProtocol(SslProtocols.Tls13);
#else
options
.WithEncryptedEndpoint()
.WithEncryptionSslProtocol(SslProtocols.Tls12);
#endif
if (!string.IsNullOrEmpty(_settings.EncryptedTcpEndPoint?.Certificate?.Path))
{
Source/MQTTnet/Client/Disconnecting/MqttClientDisconnectedEventArgs.cs
+9 -4
View File
@@ -3,14 +3,16 @@ using MQTTnet.Client.Connecting;
namespace MQTTnet.Client.Disconnecting
{
public class MqttClientDisconnectedEventArgs : EventArgs
public sealed class MqttClientDisconnectedEventArgs : EventArgs
{
public MqttClientDisconnectedEventArgs(bool clientWasConnected, Exception exception, MqttClientAuthenticateResult authenticateResult, MqttClientDisconnectReason reasonCode)
public MqttClientDisconnectedEventArgs(bool clientWasConnected, Exception exception, MqttClientAuthenticateResult authenticateResult, MqttClientDisconnectReason reason)
{
ClientWasConnected = clientWasConnected;
Exception = exception;
AuthenticateResult = authenticateResult;
ReasonCode = reasonCode;
Reason = reason;
ReasonCode = reason;
}
public bool ClientWasConnected { get; }
@@ -19,6 +21,9 @@ namespace MQTTnet.Client.Disconnecting
public MqttClientAuthenticateResult AuthenticateResult { get; }
public MqttClientDisconnectReason ReasonCode { get; set; } = MqttClientDisconnectReason.NormalDisconnection;
public MqttClientDisconnectReason Reason { get; set; }
[Obsolete("Please use 'Reason' instead. This property will be removed in the future!")]
public MqttClientDisconnectReason ReasonCode { get; set; }
}
}
Source/MQTTnet/Client/Options/MqttClientOptionsBuilderTlsParameters.cs
+4
View File
@@ -19,7 +19,11 @@ namespace MQTTnet.Client.Options
public Func<MqttClientCertificateValidationCallbackContext, bool> CertificateValidationHandler { get; set; }
#if NETCOREAPP3_1 || NET5_0
public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls13;
#else
public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12;
#endif
#if WINDOWS_UWP
public IEnumerable<IEnumerable<byte>> Certificates { get; set; }
Source/MQTTnet/Client/Options/MqttClientTlsOptions.cs
+4
View File
@@ -26,7 +26,11 @@ namespace MQTTnet.Client.Options
public List<SslApplicationProtocol> ApplicationProtocols { get; set; }
#endif
#if NETCOREAPP3_1 || NET5_0
public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls13;
#else
public SslProtocols SslProtocol { get; set; } = SslProtocols.Tls12;
#endif
[Obsolete("This property will be removed soon. Use CertificateValidationHandler instead.")]
public Func<X509Certificate, X509Chain, SslPolicyErrors, IMqttClientOptions, bool> CertificateValidationCallback { get; set; }
Source/MQTTnet/Implementations/MqttTcpChannel.Uwp.cs
+12
View File
@@ -75,11 +75,23 @@ namespace MQTTnet.Implementations
_socket.Control.IgnorableServerCertificateErrors.Add(ignorableChainValidationResult);
}
#if NETCOREAPP3_1 || NET5_0
var socketProtectionLevel = SocketProtectionLevel.Tls13;
if (_options.TlsOptions.SslProtocol == SslProtocols.Tls12)
{
socketProtectionLevel = SocketProtectionLevel.Tls12;
}
else if (_options.TlsOptions.SslProtocol == SslProtocols.Tls11)
{
socketProtectionLevel = SocketProtectionLevel.Tls11;
}
#else
var socketProtectionLevel = SocketProtectionLevel.Tls12;
if (_options.TlsOptions.SslProtocol == SslProtocols.Tls11)
{
socketProtectionLevel = SocketProtectionLevel.Tls11;
}
#endif
else if (_options.TlsOptions.SslProtocol == SslProtocols.Tls)
{
socketProtectionLevel = SocketProtectionLevel.Tls10;
Source/MQTTnet/Implementations/MqttTcpChannel.cs
+1 -1
View File
@@ -88,7 +88,7 @@ namespace MQTTnet.Implementations
ApplicationProtocols = _options.TlsOptions.ApplicationProtocols,
ClientCertificates = LoadCertificates(),
EnabledSslProtocols = _options.TlsOptions.SslProtocol,
CertificateRevocationCheckMode = _options.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.Online : X509RevocationMode.NoCheck,
CertificateRevocationCheckMode = _options.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.NoCheck : X509RevocationMode.Online,
TargetHost = _options.Server
};
Source/MQTTnet/MqttApplicationMessageBuilder.cs
+12 -1
View File
@@ -1,4 +1,5 @@
using System.Collections.Generic;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
@@ -103,6 +104,16 @@ namespace MQTTnet
return this;
}
public MqttApplicationMessageBuilder WithQualityOfServiceLevel(int qualityOfServiceLevel)
{
if (qualityOfServiceLevel < 0 || qualityOfServiceLevel > 2)
{
throw new ArgumentOutOfRangeException(nameof(qualityOfServiceLevel));
}
return WithQualityOfServiceLevel((MqttQualityOfServiceLevel)qualityOfServiceLevel);
}
public MqttApplicationMessageBuilder WithRetainFlag(bool value = true)
{
_retain = value;
Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs
+6 -3
View File
@@ -68,10 +68,13 @@ namespace MQTTnet.Tests.Mockups
public IMqttClient CreateClient()
{
var client = _mqttFactory.CreateMqttClient(ClientLogger);
_clients.Add(client);
lock (_clients)
{
var client = _mqttFactory.CreateMqttClient(ClientLogger);
_clients.Add(client);
return new TestClientWrapper(client, TestContext);
return new TestClientWrapper(client, TestContext);
}
}
public Task<IMqttServer> StartServerAsync()
Tests/MQTTnet.Core.Tests/Server_Tests.cs
+10 -2
View File
@@ -469,11 +469,19 @@ namespace MQTTnet.Tests
var client1 = await testEnvironment.ConnectClientAsync(options);
await Task.Delay(500);
MqttClientDisconnectReason disconnectReason = MqttClientDisconnectReason.NormalDisconnection;
client1.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(c =>
{
disconnectReason = c.Reason;
});
var client2 = await testEnvironment.ConnectClientAsync(options);
await Task.Delay(500);
Assert.IsFalse(client1.IsConnected);
Assert.IsTrue(client2.IsConnected);
Assert.AreEqual(MqttClientDisconnectReason.SessionTakenOver, disconnectReason);
}
}
@@ -644,11 +652,11 @@ namespace MQTTnet.Tests
{
// Clear retained message.
await client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("r" + i2)
.WithPayload(new byte[0]).WithRetainFlag().Build());
.WithPayload(new byte[0]).WithRetainFlag().WithQualityOfServiceLevel(1).Build());
// Set retained message.
await client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("r" + i2)
.WithPayload("value").WithRetainFlag().Build());
.WithPayload("value").WithRetainFlag().WithQualityOfServiceLevel(1).Build());
await client.DisconnectAsync();
}