Browse Source

Merge pull request #38 from JanEggers/Buffering

Buffering
release/3.x.x
Christian 7 years ago
committed by GitHub
parent
commit
9a25fb63ad
31 changed files with 646 additions and 643 deletions
  1. +6
    -3
      Frameworks/MQTTnet.NetFramework/Implementations/MqttServerAdapter.cs
  2. +20
    -51
      Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs
  3. +9
    -59
      Frameworks/MQTTnet.NetFramework/Implementations/MqttWebSocketChannel.cs
  4. +80
    -0
      Frameworks/MQTTnet.NetFramework/Implementations/WebSocketStream.cs
  5. +1
    -0
      Frameworks/MQTTnet.NetFramework/MQTTnet.NetFramework.csproj
  6. +2
    -1
      Frameworks/MQTTnet.NetFramework/MqttClientFactory.cs
  7. +5
    -44
      Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs
  8. +8
    -61
      Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs
  9. +80
    -0
      Frameworks/MQTTnet.NetStandard/Implementations/WebSocketStream.cs
  10. +10
    -1
      MQTTnet.Core/Adapter/IMqttCommunicationAdapter.cs
  11. +42
    -28
      MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs
  12. +0
    -77
      MQTTnet.Core/Channel/BufferedCommunicationChannel.cs
  13. +5
    -8
      MQTTnet.Core/Channel/IMqttCommunicationChannel.cs
  14. +9
    -1
      MQTTnet.Core/Client/IMqttClient.cs
  15. +37
    -24
      MQTTnet.Core/Client/MqttClient.cs
  16. +12
    -8
      MQTTnet.Core/Client/MqttPacketDispatcher.cs
  17. +0
    -32
      MQTTnet.Core/Internal/AsyncAutoResetEvent.cs
  18. +55
    -0
      MQTTnet.Core/Internal/TaskExtensions.cs
  19. +1
    -0
      MQTTnet.Core/MQTTnet.Core.csproj
  20. +3
    -4
      MQTTnet.Core/Serializer/IMqttPacketSerializer.cs
  21. +7
    -27
      MQTTnet.Core/Serializer/MqttPacketReader.cs
  22. +120
    -118
      MQTTnet.Core/Serializer/MqttPacketSerializer.cs
  23. +13
    -39
      MQTTnet.Core/Server/MqttClientMessageQueue.cs
  24. +7
    -7
      MQTTnet.Core/Server/MqttClientSession.cs
  25. +4
    -4
      MQTTnet.Core/Server/MqttClientSessionsManager.cs
  26. +1
    -0
      MQTTnet.TestApp.NetCore/MQTTnet.TestApp.NetCore.csproj
  27. +33
    -0
      Tests/MQTTnet.Core.Tests/ExtensionTests.cs
  28. +1
    -0
      Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj
  29. +19
    -29
      Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs
  30. +13
    -2
      Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs
  31. +43
    -15
      Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs

+ 6
- 3
Frameworks/MQTTnet.NetFramework/Implementations/MqttServerAdapter.cs View File

@@ -87,7 +87,9 @@ namespace MQTTnet.Implementations
try try
{ {
var clientSocket = await Task.Factory.FromAsync(_defaultEndpointSocket.BeginAccept, _defaultEndpointSocket.EndAccept, null).ConfigureAwait(false); var clientSocket = await Task.Factory.FromAsync(_defaultEndpointSocket.BeginAccept, _defaultEndpointSocket.EndAccept, null).ConfigureAwait(false);
var clientAdapter = new MqttChannelCommunicationAdapter(new BufferedCommunicationChannel(new MqttTcpChannel(clientSocket, null)), new MqttPacketSerializer());

var tcpChannel = new MqttTcpChannel(clientSocket, null);
var clientAdapter = new MqttChannelCommunicationAdapter(tcpChannel, new MqttPacketSerializer());
ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter)); ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
} }
catch (Exception exception) when (!(exception is ObjectDisposedException)) catch (Exception exception) when (!(exception is ObjectDisposedException))
@@ -110,8 +112,9 @@ namespace MQTTnet.Implementations


var sslStream = new SslStream(new NetworkStream(clientSocket)); var sslStream = new SslStream(new NetworkStream(clientSocket));
await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, SslProtocols.Tls12, false).ConfigureAwait(false); await sslStream.AuthenticateAsServerAsync(_tlsCertificate, false, SslProtocols.Tls12, false).ConfigureAwait(false);
var clientAdapter = new MqttChannelCommunicationAdapter(new MqttTcpChannel(clientSocket, sslStream), new MqttPacketSerializer());

var tcpChannel = new MqttTcpChannel(clientSocket, sslStream);
var clientAdapter = new MqttChannelCommunicationAdapter(tcpChannel, new MqttPacketSerializer());
ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter)); ClientConnected?.Invoke(this, new MqttClientConnectedEventArgs(clientSocket.RemoteEndPoint.ToString(), clientAdapter));
} }
catch (Exception exception) catch (Exception exception)


+ 20
- 51
Frameworks/MQTTnet.NetFramework/Implementations/MqttTcpChannel.cs View File

@@ -13,10 +13,16 @@ namespace MQTTnet.Implementations
{ {
public sealed class MqttTcpChannel : IMqttCommunicationChannel, IDisposable public sealed class MqttTcpChannel : IMqttCommunicationChannel, IDisposable
{ {
private Stream _dataStream;
private Stream _rawStream;
private Stream _sendStream;
private Stream _receiveStream;
private Socket _socket; private Socket _socket;
private SslStream _sslStream; private SslStream _sslStream;


public Stream RawStream => _rawStream;
public Stream SendStream => _sendStream;
public Stream ReceiveStream => _receiveStream;

/// <summary> /// <summary>
/// called on client sockets are created in connect /// called on client sockets are created in connect
/// </summary> /// </summary>
@@ -33,7 +39,7 @@ namespace MQTTnet.Implementations
{ {
_socket = socket ?? throw new ArgumentNullException(nameof(socket)); _socket = socket ?? throw new ArgumentNullException(nameof(socket));
_sslStream = sslStream; _sslStream = sslStream;
_dataStream = (Stream)sslStream ?? new NetworkStream(socket);
CreateCommStreams( socket, sslStream );
} }


public async Task ConnectAsync(MqttClientOptions options) public async Task ConnectAsync(MqttClientOptions options)
@@ -52,13 +58,10 @@ namespace MQTTnet.Implementations
{ {
_sslStream = new SslStream(new NetworkStream(_socket, true)); _sslStream = new SslStream(new NetworkStream(_socket, true));
_dataStream = _sslStream;
await _sslStream.AuthenticateAsClientAsync(options.Server, LoadCertificates(options), SslProtocols.Tls12, options.TlsOptions.CheckCertificateRevocation).ConfigureAwait(false); await _sslStream.AuthenticateAsClientAsync(options.Server, LoadCertificates(options), SslProtocols.Tls12, options.TlsOptions.CheckCertificateRevocation).ConfigureAwait(false);
} }
else
{
_dataStream = new NetworkStream(_socket);
}

CreateCommStreams( _socket, _sslStream );
} }
catch (SocketException exception) catch (SocketException exception)
{ {
@@ -79,45 +82,6 @@ namespace MQTTnet.Implementations
} }
} }


public async Task WriteAsync(byte[] buffer)
{
if (buffer == null) throw new ArgumentNullException(nameof(buffer));

try
{
await _dataStream.WriteAsync(buffer, 0, buffer.Length);
}
catch (SocketException exception)
{
throw new MqttCommunicationException(exception);
}
}

public async Task<ArraySegment<byte>> ReadAsync(int length, byte[] buffer)
{
try
{
var totalBytes = 0;

do
{
var read = await _dataStream.ReadAsync(buffer, totalBytes, length - totalBytes).ConfigureAwait(false);
if (read == 0)
{
throw new MqttCommunicationException(new SocketException((int)SocketError.Disconnecting));
}

totalBytes += read;
}
while (totalBytes < length);
return new ArraySegment<byte>(buffer, 0, length);
}
catch (SocketException exception)
{
throw new MqttCommunicationException(exception);
}
}

public void Dispose() public void Dispose()
{ {
_socket?.Dispose(); _socket?.Dispose();
@@ -127,6 +91,16 @@ namespace MQTTnet.Implementations
_sslStream = null; _sslStream = null;
} }


private void CreateCommStreams( Socket socket, SslStream sslStream )
{
_rawStream = (Stream)sslStream ?? new NetworkStream( socket );

//cannot use this as default buffering prevents from receiving the first connect message
//need two streams otherwise read and write have to be synchronized
_sendStream = new BufferedStream( _rawStream, BufferConstants.Size );
_receiveStream = new BufferedStream( _rawStream, BufferConstants.Size );
}

private static X509CertificateCollection LoadCertificates(MqttClientOptions options) private static X509CertificateCollection LoadCertificates(MqttClientOptions options)
{ {
var certificates = new X509CertificateCollection(); var certificates = new X509CertificateCollection();
@@ -142,10 +116,5 @@ namespace MQTTnet.Implementations


return certificates; return certificates;
} }

public int Peek()
{
return _socket.Available;
}
} }
} }

+ 9
- 59
Frameworks/MQTTnet.NetFramework/Implementations/MqttWebSocketChannel.cs View File

