diff --git a/Source/MQTTnet/Implementations/CrossPlatformSocket.cs b/Source/MQTTnet/Implementations/CrossPlatformSocket.cs index 18d2d892..86e0be1b 100644 --- a/Source/MQTTnet/Implementations/CrossPlatformSocket.cs +++ b/Source/MQTTnet/Implementations/CrossPlatformSocket.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.IO; using System.Net; using System.Net.Sockets; @@ -11,12 +11,14 @@ namespace MQTTnet.Implementations public sealed class CrossPlatformSocket : IDisposable { readonly Socket _socket; + readonly Action _socketDisposeAction; NetworkStream _networkStream; public CrossPlatformSocket(AddressFamily addressFamily) { _socket = new Socket(addressFamily, SocketType.Stream, ProtocolType.Tcp); + _socketDisposeAction = _socket.Dispose; } public CrossPlatformSocket() @@ -24,12 +26,16 @@ namespace MQTTnet.Implementations // 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); + + _socketDisposeAction = _socket.Dispose; } public CrossPlatformSocket(Socket socket) { _socket = socket ?? throw new ArgumentNullException(nameof(socket)); _networkStream = new NetworkStream(socket, true); + + _socketDisposeAction = _socket.Dispose; } public bool NoDelay @@ -112,7 +118,7 @@ namespace MQTTnet.Implementations _networkStream?.Dispose(); // Workaround for: https://github.com/dotnet/corefx/issues/24430 - using (cancellationToken.Register(() => _socket.Dispose())) + using (cancellationToken.Register(_socketDisposeAction)) { cancellationToken.ThrowIfCancellationRequested(); diff --git a/Source/MQTTnet/Implementations/MqttTcpChannel.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.cs index 4cb58ec8..800ab46d 100644 --- a/Source/MQTTnet/Implementations/MqttTcpChannel.cs +++ b/Source/MQTTnet/Implementations/MqttTcpChannel.cs @@ -18,10 +18,16 @@ namespace MQTTnet.Implementations { readonly IMqttClientOptions _clientOptions; readonly MqttClientTcpOptions _tcpOptions; + readonly Action _disposeAction; Stream _stream; - public MqttTcpChannel(IMqttClientOptions clientOptions) + public MqttTcpChannel() + { + _disposeAction = Dispose; + } + + public MqttTcpChannel(IMqttClientOptions clientOptions) : this() { _clientOptions = clientOptions ?? throw new ArgumentNullException(nameof(clientOptions)); _tcpOptions = (MqttClientTcpOptions)clientOptions.ChannelOptions; @@ -29,7 +35,7 @@ namespace MQTTnet.Implementations IsSecureConnection = clientOptions.ChannelOptions?.TlsOptions?.UseTls == true; } - public MqttTcpChannel(Stream stream, string endpoint, X509Certificate2 clientCertificate) + public MqttTcpChannel(Stream stream, string endpoint, X509Certificate2 clientCertificate) : this() { _stream = stream ?? throw new ArgumentNullException(nameof(stream)); @@ -149,11 +155,15 @@ namespace MQTTnet.Implementations return 0; } +#if NETCOREAPP3_0_OR_GREATER || NETSTANDARD2_1_OR_GREATER + return await stream.ReadAsync(buffer.AsMemory(offset, count), cancellationToken).ConfigureAwait(false); +#else // Workaround for: https://github.com/dotnet/corefx/issues/24430 - using (cancellationToken.Register(Dispose)) + using (cancellationToken.Register(_disposeAction)) { return await stream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false); } +#endif } catch (ObjectDisposedException) { @@ -177,18 +187,22 @@ namespace MQTTnet.Implementations try { - // Workaround for: https://github.com/dotnet/corefx/issues/24430 - using (cancellationToken.Register(Dispose)) + var stream = _stream; + + if (stream == null) { - var stream = _stream; - - if (stream == null) - { - throw new MqttCommunicationException("The TCP connection is closed."); - } + throw new MqttCommunicationException("The TCP connection is closed."); + } +#if NETCOREAPP3_0_OR_GREATER || NETSTANDARD2_1_OR_GREATER + await stream.WriteAsync(buffer.AsMemory(offset, count), cancellationToken).ConfigureAwait(false); +#else + // Workaround for: https://github.com/dotnet/corefx/issues/24430 + using (cancellationToken.Register(_disposeAction)) + { await stream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false); } +#endif } catch (ObjectDisposedException) {