diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index 8ca95868..fcc18ffa 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -3,3 +3,4 @@ * [Client] MQTTv5 features are now checked and an exception is thrown if they are used when using protocol version 3.1.1 and lower. These checks can be disabled in client options. (BREAKING CHANGE!). * [Server] Exposed MQTT v5 sent properties from the affected client in _ClientDisconnectedAsync_ event. * [Server] Fixed wrong client ID passed to _InterceptingUnsubscriptionEventArgs_ (#1631, thanks to @ghord). +* [Server] Exposed socket settings for TCP keep alive in TCP options (#1544). diff --git a/Source/MQTTnet/Implementations/CrossPlatformSocket.cs b/Source/MQTTnet/Implementations/CrossPlatformSocket.cs index d7ae1f04..ea642431 100644 --- a/Source/MQTTnet/Implementations/CrossPlatformSocket.cs +++ b/Source/MQTTnet/Implementations/CrossPlatformSocket.cs @@ -72,8 +72,8 @@ namespace MQTTnet.Implementations get => _socket.GetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveInterval) as int? ?? 0; set => _socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveInterval, value); #else - get { throw new NotSupportedException("TcpKeepAliveInterval requires at least net5.0."); } - set { throw new NotSupportedException("TcpKeepAliveInterval requires at least net5.0."); } + get { throw new NotSupportedException("TcpKeepAliveInterval requires at least netcoreapp3.0."); } + set { throw new NotSupportedException("TcpKeepAliveInterval requires at least netcoreapp3.0."); } #endif } @@ -83,8 +83,8 @@ namespace MQTTnet.Implementations get => _socket.GetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveRetryCount) as int? ?? 0; set => _socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveRetryCount, value); #else - get { throw new NotSupportedException("TcpKeepAliveRetryCount requires at least net5.0."); } - set { throw new NotSupportedException("TcpKeepAliveRetryCount requires at least net5.0."); } + get { throw new NotSupportedException("TcpKeepAliveRetryCount requires at least netcoreapp3.0."); } + set { throw new NotSupportedException("TcpKeepAliveRetryCount requires at least netcoreapp3.0."); } #endif } @@ -94,8 +94,8 @@ namespace MQTTnet.Implementations get => _socket.GetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveTime) as int? ?? 0; set => _socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveTime, value); #else - get { throw new NotSupportedException("TcpKeepAliveTime requires at least net5.0."); } - set { throw new NotSupportedException("TcpKeepAliveTime requires at least net5.0."); } + get { throw new NotSupportedException("TcpKeepAliveTime requires at least netcoreapp3.0."); } + set { throw new NotSupportedException("TcpKeepAliveTime requires at least netcoreapp3.0."); } #endif } @@ -155,7 +155,7 @@ namespace MQTTnet.Implementations } catch (ObjectDisposedException) { - // This will happen when _socket.EndAccept gets called by Task library but the socket is already disposed. + // This will happen when _socket.EndAccept_ gets called by Task library but the socket is already disposed. return null; } } @@ -181,16 +181,18 @@ namespace MQTTnet.Implementations try { -#if NET5_0_OR_GREATER +#if NETCOREAPP3_0_OR_GREATER if (_networkStream != null) { await _networkStream.DisposeAsync().ConfigureAwait(false); } - - await _socket.ConnectAsync(host, port, cancellationToken).ConfigureAwait(false); #else _networkStream?.Dispose(); +#endif +#if NET5_0_OR_GREATER + await _socket.ConnectAsync(host, port, cancellationToken).ConfigureAwait(false); +#else // Workaround for: https://github.com/dotnet/corefx/issues/24430 using (cancellationToken.Register(_socketDisposeAction)) { @@ -219,7 +221,7 @@ namespace MQTTnet.Implementations } catch (ObjectDisposedException) { - // This will happen when _socket.EndConnect gets called by Task library but the socket is already disposed. + // This will happen when _socket.EndConnect_ gets called by Task library but the socket is already disposed. cancellationToken.ThrowIfCancellationRequested(); } } @@ -246,38 +248,44 @@ namespace MQTTnet.Implementations _socket.Listen(connectionBacklog); } +#if NET452 || NET461 public async Task ReceiveAsync(ArraySegment buffer, SocketFlags socketFlags) { try { -#if NET452 || NET461 return await Task.Factory.FromAsync(SocketWrapper.BeginReceive, _socket.EndReceive, new SocketWrapper(_socket, buffer, socketFlags)).ConfigureAwait(false); -#else - return await _socket.ReceiveAsync(buffer, socketFlags).ConfigureAwait(false); -#endif } catch (ObjectDisposedException) { - // This will happen when _socket.EndReceive gets called by Task library but the socket is already disposed. + // This will happen when _socket.EndReceive_ gets called by Task library but the socket is already disposed. return -1; } } +#else + public Task ReceiveAsync(ArraySegment buffer, SocketFlags socketFlags) + { + return _socket.ReceiveAsync(buffer, socketFlags); + } +#endif +#if NET452 || NET461 public async Task SendAsync(ArraySegment buffer, SocketFlags socketFlags) { try { -#if NET452 || NET461 await Task.Factory.FromAsync(SocketWrapper.BeginSend, _socket.EndSend, new SocketWrapper(_socket, buffer, socketFlags)).ConfigureAwait(false); -#else - await _socket.SendAsync(buffer, socketFlags).ConfigureAwait(false); -#endif } catch (ObjectDisposedException) { - // This will happen when _socket.EndConnect gets called by Task library but the socket is already disposed. + // This will happen when _socket.EndSend_ gets called by Task library but the socket is already disposed. } } +#else + public Task SendAsync(ArraySegment buffer, SocketFlags socketFlags) + { + return _socket.SendAsync(buffer, socketFlags); + } +#endif #if NET452 || NET461 sealed class SocketWrapper diff --git a/Source/MQTTnet/Implementations/MqttTcpServerListener.cs b/Source/MQTTnet/Implementations/MqttTcpServerListener.cs index 9341d275..ac9a521f 100644 --- a/Source/MQTTnet/Implementations/MqttTcpServerListener.cs +++ b/Source/MQTTnet/Implementations/MqttTcpServerListener.cs @@ -86,6 +86,26 @@ namespace MQTTnet.Implementations _socket.LingerState = _options.LingerState; } + if (_options.KeepAlive.HasValue) + { + _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. diff --git a/Source/MQTTnet/Server/Options/MqttServerOptionsBuilder.cs b/Source/MQTTnet/Server/Options/MqttServerOptionsBuilder.cs index b6a50cca..3b359203 100644 --- a/Source/MQTTnet/Server/Options/MqttServerOptionsBuilder.cs +++ b/Source/MQTTnet/Server/Options/MqttServerOptionsBuilder.cs @@ -7,8 +7,6 @@ using System.Net; using System.Net.Security; using System.Security.Authentication; using MQTTnet.Certificates; -using System.Threading.Tasks; - #if !WINDOWS_UWP using System.Security.Cryptography.X509Certificates; #endif @@ -19,7 +17,22 @@ namespace MQTTnet.Server public class MqttServerOptionsBuilder { readonly MqttServerOptions _options = new MqttServerOptions(); - + + public MqttServerOptions Build() + { + return _options; + } + +#if !WINDOWS_UWP + public MqttServerOptionsBuilder WithClientCertificate(RemoteCertificateValidationCallback validationCallback = null, bool checkCertificateRevocation = false) + { + _options.TlsEndpointOptions.ClientCertificateRequired = true; + _options.TlsEndpointOptions.CheckCertificateRevocation = checkCertificateRevocation; + _options.TlsEndpointOptions.RemoteCertificateValidationCallback = validationCallback; + return this; + } +#endif + public MqttServerOptionsBuilder WithConnectionBacklog(int value) { _options.DefaultEndpointOptions.ConnectionBacklog = value; @@ -27,12 +40,6 @@ namespace MQTTnet.Server return this; } - public MqttServerOptionsBuilder WithMaxPendingMessagesPerClient(int value) - { - _options.MaxPendingMessagesPerClient = value; - return this; - } - public MqttServerOptionsBuilder WithDefaultCommunicationTimeout(TimeSpan value) { _options.DefaultCommunicationTimeout = value; @@ -45,12 +52,6 @@ namespace MQTTnet.Server return this; } - public MqttServerOptionsBuilder WithDefaultEndpointPort(int value) - { - _options.DefaultEndpointOptions.Port = value; - return this; - } - public MqttServerOptionsBuilder WithDefaultEndpointBoundIPAddress(IPAddress value) { _options.DefaultEndpointOptions.BoundInterNetworkAddress = value ?? IPAddress.Any; @@ -63,9 +64,15 @@ namespace MQTTnet.Server return this; } - public MqttServerOptionsBuilder WithoutDefaultEndpoint() + public MqttServerOptionsBuilder WithDefaultEndpointPort(int value) { - _options.DefaultEndpointOptions.IsEnabled = false; + _options.DefaultEndpointOptions.Port = value; + return this; + } + + public MqttServerOptionsBuilder WithDefaultEndpointReuseAddress() + { + _options.DefaultEndpointOptions.ReuseAddress = true; return this; } @@ -75,12 +82,6 @@ namespace MQTTnet.Server return this; } - public MqttServerOptionsBuilder WithEncryptedEndpointPort(int value) - { - _options.TlsEndpointOptions.Port = value; - return this; - } - public MqttServerOptionsBuilder WithEncryptedEndpointBoundIPAddress(IPAddress value) { _options.TlsEndpointOptions.BoundInterNetworkAddress = value; @@ -93,10 +94,91 @@ namespace MQTTnet.Server return this; } + public MqttServerOptionsBuilder WithEncryptedEndpointPort(int value) + { + _options.TlsEndpointOptions.Port = value; + return this; + } + + public MqttServerOptionsBuilder WithEncryptionSslProtocol(SslProtocols value) + { + _options.TlsEndpointOptions.SslProtocol = value; + return this; + } + + public MqttServerOptionsBuilder WithKeepAlive() + { + _options.DefaultEndpointOptions.KeepAlive = true; + _options.TlsEndpointOptions.KeepAlive = true; + return this; + } + + public MqttServerOptionsBuilder WithMaxPendingMessagesPerClient(int value) + { + _options.MaxPendingMessagesPerClient = value; + return this; + } + + public MqttServerOptionsBuilder WithoutDefaultEndpoint() + { + _options.DefaultEndpointOptions.IsEnabled = false; + return this; + } + + public MqttServerOptionsBuilder WithoutEncryptedEndpoint() + { + _options.TlsEndpointOptions.IsEnabled = false; + return this; + } + + public MqttServerOptionsBuilder WithPersistentSessions(bool value = true) + { + _options.EnablePersistentSessions = value; + return this; + } + +#if !WINDOWS_UWP + public MqttServerOptionsBuilder WithRemoteCertificateValidationCallback(RemoteCertificateValidationCallback value) + { + _options.TlsEndpointOptions.RemoteCertificateValidationCallback = value; + return this; + } +#endif + + public MqttServerOptionsBuilder WithTcpKeepAliveInterval(int value) + { + _options.DefaultEndpointOptions.TcpKeepAliveInterval = value; + _options.TlsEndpointOptions.TcpKeepAliveInterval = value; + return this; + } + + public MqttServerOptionsBuilder WithTcpKeepAliveRetryCount(int value) + { + _options.DefaultEndpointOptions.TcpKeepAliveRetryCount = value; + _options.TlsEndpointOptions.TcpKeepAliveRetryCount = value; + return this; + } + + public MqttServerOptionsBuilder WithTcpKeepAliveTime(int value) + { + _options.DefaultEndpointOptions.TcpKeepAliveTime = value; + _options.TlsEndpointOptions.TcpKeepAliveTime = value; + return this; + } + + public MqttServerOptionsBuilder WithTlsEndpointReuseAddress() + { + _options.TlsEndpointOptions.ReuseAddress = true; + return this; + } + #if !WINDOWS_UWP public MqttServerOptionsBuilder WithEncryptionCertificate(byte[] value, IMqttServerCertificateCredentials credentials = null) { - if (value == null) throw new ArgumentNullException(nameof(value)); + if (value == null) + { + throw new ArgumentNullException(nameof(value)); + } _options.TlsEndpointOptions.CertificateProvider = new BlobCertificateProvider(value) { @@ -108,103 +190,14 @@ namespace MQTTnet.Server public MqttServerOptionsBuilder WithEncryptionCertificate(X509Certificate2 certificate) { - if (certificate == null) throw new ArgumentNullException(nameof(certificate)); + if (certificate == null) + { + throw new ArgumentNullException(nameof(certificate)); + } _options.TlsEndpointOptions.CertificateProvider = new X509CertificateProvider(certificate); return this; } #endif - - public MqttServerOptionsBuilder WithEncryptionSslProtocol(SslProtocols value) - { - _options.TlsEndpointOptions.SslProtocol = value; - return this; - } - -#if !WINDOWS_UWP - public MqttServerOptionsBuilder WithClientCertificate(RemoteCertificateValidationCallback validationCallback = null, bool checkCertificateRevocation = false) - { - _options.TlsEndpointOptions.ClientCertificateRequired = true; - _options.TlsEndpointOptions.CheckCertificateRevocation = checkCertificateRevocation; - _options.TlsEndpointOptions.RemoteCertificateValidationCallback = validationCallback; - return this; - } -#endif - - public MqttServerOptionsBuilder WithoutEncryptedEndpoint() - { - _options.TlsEndpointOptions.IsEnabled = false; - return this; - } - -#if !WINDOWS_UWP - public MqttServerOptionsBuilder WithRemoteCertificateValidationCallback(RemoteCertificateValidationCallback value) - { - _options.TlsEndpointOptions.RemoteCertificateValidationCallback = value; - return this; - } -#endif - - // public MqttServerOptionsBuilder WithApplicationMessageInterceptor(IMqttServerApplicationMessageInterceptor value) - // { - // _options.ApplicationMessageInterceptor = value; - // return this; - // } - // - // public MqttServerOptionsBuilder WithApplicationMessageInterceptor(Action value) - // { - // _options.ApplicationMessageInterceptor = new MqttServerApplicationMessageInterceptorDelegate(value); - // return this; - // } - // - // public MqttServerOptionsBuilder WithApplicationMessageInterceptor(Func value) - // { - // _options.ApplicationMessageInterceptor = new MqttServerApplicationMessageInterceptorDelegate(value); - // return this; - // } - - // public MqttServerOptionsBuilder WithMultiThreadedApplicationMessageInterceptor(Action value) - // { - // _options.ApplicationMessageInterceptor = new MqttServerMultiThreadedApplicationMessageInterceptorDelegate(value); - // return this; - // } - // - // public MqttServerOptionsBuilder WithMultiThreadedApplicationMessageInterceptor(Func value) - // { - // _options.ApplicationMessageInterceptor = new MqttServerMultiThreadedApplicationMessageInterceptorDelegate(value); - // return this; - // } - - public MqttServerOptionsBuilder WithDefaultEndpointReuseAddress() - { - _options.DefaultEndpointOptions.ReuseAddress = true; - return this; - } - - public MqttServerOptionsBuilder WithTlsEndpointReuseAddress() - { - _options.TlsEndpointOptions.ReuseAddress = true; - return this; - } - - public MqttServerOptionsBuilder WithPersistentSessions(bool value = true) - { - _options.EnablePersistentSessions = value; - return this; - } - - // /// - // /// Gets or sets the client ID which is used when publishing messages from the server directly. - // /// - // public MqttServerOptionsBuilder WithClientId(string value) - // { - // _options.ClientId = value; - // return this; - // } - - public MqttServerOptions Build() - { - return _options; - } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Options/MqttServerTcpEndpointBaseOptions.cs b/Source/MQTTnet/Server/Options/MqttServerTcpEndpointBaseOptions.cs index feab55bb..1fbdb26b 100644 --- a/Source/MQTTnet/Server/Options/MqttServerTcpEndpointBaseOptions.cs +++ b/Source/MQTTnet/Server/Options/MqttServerTcpEndpointBaseOptions.cs @@ -17,6 +17,30 @@ namespace MQTTnet.Server public bool NoDelay { get; set; } = true; + /// + /// Gets or sets whether the sockets keep alive feature should be used. + /// The value _null_ indicates that the OS and framework defaults should be used. + /// + public bool? KeepAlive { get; set; } + + /// + /// Gets or sets the TCP keep alive interval. + /// The value _null_ indicates that the OS and framework defaults should be used. + /// + public int? TcpKeepAliveInterval { get; set; } + + /// + /// Gets or sets the TCP keep alive retry count. + /// The value _null_ indicates that the OS and framework defaults should be used. + /// + public int? TcpKeepAliveRetryCount { get; set; } + + /// + /// Gets or sets the TCP keep alive time. + /// The value _null_ indicates that the OS and framework defaults should be used. + /// + public int? TcpKeepAliveTime { get; set; } + public LingerOption LingerState { get; set; } = new LingerOption(true, 0); #if WINDOWS_UWP