@@ -2,6 +2,7 @@
using MQTTnet.Core.Client; using MQTTnet.Core.Client;
using MQTTnet.Core.Exceptions; using MQTTnet.Core.Exceptions;
using System; using System;
using System.IO;
using System.Net.WebSockets; using System.Net.WebSockets;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
@@ -11,8 +12,11 @@ namespace MQTTnet.Implementations
public sealed class MqttWebSocketChannel : IMqttCommunicationChannel, IDisposable public sealed class MqttWebSocketChannel : IMqttCommunicationChannel, IDisposable
{ {
private ClientWebSocket _webSocket = new ClientWebSocket(); private ClientWebSocket _webSocket = new ClientWebSocket();
private int WebSocketBufferSize;
private int WebSocketBufferOffset;

public Stream RawStream { get; private set; }

public Stream SendStream => RawStream;
public Stream ReceiveStream => RawStream;


public async Task ConnectAsync(MqttClientOptions options) public async Task ConnectAsync(MqttClientOptions options)
{ {
@@ -22,6 +26,8 @@ namespace MQTTnet.Implementations
{ {
_webSocket = new ClientWebSocket(); _webSocket = new ClientWebSocket();
await _webSocket.ConnectAsync(new Uri(options.Server), CancellationToken.None); await _webSocket.ConnectAsync(new Uri(options.Server), CancellationToken.None);

RawStream = new WebSocketStream(_webSocket);
} }
catch (WebSocketException exception) catch (WebSocketException exception)
{ {
@@ -31,6 +37,7 @@ namespace MQTTnet.Implementations


public Task DisconnectAsync() public Task DisconnectAsync()
{ {
RawStream = null;
return _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None); return _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None);
} }


@@ -38,62 +45,5 @@ namespace MQTTnet.Implementations
{ {
_webSocket?.Dispose(); _webSocket?.Dispose();
} }

public async Task<ArraySegment<byte>> ReadAsync(int length, byte[] buffer)
{
await ReadToBufferAsync(length, buffer).ConfigureAwait(false);
var result = new ArraySegment<byte>(buffer, WebSocketBufferOffset, length);
WebSocketBufferSize -= length;
WebSocketBufferOffset += length;
return result;
}

private async Task ReadToBufferAsync(int length, byte[] buffer)
{
if (WebSocketBufferSize > 0)
{
return;
}

var offset = 0;
while (_webSocket.State == WebSocketState.Open && WebSocketBufferSize < length)
{
WebSocketReceiveResult response;
do
{
response = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer, offset, buffer.Length - offset), CancellationToken.None).ConfigureAwait(false);
offset += response.Count;
} while (!response.EndOfMessage);

WebSocketBufferSize = response.Count;
if (response.MessageType == WebSocketMessageType.Close)
{
await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false);
}
}
}

public async Task WriteAsync(byte[] buffer)
{
if (buffer == null) {
throw new ArgumentNullException(nameof(buffer));
}

try
{
await _webSocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Binary, true, CancellationToken.None);
}
catch (WebSocketException exception)
{
throw new MqttCommunicationException(exception);
}
}

public int Peek()
{
return WebSocketBufferSize;
}
} }
} }

+ 80
- 0
Frameworks/MQTTnet.NetFramework/Implementations/WebSocketStream.cs View File

@@ -0,0 +1,80 @@
using System;
using System.IO;
using System.Net.WebSockets;
using System.Threading;
using System.Threading.Tasks;

namespace MQTTnet.Implementations
{
public class WebSocketStream : Stream
{
private readonly ClientWebSocket _webSocket;

public WebSocketStream( ClientWebSocket webSocket )
{
_webSocket = webSocket;
}

public override void Flush()
{
}

public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
var currentOffset = offset;
var targetOffset = offset + count;
while (_webSocket.State == WebSocketState.Open && currentOffset < targetOffset)
{
var response = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer, currentOffset, count), cancellationToken).ConfigureAwait(false);
currentOffset += response.Count;

if ( response.MessageType == WebSocketMessageType.Close )
{
await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, cancellationToken).ConfigureAwait(false);
}
}

return currentOffset - offset;
}

public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
return _webSocket.SendAsync(new ArraySegment<byte>( buffer, offset, count ), WebSocketMessageType.Binary, true, cancellationToken);
}

public override int Read( byte[] buffer, int offset, int count )
{
return ReadAsync(buffer, offset, count).GetAwaiter().GetResult();
}

public override void Write(byte[] buffer, int offset, int count)
{
WriteAsync(buffer, offset, count).GetAwaiter().GetResult();
}

public override bool CanRead => true;
public override bool CanSeek => false;
public override bool CanWrite => true;

public override long Length
{
get { throw new NotSupportedException(); }
}

public override long Position
{
get { throw new NotSupportedException(); }
set { throw new NotSupportedException(); }
}

public override long Seek(long offset, SeekOrigin origin)
{
throw new NotSupportedException();
}

public override void SetLength(long value)
{
throw new NotSupportedException();
}
}
}

+ 1
- 0
Frameworks/MQTTnet.NetFramework/MQTTnet.NetFramework.csproj View File

@@ -101,6 +101,7 @@
</ItemGroup> </ItemGroup>
<ItemGroup> <ItemGroup>
<Compile Include="Implementations\MqttWebSocketChannel.cs" /> <Compile Include="Implementations\MqttWebSocketChannel.cs" />
<Compile Include="Implementations\WebSocketStream.cs" />
<Compile Include="MqttClientFactory.cs" /> <Compile Include="MqttClientFactory.cs" />
<Compile Include="MqttServerFactory.cs" /> <Compile Include="MqttServerFactory.cs" />
<Compile Include="Implementations\MqttServerAdapter.cs" /> <Compile Include="Implementations\MqttServerAdapter.cs" />


+ 2
- 1
Frameworks/MQTTnet.NetFramework/MqttClientFactory.cs View File

@@ -22,7 +22,8 @@ namespace MQTTnet
{ {
case MqttConnectionType.Tcp: case MqttConnectionType.Tcp:
case MqttConnectionType.Tls: case MqttConnectionType.Tls:
return new BufferedCommunicationChannel( new MqttTcpChannel() );
var tcp = new MqttTcpChannel();
return tcp;
case MqttConnectionType.Ws: case MqttConnectionType.Ws:
case MqttConnectionType.Wss: case MqttConnectionType.Wss:
return new MqttWebSocketChannel(); return new MqttWebSocketChannel();


+ 5
- 44
Frameworks/MQTTnet.NetStandard/Implementations/MqttTcpChannel.cs View File

@@ -17,6 +17,11 @@ namespace MQTTnet.Implementations
private Socket _socket; private Socket _socket;
private SslStream _sslStream; private SslStream _sslStream;



public Stream ReceiveStream => _dataStream;
public Stream RawStream => _dataStream;
public Stream SendStream => _dataStream;

/// <summary> /// <summary>
/// called on client sockets are created in connect /// called on client sockets are created in connect
/// </summary> /// </summary>
@@ -78,45 +83,6 @@ namespace MQTTnet.Implementations
} }
} }


public async Task WriteAsync(byte[] buffer)
{
if (buffer == null) throw new ArgumentNullException(nameof(buffer));

try
{
await _dataStream.WriteAsync(buffer, 0, buffer.Length);
}
catch (SocketException exception)
{
throw new MqttCommunicationException(exception);
}
}

public async Task<ArraySegment<byte>> ReadAsync(int length, byte[] buffer)
{
try
{
var totalBytes = 0;

do
{
var read = await _dataStream.ReadAsync(buffer, totalBytes, length - totalBytes).ConfigureAwait(false);
if (read == 0)
{
throw new MqttCommunicationException(new SocketException((int)SocketError.Disconnecting));
}

totalBytes += read;
}
while (totalBytes < length);
return new ArraySegment<byte>(buffer, 0, length);
}
catch (SocketException exception)
{
throw new MqttCommunicationException(exception);
}
}

public void Dispose() public void Dispose()
{ {
_socket?.Dispose(); _socket?.Dispose();
@@ -141,10 +107,5 @@ namespace MQTTnet.Implementations


return certificates; return certificates;
} }

public int Peek()
{
return _socket.Available;
}
} }
} }

+ 8
- 61
Frameworks/MQTTnet.NetStandard/Implementations/MqttWebSocketChannel.cs View File

