From caa857f318a9782047d49177fc7108626c0c4b15 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sat, 9 Sep 2017 01:47:32 +0200 Subject: [PATCH 1/3] fixed bug in read async and unfied stream handling --- .../Implementations/MqttTcpChannel.cs | 41 ++++++++++--------- .../Implementations/MqttTcpChannel.cs | 36 ++++++++++------ 2 files changed, 45 insertions(+), 32 deletions(-) diff --git a/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs b/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs index c9c080d2..037538c8 100644 --- a/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs +++ b/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs @@ -7,11 +7,13 @@ using System.Threading.Tasks; using MQTTnet.Core.Channel; using MQTTnet.Core.Client; using MQTTnet.Core.Exceptions; +using System.IO; namespace MQTTnet.Implementations { public sealed class MqttTcpChannel : IMqttCommunicationChannel, IDisposable { + private Stream _dataStream; private Socket _socket; private SslStream _sslStream; @@ -40,8 +42,13 @@ namespace MQTTnet.Implementations if (options.TlsOptions.UseTls) { _sslStream = new SslStream(new NetworkStream(_socket, true)); + _dataStream = _sslStream; await _sslStream.AuthenticateAsClientAsync(options.Server, LoadCertificates(options), SslProtocols.Tls12, options.TlsOptions.CheckCertificateRevocation); } + else + { + _dataStream = new NetworkStream(_socket); + } } catch (SocketException exception) { @@ -68,16 +75,7 @@ namespace MQTTnet.Implementations try { - if (_sslStream != null) - { - return _sslStream.WriteAsync(buffer, 0, buffer.Length); - } - - return Task.Factory.FromAsync( - // ReSharper disable once AssignNullToNotNullAttribute - _socket.BeginSend(buffer, 0, buffer.Length, SocketFlags.None, null, null), - _socket.EndSend); - + return _dataStream.WriteAsync(buffer, 0, buffer.Length); } catch (SocketException exception) { @@ -85,21 +83,26 @@ namespace MQTTnet.Implementations } } - public Task ReadAsync(byte[] buffer) + public async Task ReadAsync(byte[] buffer) { if (buffer == null) throw new ArgumentNullException(nameof(buffer)); try { - if (_sslStream != null) - { - return _sslStream.ReadAsync(buffer, 0, buffer.Length); - } + int totalBytes = 0; - return Task.Factory.FromAsync( - // ReSharper disable once AssignNullToNotNullAttribute - _socket.BeginReceive(buffer, 0, buffer.Length, SocketFlags.None, null, null), - _socket.EndReceive); + do + { + var read = await _dataStream.ReadAsync(buffer, totalBytes, buffer.Length - totalBytes); + + if (read == 0) + { + throw new MqttCommunicationException(new SocketException((int)SocketError.Disconnecting)); + } + + totalBytes += read; + } + while (totalBytes < buffer.Length); } catch (SocketException exception) { diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs index a4247b09..7f59a278 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs @@ -7,11 +7,13 @@ using System.Threading.Tasks; using MQTTnet.Core.Channel; using MQTTnet.Core.Client; using MQTTnet.Core.Exceptions; +using System.IO; namespace MQTTnet.Implementations { public sealed class MqttTcpChannel : IMqttCommunicationChannel, IDisposable { + private Stream _dataStream; private Socket _socket; private SslStream _sslStream; @@ -36,12 +38,17 @@ namespace MQTTnet.Implementations } await _socket.ConnectAsync(options.Server, options.GetPort()); - + if (options.TlsOptions.UseTls) { _sslStream = new SslStream(new NetworkStream(_socket, true)); + _dataStream = _sslStream; await _sslStream.AuthenticateAsClientAsync(options.Server, LoadCertificates(options), SslProtocols.Tls12, options.TlsOptions.CheckCertificateRevocation); } + else + { + _dataStream = new NetworkStream(_socket); + } } catch (SocketException exception) { @@ -68,12 +75,7 @@ namespace MQTTnet.Implementations try { - if (_sslStream != null) - { - return _sslStream.WriteAsync(buffer, 0, buffer.Length); - } - - return _socket.SendAsync(new ArraySegment(buffer), SocketFlags.None); + return _dataStream.WriteAsync(buffer, 0, buffer.Length); } catch (SocketException exception) { @@ -81,18 +83,26 @@ namespace MQTTnet.Implementations } } - public Task ReadAsync(byte[] buffer) + public async Task ReadAsync(byte[] buffer) { if (buffer == null) throw new ArgumentNullException(nameof(buffer)); try { - if (_sslStream != null) - { - return _sslStream.ReadAsync(buffer, 0, buffer.Length); - } + int totalBytes = 0; - return _socket.ReceiveAsync(new ArraySegment(buffer), SocketFlags.None); + do + { + var read = await _dataStream.ReadAsync(buffer, totalBytes, buffer.Length - totalBytes); + + if (read == 0) + { + throw new MqttCommunicationException(new SocketException((int)SocketError.Disconnecting)); + } + + totalBytes += read; + } + while (totalBytes < buffer.Length); } catch (SocketException exception) { From 783ee4328aaf494a1c811d5355590f1dba280d23 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sat, 9 Sep 2017 02:01:45 +0200 Subject: [PATCH 2/3] fixed nullpointer exception in server --- .../Implementations/MqttTcpChannel.cs | 9 +++++++++ .../Implementations/MqttTcpChannel.cs | 8 ++++++++ 2 files changed, 17 insertions(+) diff --git a/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs b/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs index 037538c8..bb17ee35 100644 --- a/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs +++ b/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs @@ -17,14 +17,23 @@ namespace MQTTnet.Implementations private Socket _socket; private SslStream _sslStream; + /// + /// called on client sockets are created in connect + /// public MqttTcpChannel() { + } + /// + /// called on server, sockets are passed in + /// connect will not be called + /// public MqttTcpChannel(Socket socket, SslStream sslStream) { _socket = socket ?? throw new ArgumentNullException(nameof(socket)); _sslStream = sslStream; + _dataStream = (Stream)sslStream ?? new NetworkStream(socket); } public async Task ConnectAsync(MqttClientOptions options) diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs index 7f59a278..068f36c8 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs @@ -17,14 +17,22 @@ namespace MQTTnet.Implementations private Socket _socket; private SslStream _sslStream; + /// + /// called on client sockets are created in connect + /// public MqttTcpChannel() { } + /// + /// called on server, sockets are passed in + /// connect will not be called + /// public MqttTcpChannel(Socket socket, SslStream sslStream) { _socket = socket ?? throw new ArgumentNullException(nameof(socket)); _sslStream = sslStream; + _dataStream = (Stream)sslStream ?? new NetworkStream(socket); } public async Task ConnectAsync(MqttClientOptions options) From 5e417affa53ba8b054b2b43ea4ef31be4385860c Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sat, 9 Sep 2017 02:06:17 +0200 Subject: [PATCH 3/3] use configure await --- .../MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs | 2 +- .../MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs b/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs index bb17ee35..54a4ce67 100644 --- a/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs +++ b/Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs @@ -102,7 +102,7 @@ namespace MQTTnet.Implementations do { - var read = await _dataStream.ReadAsync(buffer, totalBytes, buffer.Length - totalBytes); + var read = await _dataStream.ReadAsync(buffer, totalBytes, buffer.Length - totalBytes).ConfigureAwait(false); if (read == 0) { diff --git a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs index 068f36c8..e7820a36 100644 --- a/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs +++ b/Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs @@ -101,7 +101,7 @@ namespace MQTTnet.Implementations do { - var read = await _dataStream.ReadAsync(buffer, totalBytes, buffer.Length - totalBytes); + var read = await _dataStream.ReadAsync(buffer, totalBytes, buffer.Length - totalBytes).ConfigureAwait(false); if (read == 0) {