mirror of
https://github.com/dotnet/MQTTnet.git
synced 2026-10-15 17:00:03 +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).
|
||||
* [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).
|
||||
* [Client] Restored _Server_ and _Port_ behavior of client options (#2005).
|
||||
|
||||
@@ -3,7 +3,7 @@ name: CI
|
||||
on: [push, pull_request]
|
||||
|
||||
env:
|
||||
VERSION: "4.3.4.${{github.run_number}}"
|
||||
VERSION: "4.3.6.${{github.run_number}}"
|
||||
|
||||
jobs:
|
||||
build:
|
||||
|
||||
@@ -6,6 +6,7 @@ using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Diagnostics;
|
||||
using System.Linq;
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using System.Text;
|
||||
using System.Threading;
|
||||
@@ -86,7 +87,7 @@ namespace MQTTnet.Tests.Clients.MqttClient
|
||||
await client.ConnectAsync(clientOptions);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
[TestMethod]
|
||||
[ExpectedException(typeof(InvalidOperationException))]
|
||||
public async Task Connect_Multiple_Times_Should_Fail()
|
||||
@@ -748,8 +749,8 @@ namespace MQTTnet.Tests.Clients.MqttClient
|
||||
{
|
||||
if (e.ApplicationMessage.Topic == "request")
|
||||
{
|
||||
// Use AtMostOnce here because with QoS 1 or even QoS 2 the process waits for
|
||||
// the ACK etc. The problem is that the SpinUntil below only waits until the
|
||||
// Use AtMostOnce here because with QoS 1 or even QoS 2 the process waits for
|
||||
// the ACK etc. The problem is that the SpinUntil below only waits until the
|
||||
// flag is set. It does not wait until the client has sent the ACK
|
||||
await client2.PublishStringAsync("reply");
|
||||
}
|
||||
@@ -925,5 +926,54 @@ namespace MQTTnet.Tests.Clients.MqttClient
|
||||
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.
|
||||
|
||||
using System;
|
||||
using System.Net.Sockets;
|
||||
using System.Text;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
@@ -17,7 +18,7 @@ namespace MQTTnet.Tests.Internal
|
||||
[TestMethod]
|
||||
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);
|
||||
|
||||
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))]
|
||||
public async Task Try_Connect_Invalid_Host()
|
||||
{
|
||||
var crossPlatformSocket = new CrossPlatformSocket();
|
||||
var crossPlatformSocket = new CrossPlatformSocket(ProtocolType.Tcp);
|
||||
|
||||
var cancellationToken = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
cancellationToken.Token.Register(() => crossPlatformSocket.Dispose());
|
||||
@@ -65,7 +66,7 @@ namespace MQTTnet.Tests.Internal
|
||||
[TestMethod]
|
||||
public void Set_Options()
|
||||
{
|
||||
var crossPlatformSocket = new CrossPlatformSocket();
|
||||
var crossPlatformSocket = new CrossPlatformSocket(ProtocolType.Tcp);
|
||||
|
||||
Assert.IsFalse(crossPlatformSocket.ReuseAddress);
|
||||
crossPlatformSocket.ReuseAddress = true;
|
||||
|
||||
@@ -79,11 +79,26 @@ namespace MQTTnet.Client
|
||||
{
|
||||
if (_port.HasValue)
|
||||
{
|
||||
_remoteEndPoint = new DnsEndPoint(dns.Host, _port.Value);
|
||||
_remoteEndPoint = new DnsEndPoint(dns.Host, _port.Value, dns.AddressFamily);
|
||||
}
|
||||
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;
|
||||
}
|
||||
|
||||
[Obsolete("Use WithTcpServer(... configure) or WithWebSocketServer(... configure) instead.")]
|
||||
public MqttClientOptionsBuilder WithConnectionUri(Uri uri)
|
||||
{
|
||||
if (uri == null)
|
||||
@@ -196,7 +210,6 @@ namespace MQTTnet.Client
|
||||
return this;
|
||||
}
|
||||
|
||||
[Obsolete("Use WithTcpServer(... configure) or WithWebSocketServer(... configure) instead.")]
|
||||
public MqttClientOptionsBuilder WithConnectionUri(string uri)
|
||||
{
|
||||
return WithConnectionUri(new Uri(uri, UriKind.Absolute));
|
||||
@@ -354,13 +367,18 @@ namespace MQTTnet.Client
|
||||
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();
|
||||
|
||||
// The value 0 will be updated when building the options.
|
||||
// This a backward compatibility feature.
|
||||
_remoteEndPoint = new DnsEndPoint(server, port ?? 0);
|
||||
_remoteEndPoint = new DnsEndPoint(host, port ?? 0, addressFamily);
|
||||
_port = port;
|
||||
|
||||
return this;
|
||||
|
||||
@@ -2,13 +2,16 @@
|
||||
// The .NET Foundation licenses this file to you under the MIT license.
|
||||
// See the LICENSE file in the project root for more information.
|
||||
|
||||
using System;
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using MQTTnet.Protocol;
|
||||
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public sealed class MqttClientTcpOptions : IMqttClientChannelOptions
|
||||
{
|
||||
EndPoint _remoteEndpoint;
|
||||
public AddressFamily AddressFamily { get; set; } = AddressFamily.Unspecified;
|
||||
|
||||
public int BufferSize { get; set; } = 8192;
|
||||
@@ -20,6 +23,12 @@ namespace MQTTnet.Client
|
||||
/// </summary>
|
||||
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);
|
||||
|
||||
/// <summary>
|
||||
@@ -42,13 +51,56 @@ namespace MQTTnet.Client
|
||||
/// </summary>
|
||||
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 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
|
||||
}
|
||||
|
||||
public CrossPlatformSocket()
|
||||
public CrossPlatformSocket(ProtocolType protocolType)
|
||||
{
|
||||
// Having this constructor is important because avoiding the address family as parameter
|
||||
// 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
|
||||
_socketDisposeAction = _socket.Dispose;
|
||||
@@ -202,9 +202,28 @@ namespace MQTTnet.Implementations
|
||||
using (cancellationToken.Register(_socketDisposeAction))
|
||||
{
|
||||
#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
|
||||
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
|
||||
|
||||
@@ -16,6 +16,7 @@ using MQTTnet.Channel;
|
||||
using MQTTnet.Client;
|
||||
using MQTTnet.Exceptions;
|
||||
using MQTTnet.Internal;
|
||||
using MQTTnet.Protocol;
|
||||
|
||||
namespace MQTTnet.Implementations
|
||||
{
|
||||
@@ -63,7 +64,7 @@ namespace MQTTnet.Implementations
|
||||
{
|
||||
if (_tcpOptions.AddressFamily == AddressFamily.Unspecified)
|
||||
{
|
||||
socket = new CrossPlatformSocket();
|
||||
socket = new CrossPlatformSocket(_tcpOptions.ProtocolType);
|
||||
}
|
||||
else
|
||||
{
|
||||
@@ -98,6 +99,29 @@ namespace MQTTnet.Implementations
|
||||
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);
|
||||
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
|
||||
Reference in New Issue
Block a user