@@ -2,6 +2,7 @@
using MQTTnet.Core.Client; using MQTTnet.Core.Client;
using MQTTnet.Core.Exceptions; using MQTTnet.Core.Exceptions;
using System; using System;
using System.IO;
using System.Net.WebSockets; using System.Net.WebSockets;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
@@ -11,8 +12,10 @@ namespace MQTTnet.Implementations
public sealed class MqttWebSocketChannel : IMqttCommunicationChannel, IDisposable public sealed class MqttWebSocketChannel : IMqttCommunicationChannel, IDisposable
{ {
private ClientWebSocket _webSocket = new ClientWebSocket(); private ClientWebSocket _webSocket = new ClientWebSocket();
private int _bufferSize;
private int _bufferOffset;

public Stream SendStream => RawStream;
public Stream ReceiveStream => RawStream;
public Stream RawStream { get; private set; }


public async Task ConnectAsync(MqttClientOptions options) public async Task ConnectAsync(MqttClientOptions options)
{ {
@@ -22,6 +25,8 @@ namespace MQTTnet.Implementations
{ {
_webSocket = new ClientWebSocket(); _webSocket = new ClientWebSocket();
await _webSocket.ConnectAsync(new Uri(options.Server), CancellationToken.None); await _webSocket.ConnectAsync(new Uri(options.Server), CancellationToken.None);

RawStream = new WebSocketStream(_webSocket);
} }
catch (WebSocketException exception) catch (WebSocketException exception)
{ {
@@ -31,6 +36,7 @@ namespace MQTTnet.Implementations


public Task DisconnectAsync() public Task DisconnectAsync()
{ {
RawStream = null;
return _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None); return _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None);
} }


@@ -38,64 +44,5 @@ namespace MQTTnet.Implementations
{ {
_webSocket?.Dispose(); _webSocket?.Dispose();
} }

public async Task<ArraySegment<byte>> ReadAsync(int length, byte[] buffer)
{
await ReadToBufferAsync(length, buffer).ConfigureAwait(false);

var result = new ArraySegment<byte>(buffer, _bufferOffset, length);
_bufferSize -= length;
_bufferOffset += length;

return result;
}

private async Task ReadToBufferAsync(int length, byte[] buffer)
{
if (_bufferSize > 0)
{
return;
}

var offset = 0;
while (_webSocket.State == WebSocketState.Open && _bufferSize < length)
{
WebSocketReceiveResult response;
do
{
response = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer, offset, buffer.Length - offset), CancellationToken.None).ConfigureAwait(false);
offset += response.Count;
} while (!response.EndOfMessage);

_bufferSize = response.Count;

if (response.MessageType == WebSocketMessageType.Close)
{
await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None).ConfigureAwait(false);
}
}
}

public async Task WriteAsync(byte[] buffer)
{
if (buffer == null)
{
throw new ArgumentNullException(nameof(buffer));
}

try
{
await _webSocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Binary, true, CancellationToken.None);
}
catch (WebSocketException exception)
{
throw new MqttCommunicationException(exception);
}
}

public int Peek()
{
return _bufferSize;
}
} }
} }

+ 80
- 0
Frameworks/MQTTnet.NetStandard/Implementations/WebSocketStream.cs View File

@@ -0,0 +1,80 @@
using System;
using System.IO;
using System.Net.WebSockets;
using System.Threading;
using System.Threading.Tasks;

namespace MQTTnet.Implementations
{
public class WebSocketStream : Stream
{
private readonly ClientWebSocket _webSocket;

public WebSocketStream(ClientWebSocket webSocket)
{
_webSocket = webSocket;
}

public override void Flush()
{
}

public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
var currentOffset = offset;
var targetOffset = offset + count;
while ( _webSocket.State == WebSocketState.Open && currentOffset < targetOffset )
{
var response = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer, currentOffset, count), cancellationToken).ConfigureAwait(false);
currentOffset += response.Count;

if ( response.MessageType == WebSocketMessageType.Close )
{
await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, cancellationToken).ConfigureAwait(false);
}
}

return currentOffset - offset;
}

public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
return _webSocket.SendAsync(new ArraySegment<byte>( buffer, offset, count ), WebSocketMessageType.Binary, true, cancellationToken);
}

public override int Read(byte[] buffer, int offset, int count)
{
return ReadAsync(buffer, offset, count).GetAwaiter().GetResult();
}

public override void Write(byte[] buffer, int offset, int count)
{
WriteAsync( buffer, offset, count ).GetAwaiter().GetResult();
}

public override bool CanRead => true;
public override bool CanSeek => false;
public override bool CanWrite => true;

public override long Length
{
get { throw new NotSupportedException(); }
}

public override long Position
{
get { throw new NotSupportedException(); }
set { throw new NotSupportedException(); }
}

public override long Seek(long offset, SeekOrigin origin)
{
throw new NotSupportedException();
}

public override void SetLength(long value)
{
throw new NotSupportedException();
}
}
}

+ 10
- 1
MQTTnet.Core/Adapter/IMqttCommunicationAdapter.cs View File

@@ -1,4 +1,5 @@
using System; using System;
using System.Collections.Generic;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Core.Client; using MQTTnet.Core.Client;
using MQTTnet.Core.Packets; using MQTTnet.Core.Packets;
@@ -12,10 +13,18 @@ namespace MQTTnet.Core.Adapter


Task DisconnectAsync(); Task DisconnectAsync();


Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout);
Task SendPacketsAsync( TimeSpan timeout, IEnumerable<MqttBasePacket> packets );


Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout); Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout);


IMqttPacketSerializer PacketSerializer { get; } IMqttPacketSerializer PacketSerializer { get; }
} }

public static class IMqttCommunicationAdapterExtensions
{
public static Task SendPacketsAsync( this IMqttCommunicationAdapter adapter, TimeSpan timeout, params MqttBasePacket[] packets )
{
return adapter.SendPacketsAsync( timeout, packets );
}
}
} }

+ 42
- 28
MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs View File

@@ -1,9 +1,13 @@
using System; using System;
using System.Collections.Generic;
using System.IO;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Core.Channel; using MQTTnet.Core.Channel;
using MQTTnet.Core.Client; using MQTTnet.Core.Client;
using MQTTnet.Core.Diagnostics; using MQTTnet.Core.Diagnostics;
using MQTTnet.Core.Exceptions; using MQTTnet.Core.Exceptions;
using MQTTnet.Core.Internal;
using MQTTnet.Core.Packets; using MQTTnet.Core.Packets;
using MQTTnet.Core.Serializer; using MQTTnet.Core.Serializer;


@@ -12,6 +16,7 @@ namespace MQTTnet.Core.Adapter
public class MqttChannelCommunicationAdapter : IMqttCommunicationAdapter public class MqttChannelCommunicationAdapter : IMqttCommunicationAdapter
{ {
private readonly IMqttCommunicationChannel _channel; private readonly IMqttCommunicationChannel _channel;
private readonly byte[] _readBuffer = new byte[BufferConstants.Size];


public MqttChannelCommunicationAdapter(IMqttCommunicationChannel channel, IMqttPacketSerializer serializer) public MqttChannelCommunicationAdapter(IMqttCommunicationChannel channel, IMqttPacketSerializer serializer)
{ {
@@ -23,7 +28,7 @@ namespace MQTTnet.Core.Adapter


public Task ConnectAsync(MqttClientOptions options, TimeSpan timeout) public Task ConnectAsync(MqttClientOptions options, TimeSpan timeout)
{ {
return ExecuteWithTimeoutAsync(_channel.ConnectAsync(options), timeout);
return _channel.ConnectAsync(options).TimeoutAfter(timeout);
} }


public Task DisconnectAsync() public Task DisconnectAsync()
@@ -31,25 +36,40 @@ namespace MQTTnet.Core.Adapter
return _channel.DisconnectAsync(); return _channel.DisconnectAsync();
} }


public Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout)
public async Task SendPacketsAsync( TimeSpan timeout, IEnumerable<MqttBasePacket> packets )
{ {
MqttTrace.Information(nameof(MqttChannelCommunicationAdapter), "TX >>> {0} [Timeout={1}]", packet, timeout);
lock ( _channel )
{
foreach (var packet in packets )
{
MqttTrace.Information( nameof( MqttChannelCommunicationAdapter ), "TX >>> {0} [Timeout={1}]", packet, timeout );

var writeBuffer = PacketSerializer.Serialize(packet);
_sendTask = _sendTask.ContinueWith( p => _channel.SendStream.WriteAsync( writeBuffer, 0, writeBuffer.Length ) );
}
}


return ExecuteWithTimeoutAsync(PacketSerializer.SerializeAsync(packet, _channel), timeout);
await _sendTask; // configure await false geneates stackoverflow
await _channel.SendStream.FlushAsync().TimeoutAfter( timeout ).ConfigureAwait( false );
} }


private Task _sendTask = Task.FromResult(0); // this task is used to prevent overlapping write

public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout) public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout)
{ {
MqttBasePacket packet;
Tuple<MqttPacketHeader, MemoryStream> tuple;
if (timeout > TimeSpan.Zero) if (timeout > TimeSpan.Zero)
{ {
packet = await ExecuteWithTimeoutAsync(PacketSerializer.DeserializeAsync(_channel), timeout).ConfigureAwait(false);
tuple = await ReceiveAsync(_channel.RawStream).TimeoutAfter(timeout).ConfigureAwait(false);
} }
else else
{ {
packet = await PacketSerializer.DeserializeAsync(_channel).ConfigureAwait(false);
tuple = await ReceiveAsync(_channel.ReceiveStream).ConfigureAwait(false);
} }


var packet = PacketSerializer.Deserialize(tuple.Item1, tuple.Item2);

if (packet == null) if (packet == null)
{ {
throw new MqttProtocolViolationException("Received malformed packet."); throw new MqttProtocolViolationException("Received malformed packet.");
@@ -59,34 +79,28 @@ namespace MQTTnet.Core.Adapter
return packet; return packet;
} }


private static async Task<TResult> ExecuteWithTimeoutAsync<TResult>(Task<TResult> task, TimeSpan timeout)
private async Task<Tuple<MqttPacketHeader, MemoryStream>> ReceiveAsync(Stream stream)
{ {
var timeoutTask = Task.Delay(timeout);
if (await Task.WhenAny(timeoutTask, task).ConfigureAwait(false) == timeoutTask)
{
throw new MqttCommunicationTimedOutException();
}
var header = MqttPacketReader.ReadHeaderFromSource(stream);


if (task.IsFaulted)
MemoryStream body = null;
if (header.BodyLength > 0)
{ {
throw new MqttCommunicationException(task.Exception);
var totalRead = 0;
do
{
var read = await stream.ReadAsync(_readBuffer, totalRead, header.BodyLength - totalRead)
.ConfigureAwait( false );
totalRead += read;
} while (totalRead < header.BodyLength);
body = new MemoryStream(_readBuffer, 0, header.BodyLength);
} }

return task.Result;
}

private static async Task ExecuteWithTimeoutAsync(Task task, TimeSpan timeout)
{
var timeoutTask = Task.Delay(timeout);
if (await Task.WhenAny(timeoutTask, task).ConfigureAwait(false) == timeoutTask)
else
{ {
throw new MqttCommunicationTimedOutException();
body = new MemoryStream();
} }


if (task.IsFaulted)
{
throw new MqttCommunicationException(task.Exception);
}
return Tuple.Create(header, body);
} }
} }
} }

