From 41d5b703441b9f2f1adf3887f22e51faafc4c58e Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Tue, 14 May 2024 18:09:33 +0200 Subject: [PATCH] Fix support for unix sockets (#1995) * Add support for changing the protocol type on socket level * Do not set NoDelay when not supported by protocol * Update release notes * Fix release notes --- .github/workflows/ReleaseNotes.md | 2 + Source/MQTTnet.Tests/MqttTcpChannel_Tests.cs | 4 +- .../MQTTnet.Tests/Server/Connection_Tests.cs | 4 +- .../Options/MqttClientOptionsBuilder.cs | 37 +++++++++++++------ .../Client/Options/MqttClientTcpOptions.cs | 12 ++++++ .../Implementations/CrossPlatformSocket.cs | 4 +- .../MQTTnet/Implementations/MqttTcpChannel.cs | 9 ++++- .../Implementations/MqttTcpServerListener.cs | 24 ++++++------ 8 files changed, 64 insertions(+), 32 deletions(-) diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index 53cbbce5..1016e556 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1,6 +1,8 @@ * [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). diff --git a/Source/MQTTnet.Tests/MqttTcpChannel_Tests.cs b/Source/MQTTnet.Tests/MqttTcpChannel_Tests.cs index efbca1fc..533cbb2b 100644 --- a/Source/MQTTnet.Tests/MqttTcpChannel_Tests.cs +++ b/Source/MQTTnet.Tests/MqttTcpChannel_Tests.cs @@ -19,7 +19,7 @@ namespace MQTTnet.Tests public async Task Dispose_Channel_While_Used() { var ct = new CancellationTokenSource(); - var serverSocket = new CrossPlatformSocket(AddressFamily.InterNetwork); + var serverSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp); try { @@ -38,7 +38,7 @@ namespace MQTTnet.Tests } }, ct.Token); - var clientSocket = new CrossPlatformSocket(AddressFamily.InterNetwork); + var clientSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp); await clientSocket.ConnectAsync("localhost", 50001, CancellationToken.None); var tcpChannel = new MqttTcpChannel(clientSocket.GetStream(), "test", null); diff --git a/Source/MQTTnet.Tests/Server/Connection_Tests.cs b/Source/MQTTnet.Tests/Server/Connection_Tests.cs index f4d5ea9d..ba17ae0b 100644 --- a/Source/MQTTnet.Tests/Server/Connection_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Connection_Tests.cs @@ -23,7 +23,7 @@ namespace MQTTnet.Tests.Server { await testEnvironment.StartServer(new MqttServerOptionsBuilder().WithDefaultCommunicationTimeout(TimeSpan.FromSeconds(1))); - var client = new CrossPlatformSocket(AddressFamily.InterNetwork); + var client = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp); await client.ConnectAsync("localhost", testEnvironment.ServerPort, CancellationToken.None); // Don't send anything. The server should close the connection. @@ -54,7 +54,7 @@ namespace MQTTnet.Tests.Server // Send an invalid packet and ensure that the server will close the connection and stay in a waiting state // forever. This is security related. - var client = new CrossPlatformSocket(AddressFamily.InterNetwork); + var client = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp); await client.ConnectAsync("localhost", testEnvironment.ServerPort, CancellationToken.None); var buffer = Encoding.UTF8.GetBytes("Garbage"); diff --git a/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs b/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs index 2935d582..f89f9f2d 100644 --- a/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs +++ b/Source/MQTTnet/Client/Options/MqttClientOptionsBuilder.cs @@ -8,6 +8,7 @@ using System; using System.Collections.Generic; using System.Linq; using System.Net; +using System.Net.Sockets; using System.Text; using MQTTnet.Formatter; using MQTTnet.Packets; @@ -18,14 +19,14 @@ namespace MQTTnet.Client public sealed class MqttClientOptionsBuilder { readonly MqttClientOptions _options = new MqttClientOptions(); + int? _port; [Obsolete] MqttClientWebSocketProxyOptions _proxyOptions; + EndPoint _remoteEndPoint; MqttClientTcpOptions _tcpOptions; MqttClientTlsOptions _tlsOptions; - EndPoint _remoteEndPoint; - int? _port; - + [Obsolete] MqttClientOptionsBuilderTlsParameters _tlsParameters; MqttClientWebSocketOptions _webSocketOptions; @@ -89,7 +90,7 @@ namespace MQTTnet.Client if (_tcpOptions.RemoteEndpoint == null) { - _tcpOptions.RemoteEndpoint = _remoteEndPoint; + _tcpOptions.RemoteEndpoint = _remoteEndPoint; } } else if (_webSocketOptions != null) @@ -114,6 +115,12 @@ namespace MQTTnet.Client return _options; } + public MqttClientOptionsBuilder WithAddressFamily(AddressFamily addressFamily) + { + _tcpOptions.AddressFamily = addressFamily; + return this; + } + public MqttClientOptionsBuilder WithAuthentication(string method, byte[] data) { _options.AuthenticationMethod = method; @@ -218,6 +225,14 @@ namespace MQTTnet.Client return this; } + public MqttClientOptionsBuilder WithEndPoint(EndPoint endPoint) + { + _remoteEndPoint = endPoint ?? throw new ArgumentNullException(nameof(endPoint)); + _tcpOptions = new MqttClientTcpOptions(); + + return this; + } + public MqttClientOptionsBuilder WithExtendedAuthenticationExchangeHandler(IMqttExtendedAuthenticationExchangeHandler handler) { _options.ExtendedAuthenticationExchangeHandler = handler; @@ -263,6 +278,12 @@ namespace MQTTnet.Client return this; } + public MqttClientOptionsBuilder WithProtocolType(ProtocolType protocolType) + { + _tcpOptions.ProtocolType = protocolType; + return this; + } + public MqttClientOptionsBuilder WithProtocolVersion(MqttProtocolVersion value) { if (value == MqttProtocolVersion.Unknown) @@ -344,14 +365,6 @@ namespace MQTTnet.Client return this; } - - public MqttClientOptionsBuilder WithEndPoint(EndPoint endPoint) - { - _remoteEndPoint = endPoint ?? throw new ArgumentNullException(nameof(endPoint)); - _tcpOptions = new MqttClientTcpOptions(); - - return this; - } public MqttClientOptionsBuilder WithTcpServer(Action optionsBuilder) { diff --git a/Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs b/Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs index 09e652b5..59483e89 100644 --- a/Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs +++ b/Source/MQTTnet/Client/Options/MqttClientTcpOptions.cs @@ -28,8 +28,20 @@ namespace MQTTnet.Client /// public EndPoint LocalEndpoint { get; set; } + /// + /// Enables or disables the Nagle algorithm for the socket. + /// This is only supported for TCP. + /// For other protocol types the value is ignored. + /// Default: true + /// public bool NoDelay { get; set; } = true; + /// + /// The MQTT transport is usually TCP but when using other endpoint types like + /// unix sockets it must be changed (IP for unix sockets). + /// + public ProtocolType ProtocolType { get; set; } = ProtocolType.Tcp; + public EndPoint RemoteEndpoint { get; set; } public MqttClientTlsOptions TlsOptions { get; set; } = new MqttClientTlsOptions(); diff --git a/Source/MQTTnet/Implementations/CrossPlatformSocket.cs b/Source/MQTTnet/Implementations/CrossPlatformSocket.cs index c6a8df54..c3d29c16 100644 --- a/Source/MQTTnet/Implementations/CrossPlatformSocket.cs +++ b/Source/MQTTnet/Implementations/CrossPlatformSocket.cs @@ -22,9 +22,9 @@ namespace MQTTnet.Implementations NetworkStream _networkStream; - public CrossPlatformSocket(AddressFamily addressFamily) + public CrossPlatformSocket(AddressFamily addressFamily, ProtocolType protocolType) { - _socket = new Socket(addressFamily, SocketType.Stream, ProtocolType.Tcp); + _socket = new Socket(addressFamily, SocketType.Stream, protocolType); #if !NET5_0_OR_GREATER _socketDisposeAction = _socket.Dispose; diff --git a/Source/MQTTnet/Implementations/MqttTcpChannel.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.cs index c70f3903..a5108e58 100644 --- a/Source/MQTTnet/Implementations/MqttTcpChannel.cs +++ b/Source/MQTTnet/Implementations/MqttTcpChannel.cs @@ -67,7 +67,7 @@ namespace MQTTnet.Implementations } else { - socket = new CrossPlatformSocket(_tcpOptions.AddressFamily); + socket = new CrossPlatformSocket(_tcpOptions.AddressFamily, _tcpOptions.ProtocolType); } if (_tcpOptions.LocalEndpoint != null) @@ -78,7 +78,12 @@ namespace MQTTnet.Implementations socket.ReceiveBufferSize = _tcpOptions.BufferSize; socket.SendBufferSize = _tcpOptions.BufferSize; socket.SendTimeout = (int)_clientOptions.Timeout.TotalMilliseconds; - socket.NoDelay = _tcpOptions.NoDelay; + + if (_tcpOptions.ProtocolType == ProtocolType.Tcp) + { + // Other protocol types do not support the Nagle algorithm. + socket.NoDelay = _tcpOptions.NoDelay; + } if (socket.LingerState != null) { diff --git a/Source/MQTTnet/Implementations/MqttTcpServerListener.cs b/Source/MQTTnet/Implementations/MqttTcpServerListener.cs index 48cbb25c..bb008afb 100644 --- a/Source/MQTTnet/Implementations/MqttTcpServerListener.cs +++ b/Source/MQTTnet/Implementations/MqttTcpServerListener.cs @@ -27,7 +27,7 @@ namespace MQTTnet.Implementations readonly MqttServerOptions _serverOptions; readonly MqttServerTcpEndpointBaseOptions _options; readonly MqttServerTlsTcpEndpointOptions _tlsOptions; - + CrossPlatformSocket _socket; IPEndPoint _localEndPoint; @@ -65,7 +65,7 @@ namespace MQTTnet.Implementations _logger.Info("Starting TCP listener (Endpoint={0}, TLS={1})", _localEndPoint, _tlsOptions?.CertificateProvider != null); - _socket = new CrossPlatformSocket(_addressFamily); + _socket = new CrossPlatformSocket(_addressFamily, ProtocolType.Tcp); // Usage of socket options is described here: https://docs.microsoft.com/en-us/dotnet/api/system.net.sockets.socket.setsocketoption?view=netcore-2.2 if (_options.ReuseAddress) @@ -87,33 +87,33 @@ namespace MQTTnet.Implementations { _socket.KeepAlive = _options.KeepAlive.Value; } - + if (_options.TcpKeepAliveInterval.HasValue) { _socket.TcpKeepAliveInterval = _options.TcpKeepAliveInterval.Value; } - + if (_options.TcpKeepAliveRetryCount.HasValue) { _socket.TcpKeepAliveInterval = _options.TcpKeepAliveRetryCount.Value; } - + if (_options.TcpKeepAliveTime.HasValue) { _socket.TcpKeepAliveTime = _options.TcpKeepAliveTime.Value; } - + _socket.Bind(_localEndPoint); - + // Get the local endpoint back from the socket. The port may have changed. // This can happen when port 0 is used. Then the OS will choose the next free port. _localEndPoint = (IPEndPoint)_socket.LocalEndPoint; _options.Port = _localEndPoint.Port; - + _socket.Listen(_options.ConnectionBacklog); - + _logger.Verbose("TCP listener started (Endpoint={0})", _localEndPoint); - + Task.Run(() => AcceptClientConnectionsAsync(cancellationToken), cancellationToken).RunInBackground(_logger); return true; @@ -183,7 +183,7 @@ namespace MQTTnet.Implementations clientSocket.NoDelay = _options.NoDelay; stream = clientSocket.GetStream(); var clientCertificate = _tlsOptions?.CertificateProvider?.GetCertificate(); - + if (clientCertificate != null) { if (!clientCertificate.HasPrivateKey) @@ -228,7 +228,7 @@ namespace MQTTnet.Implementations var tcpChannel = new MqttTcpChannel(stream, remoteEndPoint, clientCertificate); var bufferWriter = new MqttBufferWriter(_serverOptions.WriterBufferSize, _serverOptions.WriterBufferSizeMax); var packetFormatterAdapter = new MqttPacketFormatterAdapter(bufferWriter); - + using (var clientAdapter = new MqttChannelAdapter(tcpChannel, packetFormatterAdapter, _rootLogger)) { clientAdapter.AllowPacketFragmentation = _options.AllowPacketFragmentation;