mirror of
https://github.com/dotnet/MQTTnet.git
synced 2026-10-21 04:03:05 +08:00
Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
25c1da6a21 | ||
|
|
4e3ede42c6 | ||
|
|
918a3c8d8b | ||
|
|
290419b3b6 | ||
|
|
32b6146532 | ||
|
|
3a73ec21f7 |
@@ -1,13 +1 @@
|
|||||||
* [Core] Optimized packet serialization of PUBACK and PUBREC packets for protocol version 5.0.0 (#1939, thanks to @Y-Sindo).
|
* [Client] Restored _Server_ and _Port_ behavior of client options (#2005).
|
||||||
* [Core] The package inspector is now fully async (#1941).
|
|
||||||
* [Core] Fixed decoding of DISCONNECT packet with empty body (#1994, thanks to @Y-Sindo).
|
|
||||||
* [Client] Exposed the _EndPoint_ type to support other endpoint types (like Unix Domain Sockets) in client options (#1919).
|
|
||||||
* [Client] Fixed support for unix sockets by exposing more options (#1995).
|
|
||||||
* [Client] Added a dedicated exception when the client is not connected (#1954, thanks to @marcpiulachs).
|
|
||||||
* [Client] The client will now throw a _MqttClientUnexpectedDisconnectReceivedException_ when publishing a QoS 0 message which leads to a server disconnect (BREAKING CHANGE!, #1974, thanks to @fazho).
|
|
||||||
* [Client] Exposed the certificate selection event handler in client options (#1984).
|
|
||||||
* [Server] The server will no longer send _NoMatchingSubscribers_ when the actual subscription was non success (#1965, BREAKING CHANGE!).
|
|
||||||
* [Server] Fixed broken support for _null_ in _AddServer_ method in ASP.NET integration (#1981).
|
|
||||||
* [ManagedClient] Added a new event (SubscriptionsChangedAsync) which is fired when a subscription or unsubscription was made (#1894, thanks to @pdufrene).
|
|
||||||
* [ManagedClient] Fixed race condition when server shuts down while subscribing (#1987, thanks to @marve).
|
|
||||||
* [TopicTemplate] Added new extension which provides a template engine for topics (#1932, thanks to @simonthum).
|
|
||||||
|
|||||||
@@ -3,7 +3,7 @@ name: CI
|
|||||||
on: [push, pull_request]
|
on: [push, pull_request]
|
||||||
|
|
||||||
env:
|
env:
|
||||||
VERSION: "4.3.4.${{github.run_number}}"
|
VERSION: "4.3.6.${{github.run_number}}"
|
||||||
|
|
||||||
jobs:
|
jobs:
|
||||||
build:
|
build:
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ using System;
|
|||||||
using System.Collections.Generic;
|
using System.Collections.Generic;
|
||||||
using System.Diagnostics;
|
using System.Diagnostics;
|
||||||
using System.Linq;
|
using System.Linq;
|
||||||
|
using System.Net;
|
||||||
using System.Net.Sockets;
|
using System.Net.Sockets;
|
||||||
using System.Text;
|
using System.Text;
|
||||||
using System.Threading;
|
using System.Threading;
|
||||||
@@ -925,5 +926,54 @@ namespace MQTTnet.Tests.Clients.MqttClient
|
|||||||
Assert.IsFalse(disconnectedFired);
|
Assert.IsFalse(disconnectedFired);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[TestMethod]
|
||||||
|
public void Backward_compatible_TCP_options()
|
||||||
|
{
|
||||||
|
var options = new MqttClientOptionsBuilder().WithTcpServer("host", 3).Build();
|
||||||
|
|
||||||
|
Assert.AreEqual("host", ((MqttClientTcpOptions)options.ChannelOptions).Server);
|
||||||
|
Assert.AreEqual(3, ((MqttClientTcpOptions)options.ChannelOptions).Port);
|
||||||
|
|
||||||
|
options = new MqttClientOptions
|
||||||
|
{
|
||||||
|
ChannelOptions = new MqttClientTcpOptions
|
||||||
|
{
|
||||||
|
Server = "host",
|
||||||
|
Port = 3
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
Assert.AreEqual("host:3", options.ChannelOptions.ToString());
|
||||||
|
Assert.AreEqual("host", ((MqttClientTcpOptions)options.ChannelOptions).Server);
|
||||||
|
Assert.AreEqual(3, ((MqttClientTcpOptions)options.ChannelOptions).Port);
|
||||||
|
|
||||||
|
options = new MqttClientOptionsBuilder().WithEndPoint(new DnsEndPoint("host", 3)).Build();
|
||||||
|
Assert.AreEqual("Unspecified/host:3", options.ChannelOptions.ToString());
|
||||||
|
Assert.AreEqual("host", ((MqttClientTcpOptions)options.ChannelOptions).Server);
|
||||||
|
Assert.AreEqual(3, ((MqttClientTcpOptions)options.ChannelOptions).Port);
|
||||||
|
|
||||||
|
options = new MqttClientOptionsBuilder().WithTcpServer("host").Build();
|
||||||
|
|
||||||
|
Assert.AreEqual("host", ((MqttClientTcpOptions)options.ChannelOptions).Server);
|
||||||
|
Assert.AreEqual(1883, ((MqttClientTcpOptions)options.ChannelOptions).Port);
|
||||||
|
Assert.AreEqual("Unspecified/host:1883", options.ChannelOptions.ToString());
|
||||||
|
|
||||||
|
options = new MqttClientOptionsBuilder().WithTlsOptions(o => o.UseTls()).WithTcpServer("host").Build();
|
||||||
|
|
||||||
|
Assert.AreEqual("host", ((MqttClientTcpOptions)options.ChannelOptions).Server);
|
||||||
|
Assert.AreEqual(8883, ((MqttClientTcpOptions)options.ChannelOptions).Port);
|
||||||
|
|
||||||
|
options = new MqttClientOptions
|
||||||
|
{
|
||||||
|
ChannelOptions = new MqttClientTcpOptions
|
||||||
|
{
|
||||||
|
Server = "host"
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
Assert.AreEqual("host", ((MqttClientTcpOptions)options.ChannelOptions).Server);
|
||||||
|
Assert.AreEqual(null, ((MqttClientTcpOptions)options.ChannelOptions).Port);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -3,6 +3,7 @@
|
|||||||
// See the LICENSE file in the project root for more information.
|
// See the LICENSE file in the project root for more information.
|
||||||
|
|
||||||
using System;
|
using System;
|
||||||
|
using System.Net.Sockets;
|
||||||
using System.Text;
|
using System.Text;
|
||||||
using System.Threading;
|
using System.Threading;
|
||||||
using System.Threading.Tasks;
|
using System.Threading.Tasks;
|
||||||
@@ -17,7 +18,7 @@ namespace MQTTnet.Tests.Internal
|
|||||||
[TestMethod]
|
[TestMethod]
|
||||||
public async Task Connect_Send_Receive()
|
public async Task Connect_Send_Receive()
|
||||||
{
|
{
|
||||||
var crossPlatformSocket = new CrossPlatformSocket();
|
var crossPlatformSocket = new CrossPlatformSocket(ProtocolType.Tcp);
|
||||||
await crossPlatformSocket.ConnectAsync("www.google.de", 80, CancellationToken.None);
|
await crossPlatformSocket.ConnectAsync("www.google.de", 80, CancellationToken.None);
|
||||||
|
|
||||||
var requestBuffer = Encoding.UTF8.GetBytes("GET / HTTP/1.1\r\nHost: www.google.de\r\n\r\n");
|
var requestBuffer = Encoding.UTF8.GetBytes("GET / HTTP/1.1\r\nHost: www.google.de\r\n\r\n");
|
||||||
@@ -36,7 +37,7 @@ namespace MQTTnet.Tests.Internal
|
|||||||
[ExpectedException(typeof(OperationCanceledException))]
|
[ExpectedException(typeof(OperationCanceledException))]
|
||||||
public async Task Try_Connect_Invalid_Host()
|
public async Task Try_Connect_Invalid_Host()
|
||||||
{
|
{
|
||||||
var crossPlatformSocket = new CrossPlatformSocket();
|
var crossPlatformSocket = new CrossPlatformSocket(ProtocolType.Tcp);
|
||||||
|
|
||||||
var cancellationToken = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
var cancellationToken = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||||
cancellationToken.Token.Register(() => crossPlatformSocket.Dispose());
|
cancellationToken.Token.Register(() => crossPlatformSocket.Dispose());
|
||||||
@@ -65,7 +66,7 @@ namespace MQTTnet.Tests.Internal
|
|||||||
[TestMethod]
|
[TestMethod]
|
||||||
public void Set_Options()
|
public void Set_Options()
|
||||||
{
|
{
|
||||||
var crossPlatformSocket = new CrossPlatformSocket();
|
var crossPlatformSocket = new CrossPlatformSocket(ProtocolType.Tcp);
|
||||||
|
|
||||||
Assert.IsFalse(crossPlatformSocket.ReuseAddress);
|
Assert.IsFalse(crossPlatformSocket.ReuseAddress);
|
||||||
crossPlatformSocket.ReuseAddress = true;
|
crossPlatformSocket.ReuseAddress = true;
|
||||||
|
|||||||
@@ -79,11 +79,26 @@ namespace MQTTnet.Client
|
|||||||
{
|
{
|
||||||
if (_port.HasValue)
|
if (_port.HasValue)
|
||||||
{
|
{
|
||||||
_remoteEndPoint = new DnsEndPoint(dns.Host, _port.Value);
|
_remoteEndPoint = new DnsEndPoint(dns.Host, _port.Value, dns.AddressFamily);
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
_remoteEndPoint = new DnsEndPoint(dns.Host, tlsOptions?.UseTls == false ? MqttPorts.Default : MqttPorts.Secure);
|
_remoteEndPoint = new DnsEndPoint(dns.Host, tlsOptions?.UseTls == false ? MqttPorts.Default : MqttPorts.Secure, dns.AddressFamily);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (_remoteEndPoint is IPEndPoint ip)
|
||||||
|
{
|
||||||
|
if (ip.Port == 0)
|
||||||
|
{
|
||||||
|
if (_port.HasValue)
|
||||||
|
{
|
||||||
|
_remoteEndPoint = new IPEndPoint(ip.Address, _port.Value);
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
_remoteEndPoint = new IPEndPoint(ip.Address, tlsOptions?.UseTls == false ? MqttPorts.Default : MqttPorts.Secure);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -152,7 +167,6 @@ namespace MQTTnet.Client
|
|||||||
return this;
|
return this;
|
||||||
}
|
}
|
||||||
|
|
||||||
[Obsolete("Use WithTcpServer(... configure) or WithWebSocketServer(... configure) instead.")]
|
|
||||||
public MqttClientOptionsBuilder WithConnectionUri(Uri uri)
|
public MqttClientOptionsBuilder WithConnectionUri(Uri uri)
|
||||||
{
|
{
|
||||||
if (uri == null)
|
if (uri == null)
|
||||||
@@ -196,7 +210,6 @@ namespace MQTTnet.Client
|
|||||||
return this;
|
return this;
|
||||||
}
|
}
|
||||||
|
|
||||||
[Obsolete("Use WithTcpServer(... configure) or WithWebSocketServer(... configure) instead.")]
|
|
||||||
public MqttClientOptionsBuilder WithConnectionUri(string uri)
|
public MqttClientOptionsBuilder WithConnectionUri(string uri)
|
||||||
{
|
{
|
||||||
return WithConnectionUri(new Uri(uri, UriKind.Absolute));
|
return WithConnectionUri(new Uri(uri, UriKind.Absolute));
|
||||||
@@ -354,13 +367,18 @@ namespace MQTTnet.Client
|
|||||||
return this;
|
return this;
|
||||||
}
|
}
|
||||||
|
|
||||||
public MqttClientOptionsBuilder WithTcpServer(string server, int? port = null)
|
public MqttClientOptionsBuilder WithTcpServer(string host, int? port = null, AddressFamily addressFamily = AddressFamily.Unspecified)
|
||||||
{
|
{
|
||||||
|
if (host == null)
|
||||||
|
{
|
||||||
|
throw new ArgumentNullException(nameof(host));
|
||||||
|
}
|
||||||
|
|
||||||
_tcpOptions = new MqttClientTcpOptions();
|
_tcpOptions = new MqttClientTcpOptions();
|
||||||
|
|
||||||
// The value 0 will be updated when building the options.
|
// The value 0 will be updated when building the options.
|
||||||
// This a backward compatibility feature.
|
// This a backward compatibility feature.
|
||||||
_remoteEndPoint = new DnsEndPoint(server, port ?? 0);
|
_remoteEndPoint = new DnsEndPoint(host, port ?? 0, addressFamily);
|
||||||
_port = port;
|
_port = port;
|
||||||
|
|
||||||
return this;
|
return this;
|
||||||
|
|||||||
@@ -2,13 +2,16 @@
|
|||||||
// The .NET Foundation licenses this file to you under the MIT license.
|
// The .NET Foundation licenses this file to you under the MIT license.
|
||||||
// See the LICENSE file in the project root for more information.
|
// See the LICENSE file in the project root for more information.
|
||||||
|
|
||||||
|
using System;
|
||||||
using System.Net;
|
using System.Net;
|
||||||
using System.Net.Sockets;
|
using System.Net.Sockets;
|
||||||
|
using MQTTnet.Protocol;
|
||||||
|
|
||||||
namespace MQTTnet.Client
|
namespace MQTTnet.Client
|
||||||
{
|
{
|
||||||
public sealed class MqttClientTcpOptions : IMqttClientChannelOptions
|
public sealed class MqttClientTcpOptions : IMqttClientChannelOptions
|
||||||
{
|
{
|
||||||
|
EndPoint _remoteEndpoint;
|
||||||
public AddressFamily AddressFamily { get; set; } = AddressFamily.Unspecified;
|
public AddressFamily AddressFamily { get; set; } = AddressFamily.Unspecified;
|
||||||
|
|
||||||
public int BufferSize { get; set; } = 8192;
|
public int BufferSize { get; set; } = 8192;
|
||||||
@@ -20,6 +23,12 @@ namespace MQTTnet.Client
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
public bool? DualMode { get; set; }
|
public bool? DualMode { get; set; }
|
||||||
|
|
||||||
|
[Obsolete("Use RemoteEndpoint or MqttClientOptionsBuilder instead.")]
|
||||||
|
public int? Port { get; set; }
|
||||||
|
|
||||||
|
[Obsolete("Use RemoteEndpoint or MqttClientOptionsBuilder instead.")]
|
||||||
|
public string Server { get; set; }
|
||||||
|
|
||||||
public LingerOption LingerState { get; set; } = new LingerOption(true, 0);
|
public LingerOption LingerState { get; set; } = new LingerOption(true, 0);
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
@@ -42,13 +51,56 @@ namespace MQTTnet.Client
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
public ProtocolType ProtocolType { get; set; } = ProtocolType.Tcp;
|
public ProtocolType ProtocolType { get; set; } = ProtocolType.Tcp;
|
||||||
|
|
||||||
public EndPoint RemoteEndpoint { get; set; }
|
public EndPoint RemoteEndpoint
|
||||||
|
{
|
||||||
|
get => _remoteEndpoint;
|
||||||
|
set
|
||||||
|
{
|
||||||
|
_remoteEndpoint = value;
|
||||||
|
|
||||||
|
if (_remoteEndpoint is DnsEndPoint dnsEndPoint)
|
||||||
|
{
|
||||||
|
Server = dnsEndPoint.Host;
|
||||||
|
Port = dnsEndPoint.Port;
|
||||||
|
}
|
||||||
|
else if (_remoteEndpoint is IPEndPoint ipEndPoint)
|
||||||
|
{
|
||||||
|
Server = ipEndPoint.Address.ToString();
|
||||||
|
Port = ipEndPoint.Port;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
|
public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions();
|
||||||
|
|
||||||
public override string ToString()
|
public override string ToString()
|
||||||
{
|
{
|
||||||
return RemoteEndpoint?.ToString() ?? string.Empty;
|
if (RemoteEndpoint != null)
|
||||||
|
{
|
||||||
|
return RemoteEndpoint.ToString();
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!string.IsNullOrEmpty(Server))
|
||||||
|
{
|
||||||
|
return $"{Server}:{GetPort()}";
|
||||||
|
}
|
||||||
|
|
||||||
|
return string.Empty;
|
||||||
|
}
|
||||||
|
|
||||||
|
int GetPort()
|
||||||
|
{
|
||||||
|
if (Port.HasValue)
|
||||||
|
{
|
||||||
|
return Port.Value;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (TlsOptions?.UseTls == true)
|
||||||
|
{
|
||||||
|
return MqttPorts.Secure;
|
||||||
|
}
|
||||||
|
|
||||||
|
return MqttPorts.Default;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -31,11 +31,11 @@ namespace MQTTnet.Implementations
|
|||||||
#endif
|
#endif
|
||||||
}
|
}
|
||||||
|
|
||||||
public CrossPlatformSocket()
|
public CrossPlatformSocket(ProtocolType protocolType)
|
||||||
{
|
{
|
||||||
// Having this constructor is important because avoiding the address family as parameter
|
// Having this constructor is important because avoiding the address family as parameter
|
||||||
// will make use of dual mode in the .net framework.
|
// will make use of dual mode in the .net framework.
|
||||||
_socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
|
_socket = new Socket(SocketType.Stream, protocolType);
|
||||||
|
|
||||||
#if !NET5_0_OR_GREATER
|
#if !NET5_0_OR_GREATER
|
||||||
_socketDisposeAction = _socket.Dispose;
|
_socketDisposeAction = _socket.Dispose;
|
||||||
@@ -202,9 +202,28 @@ namespace MQTTnet.Implementations
|
|||||||
using (cancellationToken.Register(_socketDisposeAction))
|
using (cancellationToken.Register(_socketDisposeAction))
|
||||||
{
|
{
|
||||||
#if NET452 || NET461
|
#if NET452 || NET461
|
||||||
await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, endPoint, null).ConfigureAwait(false);
|
// This is a fix for Mono which behaves differently than dotnet.
|
||||||
|
// The connection will not be established when the DNS endpoint is used.
|
||||||
|
if (endPoint is DnsEndPoint dns && dns.AddressFamily == AddressFamily.Unspecified)
|
||||||
|
{
|
||||||
|
await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, dns.Host, dns.Port, null).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
await Task.Factory.FromAsync(_socket.BeginConnect, _socket.EndConnect, endPoint, null).ConfigureAwait(false);
|
||||||
|
}
|
||||||
#else
|
#else
|
||||||
await _socket.ConnectAsync(endPoint).ConfigureAwait(false);
|
|
||||||
|
// This is a fix for Mono which behaves differently than dotnet.
|
||||||
|
// The connection will not be established when the DNS endpoint is used.
|
||||||
|
if (endPoint is DnsEndPoint dns && dns.AddressFamily == AddressFamily.Unspecified)
|
||||||
|
{
|
||||||
|
await _socket.ConnectAsync(dns.Host, dns.Port).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
await _socket.ConnectAsync(endPoint).ConfigureAwait(false);
|
||||||
|
}
|
||||||
#endif
|
#endif
|
||||||
}
|
}
|
||||||
#endif
|
#endif
|
||||||
|
|||||||
@@ -16,6 +16,7 @@ using MQTTnet.Channel;
|
|||||||
using MQTTnet.Client;
|
using MQTTnet.Client;
|
||||||
using MQTTnet.Exceptions;
|
using MQTTnet.Exceptions;
|
||||||
using MQTTnet.Internal;
|
using MQTTnet.Internal;
|
||||||
|
using MQTTnet.Protocol;
|
||||||
|
|
||||||
namespace MQTTnet.Implementations
|
namespace MQTTnet.Implementations
|
||||||
{
|
{
|
||||||
@@ -63,7 +64,7 @@ namespace MQTTnet.Implementations
|
|||||||
{
|
{
|
||||||
if (_tcpOptions.AddressFamily == AddressFamily.Unspecified)
|
if (_tcpOptions.AddressFamily == AddressFamily.Unspecified)
|
||||||
{
|
{
|
||||||
socket = new CrossPlatformSocket();
|
socket = new CrossPlatformSocket(_tcpOptions.ProtocolType);
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
@@ -98,6 +99,29 @@ namespace MQTTnet.Implementations
|
|||||||
socket.DualMode = _tcpOptions.DualMode.Value;
|
socket.DualMode = _tcpOptions.DualMode.Value;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// This block is only for backward compatibility.
|
||||||
|
if (_tcpOptions.RemoteEndpoint == null && !string.IsNullOrEmpty(_tcpOptions.Server))
|
||||||
|
{
|
||||||
|
int port;
|
||||||
|
if (_tcpOptions.Port.HasValue)
|
||||||
|
{
|
||||||
|
port = _tcpOptions.Port.Value;
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
if (_tcpOptions.TlsOptions?.UseTls == true)
|
||||||
|
{
|
||||||
|
port = MqttPorts.Secure;
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
port = MqttPorts.Default;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
_tcpOptions.RemoteEndpoint = new DnsEndPoint(_tcpOptions.Server, port, AddressFamily.Unspecified);
|
||||||
|
}
|
||||||
|
|
||||||
await socket.ConnectAsync(_tcpOptions.RemoteEndpoint, cancellationToken).ConfigureAwait(false);
|
await socket.ConnectAsync(_tcpOptions.RemoteEndpoint, cancellationToken).ConfigureAwait(false);
|
||||||
|
|
||||||
cancellationToken.ThrowIfCancellationRequested();
|
cancellationToken.ThrowIfCancellationRequested();
|
||||||
|
|||||||
Reference in New Issue
Block a user