+ 0
- 77
MQTTnet.Core/Channel/BufferedCommunicationChannel.cs View File

@@ -1,77 +0,0 @@
using System.Threading.Tasks;
using MQTTnet.Core.Client;
using System;

namespace MQTTnet.Core.Channel
{
public class BufferedCommunicationChannel : IMqttCommunicationChannel
{
private readonly IMqttCommunicationChannel _inner;
private int _bufferSize;
private int _bufferOffset;

public BufferedCommunicationChannel(IMqttCommunicationChannel inner)
{
_inner = inner;
}

public Task ConnectAsync(MqttClientOptions options)
{
return _inner.ConnectAsync(options);
}

public Task DisconnectAsync()
{
return _inner.DisconnectAsync();
}

public int Peek()
{
return _inner.Peek();
}

public async Task<ArraySegment<byte>> ReadAsync(int length, byte[] buffer)
{
//read from buffer
if (_bufferSize > 0)
{
return ReadFomBuffer(length, buffer);
}

var available = _inner.Peek();
// if there are less or equal bytes available then requested then just read em
if (available <= length)
{
return await _inner.ReadAsync(length, buffer);
}

//if more bytes are available than requested do buffer them to reduce calls to network buffers
await WriteToBuffer(available, buffer).ConfigureAwait(false);
return ReadFomBuffer(length, buffer);
}

private async Task WriteToBuffer(int available, byte[] buffer)
{
await _inner.ReadAsync(available, buffer).ConfigureAwait(false);
_bufferSize = available;
_bufferOffset = 0;
}

private ArraySegment<byte> ReadFomBuffer(int length, byte[] buffer)
{
var result = new ArraySegment<byte>(buffer, _bufferOffset, length);
_bufferSize -= length;
_bufferOffset += length;

if (_bufferSize < 0)
{
}
return result;
}

public Task WriteAsync(byte[] buffer)
{
return _inner.WriteAsync(buffer);
}
}
}

+ 5
- 8
MQTTnet.Core/Channel/IMqttCommunicationChannel.cs View File

@@ -1,6 +1,6 @@
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Core.Client; using MQTTnet.Core.Client;
using System;
using System.IO;


namespace MQTTnet.Core.Channel namespace MQTTnet.Core.Channel
{ {
@@ -9,14 +9,11 @@ namespace MQTTnet.Core.Channel
Task ConnectAsync(MqttClientOptions options); Task ConnectAsync(MqttClientOptions options);


Task DisconnectAsync(); Task DisconnectAsync();
Stream SendStream { get; }


Task WriteAsync(byte[] buffer);
Stream ReceiveStream { get; }


/// <summary>
/// get the currently available number of bytes without reading them
/// </summary>
int Peek();

Task<ArraySegment<byte>> ReadAsync(int length, byte[] buffer);
Stream RawStream { get; }
} }
} }

+ 9
- 1
MQTTnet.Core/Client/IMqttClient.cs View File

@@ -15,10 +15,18 @@ namespace MQTTnet.Core.Client


Task ConnectAsync(MqttApplicationMessage willApplicationMessage = null); Task ConnectAsync(MqttApplicationMessage willApplicationMessage = null);
Task DisconnectAsync(); Task DisconnectAsync();
Task PublishAsync(MqttApplicationMessage applicationMessage);
Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages);
Task<IList<MqttSubscribeResult>> SubscribeAsync(IList<TopicFilter> topicFilters); Task<IList<MqttSubscribeResult>> SubscribeAsync(IList<TopicFilter> topicFilters);
Task<IList<MqttSubscribeResult>> SubscribeAsync(params TopicFilter[] topicFilters); Task<IList<MqttSubscribeResult>> SubscribeAsync(params TopicFilter[] topicFilters);
Task Unsubscribe(IList<string> topicFilters); Task Unsubscribe(IList<string> topicFilters);
Task Unsubscribe(params string[] topicFilters); Task Unsubscribe(params string[] topicFilters);
} }

public static class IMqttClientExtensions
{
public static Task PublishAsync( this IMqttClient client, params MqttApplicationMessage[] applicationMessages )
{
return client.PublishAsync( applicationMessages );
}
}
} }

+ 37
- 24
MQTTnet.Core/Client/MqttClient.cs View File

@@ -161,32 +161,43 @@ namespace MQTTnet.Core.Client
return SendAndReceiveAsync<MqttUnsubAckPacket>(unsubscribePacket); return SendAndReceiveAsync<MqttUnsubAckPacket>(unsubscribePacket);
} }


public Task PublishAsync(MqttApplicationMessage applicationMessage)
public async Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages)
{ {
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
ThrowIfNotConnected(); ThrowIfNotConnected();


var publishPacket = applicationMessage.ToPublishPacket();
var publishPackets = applicationMessages.Select(m => m.ToPublishPacket());


if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
foreach (var qosGroup in publishPackets.GroupBy(p => p.QualityOfServiceLevel))
{ {
// No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
return SendAsync(publishPacket);
}

if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
{
publishPacket.PacketIdentifier = GetNewPacketIdentifier();
return SendAndReceiveAsync<MqttPubAckPacket>(publishPacket);
}

if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
{
publishPacket.PacketIdentifier = GetNewPacketIdentifier();
return PublishExactlyOncePacketAsync(publishPacket);
var qosPackets = qosGroup.ToArray();
switch ( qosGroup.Key )
{
case MqttQualityOfServiceLevel.AtMostOnce:
// No packet identifier is used for QoS 0 [3.3.2.2 Packet Identifier]
await _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, qosPackets);
break;
case MqttQualityOfServiceLevel.AtLeastOnce:
{
foreach (var publishPacket in qosPackets)
{
publishPacket.PacketIdentifier = GetNewPacketIdentifier();
await SendAndReceiveAsync<MqttPubAckPacket>(publishPacket);
}
break;
}
case MqttQualityOfServiceLevel.ExactlyOnce:
{
foreach (var publishPacket in qosPackets)
{
publishPacket.PacketIdentifier = GetNewPacketIdentifier();
await PublishExactlyOncePacketAsync( publishPacket );
}
break;
}
default:
throw new InvalidOperationException();
}
} }

throw new InvalidOperationException();
} }


private async Task PublishExactlyOncePacketAsync(MqttBasePacket publishPacket) private async Task PublishExactlyOncePacketAsync(MqttBasePacket publishPacket)
@@ -277,12 +288,14 @@ namespace MQTTnet.Core.Client
if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce) if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
{ {
FireApplicationMessageReceivedEvent(publishPacket); FireApplicationMessageReceivedEvent(publishPacket);
return;
} }


if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
{ {
FireApplicationMessageReceivedEvent(publishPacket); FireApplicationMessageReceivedEvent(publishPacket);
await SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }); await SendAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier });
return;
} }


if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
@@ -295,6 +308,7 @@ namespace MQTTnet.Core.Client


FireApplicationMessageReceivedEvent(publishPacket); FireApplicationMessageReceivedEvent(publishPacket);
await SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }); await SendAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier });
return;
} }


throw new MqttCommunicationException("Received a not supported QoS level."); throw new MqttCommunicationException("Received a not supported QoS level.");
@@ -312,14 +326,13 @@ namespace MQTTnet.Core.Client


private Task SendAsync(MqttBasePacket packet) private Task SendAsync(MqttBasePacket packet)
{ {
return _adapter.SendPacketAsync(packet, _options.DefaultCommunicationTimeout);
return _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, packet);
} }


private async Task<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(MqttBasePacket requestPacket) where TResponsePacket : MqttBasePacket private async Task<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(MqttBasePacket requestPacket) where TResponsePacket : MqttBasePacket
{ {
await _adapter.SendPacketAsync(requestPacket, _options.DefaultCommunicationTimeout).ConfigureAwait(false);

return (TResponsePacket)await _packetDispatcher.WaitForPacketAsync(requestPacket, typeof(TResponsePacket), _options.DefaultCommunicationTimeout).ConfigureAwait(false);
await _adapter.SendPacketsAsync( _options.DefaultCommunicationTimeout, requestPacket ).ConfigureAwait(false);
return (TResponsePacket)await _packetDispatcher.WaitForPacketAsync(requestPacket, typeof( TResponsePacket ), _options.DefaultCommunicationTimeout).ConfigureAwait(false);
} }


private ushort GetNewPacketIdentifier() private ushort GetNewPacketIdentifier()


+ 12
- 8
MQTTnet.Core/Client/MqttPacketDispatcher.cs View File

@@ -3,6 +3,7 @@ using System.Collections.Generic;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Core.Diagnostics; using MQTTnet.Core.Diagnostics;
using MQTTnet.Core.Exceptions; using MQTTnet.Core.Exceptions;
using MQTTnet.Core.Internal;
using MQTTnet.Core.Packets; using MQTTnet.Core.Packets;
using System.Collections.Concurrent; using System.Collections.Concurrent;


@@ -22,16 +23,19 @@ namespace MQTTnet.Core.Client
var packetAwaiter = AddPacketAwaiter(request, responseType); var packetAwaiter = AddPacketAwaiter(request, responseType);
DispatchPendingPackets(); DispatchPendingPackets();


var hasTimeout = await Task.WhenAny(Task.Delay(timeout), packetAwaiter.Task).ConfigureAwait(false) != packetAwaiter.Task;
RemovePacketAwaiter(request, responseType);

if (hasTimeout)
try
{ {
MqttTrace.Warning(nameof(MqttPacketDispatcher), "Timeout while waiting for packet.");
throw new MqttCommunicationTimedOutException();
return await packetAwaiter.Task.TimeoutAfter( timeout );
}
catch ( MqttCommunicationTimedOutException )
{
MqttTrace.Warning( nameof( MqttPacketDispatcher ), "Timeout while waiting for packet." );
throw;
}
finally
{
RemovePacketAwaiter(request, responseType);
} }

return packetAwaiter.Task.Result;
} }


public void Dispatch(MqttBasePacket packet) public void Dispatch(MqttBasePacket packet)


+ 0
- 32
MQTTnet.Core/Internal/AsyncAutoResetEvent.cs View File

@@ -1,32 +0,0 @@
using System.Collections.Generic;
using System.Threading.Tasks;

namespace MQTTnet.Core.Internal
{
public sealed class AsyncGate
{
private readonly Queue<TaskCompletionSource<bool>> _waitingTasks = new Queue<TaskCompletionSource<bool>>();

public Task WaitOneAsync()
{
var tcs = new TaskCompletionSource<bool>();
lock (_waitingTasks)
{
_waitingTasks.Enqueue(tcs);
}

return tcs.Task;
}

public void Set()
{
lock (_waitingTasks)
{
if (_waitingTasks.Count > 0)
{
_waitingTasks.Dequeue().SetResult(true);
}
}
}
}
}

+ 55
- 0
MQTTnet.Core/Internal/TaskExtensions.cs View File

@@ -0,0 +1,55 @@
using System;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Core.Exceptions;

namespace MQTTnet.Core.Internal
{
public static class TaskExtensions
{
public static Task TimeoutAfter( this Task task, TimeSpan timeout )
{
return TimeoutAfter( task.ContinueWith( t => 0 ), timeout );
}

public static async Task<TResult> TimeoutAfter<TResult>(this Task<TResult> task, TimeSpan timeout)
{
using (var cancellationTokenSource = new CancellationTokenSource())
{
var tcs = new TaskCompletionSource<TResult>();

cancellationTokenSource.Token.Register(() =>
{
tcs.TrySetCanceled();
} );
try
{
cancellationTokenSource.CancelAfter(timeout);
task.ContinueWith( t =>
{
if (t.IsFaulted)
{
tcs.TrySetException(t.Exception);
}

if (t.IsCompleted)
{
tcs.TrySetResult(t.Result);
}
}, cancellationTokenSource.Token );

return await tcs.Task;
}
catch (TaskCanceledException)
{
throw new MqttCommunicationTimedOutException();
}
catch (Exception e)
{
throw new MqttCommunicationException(e);
}
}
}
}
}

+ 1
- 0
MQTTnet.Core/MQTTnet.Core.csproj View File

@@ -5,6 +5,7 @@
<AssemblyName>MQTTnet.Core</AssemblyName> <AssemblyName>MQTTnet.Core</AssemblyName>
<RootNamespace>MQTTnet.Core</RootNamespace> <RootNamespace>MQTTnet.Core</RootNamespace>
<GeneratePackageOnBuild>False</GeneratePackageOnBuild> <GeneratePackageOnBuild>False</GeneratePackageOnBuild>
<DebugType>Full</DebugType>
<Description></Description> <Description></Description>
<Product></Product> <Product></Product>
<Company></Company> <Company></Company>


+ 3
- 4
MQTTnet.Core/Serializer/IMqttPacketSerializer.cs View File

@@ -1,5 +1,4 @@
using System.Threading.Tasks;
using MQTTnet.Core.Channel;
using System.IO;
using MQTTnet.Core.Packets; using MQTTnet.Core.Packets;


namespace MQTTnet.Core.Serializer namespace MQTTnet.Core.Serializer
@@ -8,8 +7,8 @@ namespace MQTTnet.Core.Serializer
{ {
MqttProtocolVersion ProtocolVersion { get; set; } MqttProtocolVersion ProtocolVersion { get; set; }


Task SerializeAsync(MqttBasePacket mqttPacket, IMqttCommunicationChannel destination);
byte[] Serialize(MqttBasePacket mqttPacket);


Task<MqttBasePacket> DeserializeAsync(IMqttCommunicationChannel source);
MqttBasePacket Deserialize(MqttPacketHeader header, MemoryStream stream);
} }
} }

+ 7
- 27
MQTTnet.Core/Serializer/MqttPacketReader.cs View File

@@ -14,7 +14,7 @@ namespace MQTTnet.Core.Serializer
private readonly MqttPacketHeader _header; private readonly MqttPacketHeader _header;


public MqttPacketReader(Stream stream, MqttPacketHeader header) public MqttPacketReader(Stream stream, MqttPacketHeader header)
: base(stream)
: base(stream, Encoding.UTF8, true)
{ {
_header = header; _header = header;
} }
@@ -49,13 +49,11 @@ namespace MQTTnet.Core.Serializer
return ReadBytes(_header.BodyLength - (int)BaseStream.Position); return ReadBytes(_header.BodyLength - (int)BaseStream.Position);
} }


public static async Task<MqttPacketHeader> ReadHeaderFromSourceAsync(IMqttCommunicationChannel source, byte[] buffer)
public static MqttPacketHeader ReadHeaderFromSource(Stream stream)
{ {
var fixedHeader = await ReadStreamByteAsync(source, buffer).ConfigureAwait(false);
var byteReader = new ByteReader(fixedHeader);
byteReader.Read(4);
var controlPacketType = (MqttControlPacketType)byteReader.Read(4);
var bodyLength = await ReadBodyLengthFromSourceAsync(source, buffer).ConfigureAwait(false);
var fixedHeader = (byte)stream.ReadByte();
var controlPacketType = (MqttControlPacketType)(fixedHeader >> 4);
var bodyLength = ReadBodyLengthFromSource(stream);


return new MqttPacketHeader() return new MqttPacketHeader()
{ {
@@ -64,26 +62,8 @@ namespace MQTTnet.Core.Serializer
BodyLength = bodyLength BodyLength = bodyLength
}; };
} }

private static async Task<byte> ReadStreamByteAsync(IMqttCommunicationChannel source, byte[] readBuffer)
{
var result = await ReadFromSourceAsync(source, 1, readBuffer).ConfigureAwait(false);
return result.Array[result.Offset];
}

public static async Task<ArraySegment<byte>> ReadFromSourceAsync(IMqttCommunicationChannel source, int length, byte[] buffer)
{
try
{
return await source.ReadAsync(length, buffer);
}
catch (Exception exception)
{
throw new MqttCommunicationException(exception);
}
}
private static async Task<int> ReadBodyLengthFromSourceAsync(IMqttCommunicationChannel source, byte[] buffer)
private static int ReadBodyLengthFromSource(Stream stream)
{ {
// Alorithm taken from http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html. // Alorithm taken from http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html.
var multiplier = 1; var multiplier = 1;
@@ -91,7 +71,7 @@ namespace MQTTnet.Core.Serializer
byte encodedByte; byte encodedByte;
do do
{ {
encodedByte = await ReadStreamByteAsync(source, buffer).ConfigureAwait(false);
encodedByte = (byte)stream.ReadByte();
value += (encodedByte & 127) * multiplier; value += (encodedByte & 127) * multiplier;
multiplier *= 128; multiplier *= 128;
if (multiplier > 128 * 128 * 128) if (multiplier > 128 * 128 * 128)


+ 120
- 118
MQTTnet.Core/Serializer/MqttPacketSerializer.cs View File

@@ -17,12 +17,10 @@ namespace MQTTnet.Core.Serializer
private static byte[] ProtocolVersionV310Name { get; } = Encoding.UTF8.GetBytes("MQIs"); private static byte[] ProtocolVersionV310Name { get; } = Encoding.UTF8.GetBytes("MQIs");


public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311; public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311;
private byte[] _readBuffer = new byte[BufferConstants.Size];


public async Task SerializeAsync(MqttBasePacket packet, IMqttCommunicationChannel destination)
public byte[] Serialize(MqttBasePacket packet)
{ {
if (packet == null) throw new ArgumentNullException(nameof(packet)); if (packet == null) throw new ArgumentNullException(nameof(packet));
if (destination == null) throw new ArgumentNullException(nameof(destination));


using (var stream = new MemoryStream()) using (var stream = new MemoryStream())
using (var writer = new MqttPacketWriter(stream)) using (var writer = new MqttPacketWriter(stream))
@@ -32,9 +30,12 @@ namespace MQTTnet.Core.Serializer


var body = stream.ToArray(); var body = stream.ToArray();
MqttPacketWriter.BuildLengthHeader(body.Length, header); MqttPacketWriter.BuildLengthHeader(body.Length, header);
await destination.WriteAsync(header.ToArray()).ConfigureAwait(false);
await destination.WriteAsync(body).ConfigureAwait(false);
var headerArray = header.ToArray();
var writeBuffer = new byte[header.Count + body.Length];
Buffer.BlockCopy( headerArray, 0, writeBuffer, 0, headerArray.Length );
Buffer.BlockCopy( body, 0, writeBuffer, headerArray.Length, body.Length );

return writeBuffer;
} }
} }


@@ -113,120 +114,112 @@ namespace MQTTnet.Core.Serializer
throw new MqttProtocolViolationException("Packet type invalid."); throw new MqttProtocolViolationException("Packet type invalid.");
} }


public async Task<MqttBasePacket> DeserializeAsync(IMqttCommunicationChannel source)
public MqttBasePacket Deserialize(MqttPacketHeader header, MemoryStream body)
{ {
if (source == null) throw new ArgumentNullException(nameof(source));

var header = await MqttPacketReader.ReadHeaderFromSourceAsync(source, _readBuffer).ConfigureAwait(false);

var body = await GetBody(source, header).ConfigureAwait(false);

using (var mqttPacketReader = new MqttPacketReader(body, header))
if (header == null) throw new ArgumentNullException(nameof(header));
if (body == null) throw new ArgumentNullException(nameof(body));
using (var reader = new MqttPacketReader(body, header))
{ {
switch (header.ControlPacketType)
{
case MqttControlPacketType.Connect:
{
return DeserializeConnect(mqttPacketReader);
}

case MqttControlPacketType.ConnAck:
{
return DeserializeConnAck(mqttPacketReader);
}

case MqttControlPacketType.Disconnect:
{
return new MqttDisconnectPacket();
}

case MqttControlPacketType.Publish:
{
return DeserializePublish(mqttPacketReader, header);
}

case MqttControlPacketType.PubAck:
{
return new MqttPubAckPacket
{
PacketIdentifier = mqttPacketReader.ReadUInt16()
};
}

case MqttControlPacketType.PubRec:
{
return new MqttPubRecPacket
{
PacketIdentifier = mqttPacketReader.ReadUInt16()
};
}

case MqttControlPacketType.PubRel:
{
return new MqttPubRelPacket
{
PacketIdentifier = mqttPacketReader.ReadUInt16()
};
}

case MqttControlPacketType.PubComp:
{
return new MqttPubCompPacket
{
PacketIdentifier = mqttPacketReader.ReadUInt16()
};
}

case MqttControlPacketType.PingReq:
{
return new MqttPingReqPacket();
}

case MqttControlPacketType.PingResp:
{
return new MqttPingRespPacket();
}

case MqttControlPacketType.Subscribe:
{
return DeserializeSubscribe(mqttPacketReader);
}

case MqttControlPacketType.SubAck:
{
return DeserializeSubAck(mqttPacketReader);
}

case MqttControlPacketType.Unsubscibe:
{
return DeserializeUnsubscribe(mqttPacketReader);
}

case MqttControlPacketType.UnsubAck:
{
return new MqttUnsubAckPacket
{
PacketIdentifier = mqttPacketReader.ReadUInt16()
};
}

default:
{
throw new MqttProtocolViolationException($"Packet type ({(int)header.ControlPacketType}) not supported.");
}
}
return Deserialize( header, reader );
} }
} }


private async Task<MemoryStream> GetBody(IMqttCommunicationChannel source, MqttPacketHeader header)
private static MqttBasePacket Deserialize(MqttPacketHeader header, MqttPacketReader reader)
{ {
if (header.BodyLength > 0)
switch (header.ControlPacketType)
{ {
var segment = await MqttPacketReader.ReadFromSourceAsync(source, header.BodyLength, _readBuffer).ConfigureAwait(false);
return new MemoryStream(segment.Array, segment.Offset, segment.Count);
}
case MqttControlPacketType.Connect:
{
return DeserializeConnect( reader );
}

case MqttControlPacketType.ConnAck:
{
return DeserializeConnAck( reader );
}

case MqttControlPacketType.Disconnect:
{
return new MqttDisconnectPacket();
}

case MqttControlPacketType.Publish:
{
return DeserializePublish( reader, header );
}

case MqttControlPacketType.PubAck:
{
return new MqttPubAckPacket
{
PacketIdentifier = reader.ReadUInt16()
};
}

case MqttControlPacketType.PubRec:
{
return new MqttPubRecPacket
{
PacketIdentifier = reader.ReadUInt16()
};
}

case MqttControlPacketType.PubRel:
{
return new MqttPubRelPacket
{
PacketIdentifier = reader.ReadUInt16()
};
}

case MqttControlPacketType.PubComp:
{
return new MqttPubCompPacket
{
PacketIdentifier = reader.ReadUInt16()
};
}

case MqttControlPacketType.PingReq:
{
return new MqttPingReqPacket();
}


return new MemoryStream();
case MqttControlPacketType.PingResp:
{
return new MqttPingRespPacket();
}

case MqttControlPacketType.Subscribe:
{
return DeserializeSubscribe( reader );
}

case MqttControlPacketType.SubAck:
{
return DeserializeSubAck( reader );
}

case MqttControlPacketType.Unsubscibe:
{
return DeserializeUnsubscribe( reader );
}

case MqttControlPacketType.UnsubAck:
{
return new MqttUnsubAckPacket
{
PacketIdentifier = reader.ReadUInt16()
};
}

default:
{
throw new MqttProtocolViolationException(
$"Packet type ({(int) header.ControlPacketType}) not supported." );
}
}
} }


private static MqttBasePacket DeserializeUnsubscribe(MqttPacketReader reader) private static MqttBasePacket DeserializeUnsubscribe(MqttPacketReader reader)
@@ -267,7 +260,7 @@ namespace MQTTnet.Core.Serializer
var retain = fixedHeader.Read(); var retain = fixedHeader.Read();
var qualityOfServiceLevel = (MqttQualityOfServiceLevel)fixedHeader.Read(2); var qualityOfServiceLevel = (MqttQualityOfServiceLevel)fixedHeader.Read(2);
var dup = fixedHeader.Read(); var dup = fixedHeader.Read();
var topic = reader.ReadStringWithLengthPrefix(); var topic = reader.ReadStringWithLengthPrefix();


ushort packetIdentifier = 0; ushort packetIdentifier = 0;
@@ -519,12 +512,21 @@ namespace MQTTnet.Core.Serializer
writer.Write(packet.Payload); writer.Write(packet.Payload);
} }


var fixedHeader = new ByteWriter();
fixedHeader.Write(packet.Retain);
fixedHeader.Write((byte)packet.QualityOfServiceLevel, 2);
fixedHeader.Write(packet.Dup);
byte fixedHeader = 0;

if (packet.Retain)
{
fixedHeader |= 0x01;
}


return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader.Value);
fixedHeader |= (byte)((byte)packet.QualityOfServiceLevel << 1);

if ( packet.Dup )
{
fixedHeader |= 0x08;
}
return MqttPacketWriter.BuildFixedHeader(MqttControlPacketType.Publish, fixedHeader);
} }


private static byte Serialize(MqttPubAckPacket packet, MqttPacketWriter writer) private static byte Serialize(MqttPubAckPacket packet, MqttPacketWriter writer)


+ 13
- 39
MQTTnet.Core/Server/MqttClientMessageQueue.cs View File

@@ -1,20 +1,17 @@
using System; using System;
using System.Collections.Generic;
using System.Linq;
using System.Collections.Concurrent;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Core.Adapter; using MQTTnet.Core.Adapter;
using MQTTnet.Core.Diagnostics; using MQTTnet.Core.Diagnostics;
using MQTTnet.Core.Exceptions; using MQTTnet.Core.Exceptions;
using MQTTnet.Core.Internal;
using MQTTnet.Core.Packets; using MQTTnet.Core.Packets;


namespace MQTTnet.Core.Server namespace MQTTnet.Core.Server
{ {
public sealed class MqttClientMessageQueue public sealed class MqttClientMessageQueue
{ {
private readonly List<MqttClientPublishPacketContext> _pendingPublishPackets = new List<MqttClientPublishPacketContext>();
private readonly AsyncGate _gate = new AsyncGate();
private readonly BlockingCollection<MqttClientPublishPacketContext> _pendingPublishPackets = new BlockingCollection<MqttClientPublishPacketContext>();


private readonly MqttServerOptions _options; private readonly MqttServerOptions _options;
private CancellationTokenSource _cancellationTokenSource; private CancellationTokenSource _cancellationTokenSource;
@@ -43,55 +40,38 @@ namespace MQTTnet.Core.Server
_adapter = null; _adapter = null;
_cancellationTokenSource?.Cancel(); _cancellationTokenSource?.Cancel();
_cancellationTokenSource = null; _cancellationTokenSource = null;
_pendingPublishPackets?.Dispose();
} }


public void Enqueue(MqttPublishPacket publishPacket) public void Enqueue(MqttPublishPacket publishPacket)
{ {
if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));


lock (_pendingPublishPackets)
{
_pendingPublishPackets.Add(new MqttClientPublishPacketContext(publishPacket));
_gate.Set();
}
_pendingPublishPackets.Add( new MqttClientPublishPacketContext( publishPacket ) );
} }


private async Task SendPendingPublishPacketsAsync(CancellationToken cancellationToken) private async Task SendPendingPublishPacketsAsync(CancellationToken cancellationToken)
{ {
while (!cancellationToken.IsCancellationRequested)
foreach (var publishPacket in _pendingPublishPackets.GetConsumingEnumerable(cancellationToken))
{ {
try try
{ {
await _gate.WaitOneAsync().ConfigureAwait(false);
if (cancellationToken.IsCancellationRequested)
if ( cancellationToken.IsCancellationRequested )
{ {
return; return;
} }


if (_adapter == null)
if ( _adapter == null )
{ {
continue; continue;
} }


List<MqttClientPublishPacketContext> pendingPublishPackets;
lock (_pendingPublishPackets)
{
pendingPublishPackets = _pendingPublishPackets.ToList();
}

foreach (var publishPacket in pendingPublishPackets)
{
await TrySendPendingPublishPacketAsync(publishPacket).ConfigureAwait(false);
}
await TrySendPendingPublishPacketAsync( publishPacket ).ConfigureAwait( false );
} }
catch (Exception e)
catch ( Exception e )
{ {
MqttTrace.Error(nameof(MqttClientMessageQueue), e, "Error while sending pending publish packets.");
MqttTrace.Error( nameof( MqttClientMessageQueue ), e, "Error while sending pending publish packets." );
} }
finally
{
Cleanup();
}
} }
} }


@@ -105,30 +85,24 @@ namespace MQTTnet.Core.Server
} }


publishPacketContext.PublishPacket.Dup = publishPacketContext.SendTries > 0; publishPacketContext.PublishPacket.Dup = publishPacketContext.SendTries > 0;
await _adapter.SendPacketAsync(publishPacketContext.PublishPacket, _options.DefaultCommunicationTimeout).ConfigureAwait(false);
await _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, publishPacketContext.PublishPacket).ConfigureAwait(false);


publishPacketContext.IsSent = true; publishPacketContext.IsSent = true;
} }
catch (MqttCommunicationException exception) catch (MqttCommunicationException exception)
{ {
MqttTrace.Warning(nameof(MqttClientMessageQueue), exception, "Sending publish packet failed."); MqttTrace.Warning(nameof(MqttClientMessageQueue), exception, "Sending publish packet failed.");
_pendingPublishPackets.Add( publishPacketContext );
} }
catch (Exception exception) catch (Exception exception)
{ {
MqttTrace.Error(nameof(MqttClientMessageQueue), exception, "Sending publish packet failed."); MqttTrace.Error(nameof(MqttClientMessageQueue), exception, "Sending publish packet failed.");
_pendingPublishPackets.Add( publishPacketContext );
} }
finally finally
{ {
publishPacketContext.SendTries++; publishPacketContext.SendTries++;
} }
} }

private void Cleanup()
{
lock (_pendingPublishPackets)
{
_pendingPublishPackets.RemoveAll(p => p.IsSent);
}
}
} }
} }

+ 7
- 7
MQTTnet.Core/Server/MqttClientSession.cs View File

@@ -103,12 +103,12 @@ namespace MQTTnet.Core.Server
{ {
if (packet is MqttSubscribePacket subscribePacket) if (packet is MqttSubscribePacket subscribePacket)
{ {
return Adapter.SendPacketAsync(_subscriptionsManager.Subscribe(subscribePacket), _options.DefaultCommunicationTimeout);
return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _subscriptionsManager.Subscribe(subscribePacket));
} }


if (packet is MqttUnsubscribePacket unsubscribePacket) if (packet is MqttUnsubscribePacket unsubscribePacket)
{ {
return Adapter.SendPacketAsync(_subscriptionsManager.Unsubscribe(unsubscribePacket), _options.DefaultCommunicationTimeout);
return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, _subscriptionsManager.Unsubscribe(unsubscribePacket));
} }


if (packet is MqttPublishPacket publishPacket) if (packet is MqttPublishPacket publishPacket)
@@ -123,7 +123,7 @@ namespace MQTTnet.Core.Server


if (packet is MqttPubRecPacket pubRecPacket) if (packet is MqttPubRecPacket pubRecPacket)
{ {
return Adapter.SendPacketAsync(pubRecPacket.CreateResponse<MqttPubRelPacket>(), _options.DefaultCommunicationTimeout);
return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, pubRecPacket.CreateResponse<MqttPubRelPacket>());
} }


if (packet is MqttPubAckPacket || packet is MqttPubCompPacket) if (packet is MqttPubAckPacket || packet is MqttPubCompPacket)
@@ -134,7 +134,7 @@ namespace MQTTnet.Core.Server


if (packet is MqttPingReqPacket) if (packet is MqttPingReqPacket)
{ {
return Adapter.SendPacketAsync(new MqttPingRespPacket(), _options.DefaultCommunicationTimeout);
return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new MqttPingRespPacket());
} }


if (packet is MqttDisconnectPacket || packet is MqttConnectPacket) if (packet is MqttDisconnectPacket || packet is MqttConnectPacket)
@@ -160,7 +160,7 @@ namespace MQTTnet.Core.Server
if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce) if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtLeastOnce)
{ {
_publishPacketReceivedCallback(this, publishPacket); _publishPacketReceivedCallback(this, publishPacket);
return Adapter.SendPacketAsync(new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier });
} }


if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce) if (publishPacket.QualityOfServiceLevel == MqttQualityOfServiceLevel.ExactlyOnce)
@@ -173,7 +173,7 @@ namespace MQTTnet.Core.Server


_publishPacketReceivedCallback(this, publishPacket); _publishPacketReceivedCallback(this, publishPacket);


return Adapter.SendPacketAsync(new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier });
} }


throw new MqttCommunicationException("Received a not supported QoS level."); throw new MqttCommunicationException("Received a not supported QoS level.");
@@ -186,7 +186,7 @@ namespace MQTTnet.Core.Server
_unacknowledgedPublishPackets.Remove(pubRelPacket.PacketIdentifier); _unacknowledgedPublishPackets.Remove(pubRelPacket.PacketIdentifier);
} }


return Adapter.SendPacketAsync(new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier }, _options.DefaultCommunicationTimeout);
return Adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier });
} }
} }
} }

+ 4
- 4
MQTTnet.Core/Server/MqttClientSessionsManager.cs View File

@@ -40,21 +40,21 @@ namespace MQTTnet.Core.Server
var connectReturnCode = ValidateConnection(connectPacket); var connectReturnCode = ValidateConnection(connectPacket);
if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted) if (connectReturnCode != MqttConnectReturnCode.ConnectionAccepted)
{ {
await eventArgs.ClientAdapter.SendPacketAsync(new MqttConnAckPacket
await eventArgs.ClientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new MqttConnAckPacket
{ {
ConnectReturnCode = connectReturnCode ConnectReturnCode = connectReturnCode
}, _options.DefaultCommunicationTimeout).ConfigureAwait(false);
}).ConfigureAwait(false);


return; return;
} }


var clientSession = GetOrCreateClientSession(connectPacket); var clientSession = GetOrCreateClientSession(connectPacket);


await eventArgs.ClientAdapter.SendPacketAsync(new MqttConnAckPacket
await eventArgs.ClientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, new MqttConnAckPacket
{ {
ConnectReturnCode = connectReturnCode, ConnectReturnCode = connectReturnCode,
IsSessionPresent = clientSession.IsExistingSession IsSessionPresent = clientSession.IsExistingSession
}, _options.DefaultCommunicationTimeout).ConfigureAwait(false);
}).ConfigureAwait(false);


await clientSession.Session.RunAsync(eventArgs.Identifier, connectPacket.WillMessage, eventArgs.ClientAdapter).ConfigureAwait(false); await clientSession.Session.RunAsync(eventArgs.Identifier, connectPacket.WillMessage, eventArgs.ClientAdapter).ConfigureAwait(false);
} }


+ 1
- 0
MQTTnet.TestApp.NetCore/MQTTnet.TestApp.NetCore.csproj View File

@@ -2,6 +2,7 @@


<PropertyGroup> <PropertyGroup>
<OutputType>Exe</OutputType> <OutputType>Exe</OutputType>
<DebugType>Full</DebugType>
<TargetFramework>netcoreapp2.0</TargetFramework> <TargetFramework>netcoreapp2.0</TargetFramework>
</PropertyGroup> </PropertyGroup>




+ 33
- 0
Tests/MQTTnet.Core.Tests/ExtensionTests.cs View File

@@ -0,0 +1,33 @@
using System;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Core.Exceptions;
using MQTTnet.Core.Internal;

namespace MQTTnet.Core.Tests
{
[TestClass]
public class ExtensionTests
{
[ExpectedException(typeof( MqttCommunicationTimedOutException ) )]
[TestMethod]
public async Task TestTimeoutAfter()
{
await Task.Delay(TimeSpan.FromMilliseconds(500)).TimeoutAfter(TimeSpan.FromMilliseconds(100));
}

[ExpectedException(typeof( MqttCommunicationTimedOutException))]
[TestMethod]
public async Task TestTimeoutAfterWithResult()
{
await Task.Delay(TimeSpan.FromMilliseconds(500)).ContinueWith(t => 5).TimeoutAfter(TimeSpan.FromMilliseconds(100));
}
[TestMethod]
public async Task TestTimeoutAfterCompleteInTime()
{
var result = await Task.Delay( TimeSpan.FromMilliseconds( 100 ) ).ContinueWith( t => 5 ).TimeoutAfter( TimeSpan.FromMilliseconds( 500 ) );
Assert.AreEqual( 5, result );
}
}
}

+ 1
- 0
Tests/MQTTnet.Core.Tests/MQTTnet.Core.Tests.csproj View File

@@ -86,6 +86,7 @@
<ItemGroup> <ItemGroup>
<Compile Include="ByteReaderTests.cs" /> <Compile Include="ByteReaderTests.cs" />
<Compile Include="ByteWriterTests.cs" /> <Compile Include="ByteWriterTests.cs" />
<Compile Include="ExtensionTests.cs" />
<Compile Include="MqttPacketSerializerTests.cs" /> <Compile Include="MqttPacketSerializerTests.cs" />
<Compile Include="MqttServerTests.cs" /> <Compile Include="MqttServerTests.cs" />
<Compile Include="MqttSubscriptionsManagerTests.cs" /> <Compile Include="MqttSubscriptionsManagerTests.cs" />


+ 19
- 29
Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs View File

@@ -391,6 +391,12 @@ namespace MQTTnet.Core.Tests
{ {
private readonly MemoryStream _stream = new MemoryStream(); private readonly MemoryStream _stream = new MemoryStream();


public Stream ReceiveStream => _stream;

public Stream RawStream => _stream;

public Stream SendStream => _stream;

public bool IsConnected { get; } = true; public bool IsConnected { get; } = true;


public TestChannel() public TestChannel()
@@ -413,34 +419,16 @@ namespace MQTTnet.Core.Tests
return Task.FromResult(0); return Task.FromResult(0);
} }


public Task WriteAsync(byte[] buffer)
{
return _stream.WriteAsync(buffer, 0, buffer.Length);
}

public async Task<ArraySegment<byte>> ReadAsync(int length, byte[] buffer)
{
await _stream.ReadAsync(buffer, 0, length);
return new ArraySegment<byte>(buffer, 0, length);
}

public byte[] ToArray() public byte[] ToArray()
{ {
return _stream.ToArray(); return _stream.ToArray();
} }

public int Peek()
{
return (int)_stream.Length - (int)_stream.Position;
}
} }


private static void SerializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311) private static void SerializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
{ {
var serializer = new MqttPacketSerializer { ProtocolVersion = protocolVersion }; var serializer = new MqttPacketSerializer { ProtocolVersion = protocolVersion };
var channel = new TestChannel();
serializer.SerializeAsync(packet, channel).Wait();
var buffer = channel.ToArray();
var buffer = serializer.Serialize(packet);


Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(buffer)); Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(buffer));
} }
@@ -448,19 +436,21 @@ namespace MQTTnet.Core.Tests
private static void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value) private static void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value)
{ {
var serializer = new MqttPacketSerializer(); var serializer = new MqttPacketSerializer();
var buffer1 = serializer.Serialize(packet);


var channel1 = new TestChannel();
serializer.SerializeAsync(packet, channel1).Wait();
var buffer1 = channel1.ToArray();

var channel2 = new TestChannel(buffer1);
var deserializedPacket = serializer.DeserializeAsync(channel2).Result;
var buffer2 = channel2.ToArray();
using (var headerStream = new MemoryStream( buffer1 ))
{
var header = MqttPacketReader.ReadHeaderFromSource( headerStream );


var channel3 = new TestChannel(buffer2);
serializer.SerializeAsync(deserializedPacket, channel3).Wait();
using (var bodyStream = new MemoryStream( buffer1, (int)headerStream.Position, header.BodyLength ))
{
var deserializedPacket = serializer.Deserialize(header, bodyStream);
var buffer2 = serializer.Serialize( deserializedPacket );


Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(channel3.ToArray()));
Assert.AreEqual( expectedBase64Value, Convert.ToBase64String( buffer2 ) );
}
}
} }
} }
} }

+ 13
- 2
Tests/MQTTnet.Core.Tests/TestMqttCommunicationAdapter.cs View File

@@ -1,5 +1,7 @@
using System; using System;
using System.Collections.Concurrent; using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using MQTTnet.Core.Adapter; using MQTTnet.Core.Adapter;
using MQTTnet.Core.Client; using MQTTnet.Core.Client;
@@ -26,11 +28,15 @@ namespace MQTTnet.Core.Tests
return Task.FromResult(0); return Task.FromResult(0);
} }


public Task SendPacketAsync(MqttBasePacket packet, TimeSpan timeout)
public Task SendPacketsAsync(TimeSpan timeout, IEnumerable<MqttBasePacket> packets)
{ {
ThrowIfPartnerIsNull(); ThrowIfPartnerIsNull();


Partner.SendPacketInternal(packet);
foreach (var packet in packets)
{
Partner.SendPacketInternal(packet);
}

return Task.FromResult(0); return Task.FromResult(0);
} }


@@ -41,6 +47,11 @@ namespace MQTTnet.Core.Tests
return Task.Run(() => _incomingPackets.Take()); return Task.Run(() => _incomingPackets.Take());
} }


public IEnumerable<MqttBasePacket> ReceivePackets( CancellationToken cancellationToken )
{
return _incomingPackets.GetConsumingEnumerable();
}

private void SendPacketInternal(MqttBasePacket packet) private void SendPacketInternal(MqttBasePacket packet)
{ {
if (packet == null) throw new ArgumentNullException(nameof(packet)); if (packet == null) throw new ArgumentNullException(nameof(packet));


+ 43
- 15
Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs View File

@@ -7,6 +7,7 @@ using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq; using System.Linq;
using System.Text; using System.Text;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;


namespace MQTTnet.TestApp.NetFramework namespace MQTTnet.TestApp.NetFramework
@@ -16,7 +17,7 @@ namespace MQTTnet.TestApp.NetFramework
public static async Task RunAsync() public static async Task RunAsync()
{ {
var server = Task.Run(() => RunServerAsync()); var server = Task.Run(() => RunServerAsync());
var client = Task.Run(() => RunClientAsync(50, TimeSpan.FromMilliseconds(10)));
var client = Task.Run(() => RunClientAsync(300, TimeSpan.FromMilliseconds(10)));


await Task.WhenAll(server, client).ConfigureAwait(false); await Task.WhenAll(server, client).ConfigureAwait(false);
} }
@@ -77,29 +78,37 @@ namespace MQTTnet.TestApp.NetFramework
Console.WriteLine("### WAITING FOR APPLICATION MESSAGES ###"); Console.WriteLine("### WAITING FOR APPLICATION MESSAGES ###");


var last = DateTime.Now; var last = DateTime.Now;
var msgs = 0;
var msgCount = 0;


while (true) while (true)
{ {
for (int i = 0; i < msgChunkSize; i++)
var msgs = Enumerable.Range( 0, msgChunkSize )
.Select( i => CreateMessage() )
.ToList();

if (false)
{ {
var applicationMessage = new MqttApplicationMessage(
"A/B/C",
Encoding.UTF8.GetBytes("Hello World"),
MqttQualityOfServiceLevel.AtLeastOnce,
false
);

//do not await to send as much messages as possible
await client.PublishAsync(applicationMessage);
msgs++;
//send concurrent (test for raceconditions)
var sendTasks = msgs
.Select( msg => PublishSingleMessage( client, msg, ref msgCount ) )
.ToList();

await Task.WhenAll( sendTasks );
} }
else
{
await client.PublishAsync( msgs );
msgCount += msgs.Count;
//send multiple
}



var now = DateTime.Now; var now = DateTime.Now;
if (last < now - TimeSpan.FromSeconds(1)) if (last < now - TimeSpan.FromSeconds(1))
{ {
Console.WriteLine( $"sending {msgs} inteded {msgChunkSize / interval.TotalSeconds}" );
msgs = 0;
Console.WriteLine( $"sending {msgCount} inteded {msgChunkSize / interval.TotalSeconds}" );
msgCount = 0;
last = now; last = now;
} }


@@ -112,6 +121,25 @@ namespace MQTTnet.TestApp.NetFramework
} }
} }


private static MqttApplicationMessage CreateMessage()
{
return new MqttApplicationMessage(
"A/B/C",
Encoding.UTF8.GetBytes( "Hello World" ),
MqttQualityOfServiceLevel.AtMostOnce,
false
);
}

private static Task PublishSingleMessage( IMqttClient client, MqttApplicationMessage applicationMessage, ref int count )
{
Interlocked.Increment( ref count );
return Task.Run( () =>
{
return client.PublishAsync( applicationMessage );
} );
}

private static void RunServerAsync() private static void RunServerAsync()
{ {
try try


Loading…
Cancel
Save