YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-18 10:26:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix Unit Tests.

Browse Source
This commit is contained in:
Christian Kratky
2020-07-26 12:01:33 +02:00
parent 3260920828
commit 733c587e45
4 changed files with 52 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Build/MQTTnet.nuspec
+1 -1
View File
@@ -17,7 +17,7 @@
* [Server] Adjusted some namespaces (BREAKING CHANGE!)
* [Server] Added state checks (throw if not started etc.) for most server APIs.
* [Server] Exposed real X509Certificate2 (instead byte array) to TLS options (thanks to @borigas).
* [Server] Fixed memory leak with TCP sockets (MqttServer is now sealed!) (BREAKING CHANGE!)
* [Server] Fixed memory leak with TCP sockets (MqttServer is now Disposable!) (BREAKING CHANGE!)
* [Core] Fixed a null reference exception in the MqttTcpChannel with WriteAsync and ReadAsync.
* [Core] Added server interceptor for undelivered messages (thanks to @cshark-inator).
* [nuget] Added support for SourceLink (thanks to @JTOne123).
Source/MQTTnet/Adapter/MqttChannelAdapter.cs
+46 -51
View File
@@ -16,8 +16,8 @@ namespace MQTTnet.Adapter
{
public sealed class MqttChannelAdapter : Disposable, IMqttChannelAdapter
{
const uint _errorOperationAborted = 0x800703E3;
const int _readBufferSize = 4096; // TODO: Move buffer size to config
const uint ErrorOperationAborted = 0x800703E3;
const int ReadBufferSize = 4096;
readonly IMqttNetScopedLogger _logger;
readonly IMqttChannel _channel;
@@ -25,7 +25,7 @@ namespace MQTTnet.Adapter
readonly byte[] _fixedHeaderBuffer = new byte[2];
readonly SemaphoreSlim _writerSemaphore = new SemaphoreSlim(1, 1);
readonly AsyncLock _syncRoot = new AsyncLock();
long _bytesReceived;
long _bytesSent;
@@ -60,7 +60,7 @@ namespace MQTTnet.Adapter
{
cancellationToken.ThrowIfCancellationRequested();
ThrowIfDisposed();
try
{
if (timeout == TimeSpan.Zero)
@@ -87,7 +87,7 @@ namespace MQTTnet.Adapter
{
cancellationToken.ThrowIfCancellationRequested();
ThrowIfDisposed();
try
{
if (timeout == TimeSpan.Zero)
@@ -115,56 +115,51 @@ namespace MQTTnet.Adapter
{
cancellationToken.ThrowIfCancellationRequested();
ThrowIfDisposed();
try
{
await _writerSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
using (await _syncRoot.WaitAsync(cancellationToken).ConfigureAwait(false))
{
// Check for cancellation here again because "WaitAsync" might take some time.
cancellationToken.ThrowIfCancellationRequested();
try
{
var packetData = PacketFormatterAdapter.Encode(packet);
if (timeout == TimeSpan.Zero)
{
await _channel.WriteAsync(packetData.Array, packetData.Offset, packetData.Count, cancellationToken).ConfigureAwait(false);
}
else
{
await MqttTaskTimeout.WaitAsync(
t => _channel.WriteAsync(packetData.Array, packetData.Offset, packetData.Count, t), timeout, cancellationToken).ConfigureAwait(false);
}
Interlocked.Add(ref _bytesReceived, packetData.Count);
_logger.Verbose("TX ({0} bytes) >>> {1}", packetData.Count, packet);
}
catch (Exception exception)
{
if (IsWrappedException(exception))
{
throw;
}
WrapException(exception);
}
finally
{
PacketFormatterAdapter.FreeBuffer();
}
}
}
catch (ObjectDisposedException)
{
throw new OperationCanceledException();
}
try
{
var packetData = PacketFormatterAdapter.Encode(packet);
if (timeout == TimeSpan.Zero)
{
await _channel.WriteAsync(packetData.Array, packetData.Offset, packetData.Count, cancellationToken).ConfigureAwait(false);
}
else
{
await MqttTaskTimeout.WaitAsync(
t => _channel.WriteAsync(packetData.Array, packetData.Offset, packetData.Count, t), timeout, cancellationToken).ConfigureAwait(false);
}
Interlocked.Add(ref _bytesReceived, packetData.Count);
_logger.Verbose("TX ({0} bytes) >>> {1}", packetData.Count, packet);
}
catch (Exception exception)
{
if (IsWrappedException(exception))
{
throw;
}
WrapException(exception);
}
finally
{
PacketFormatterAdapter.FreeBuffer();
try
{
_writerSemaphore.Release();
}
catch (ObjectDisposedException)
{
throw new OperationCanceledException();
}
}
}
public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken)
@@ -236,7 +231,7 @@ namespace MQTTnet.Adapter
if (disposing)
{
_channel.Dispose();
_writerSemaphore.Dispose();
_syncRoot.Dispose();
}
base.Dispose(disposing);
@@ -268,7 +263,7 @@ namespace MQTTnet.Adapter
var body = new byte[fixedHeader.RemainingLength];
var bodyOffset = 0;
var chunkSize = Math.Min(_readBufferSize, fixedHeader.RemainingLength);
var chunkSize = Math.Min(ReadBufferSize, fixedHeader.RemainingLength);
do
{
@@ -327,7 +322,7 @@ namespace MQTTnet.Adapter
if (exception is COMException comException)
{
if ((uint)comException.HResult == _errorOperationAborted)
if ((uint)comException.HResult == ErrorOperationAborted)
{
throw new OperationCanceledException();
}
Tests/MQTTnet.Core.Tests/MqttClient_Tests.cs
+2
View File
@@ -125,6 +125,8 @@ namespace MQTTnet.Tests
await client1.PublishAsync("request", null, MqttQualityOfServiceLevel.AtLeastOnce);
await Task.Delay(500);
SpinWait.SpinUntil(() => replyReceived, TimeSpan.FromSeconds(10));
Assert.IsTrue(replyReceived);
Tests/MQTTnet.Core.Tests/Server_Tests.cs
+3 -3
View File
@@ -304,7 +304,7 @@ namespace MQTTnet.Tests
await c2.PublishAsync(messageBuilder.Build()).ConfigureAwait(false);
}
SpinWait.SpinUntil(() => receivedMessagesCount == 500, 2000);
SpinWait.SpinUntil(() => receivedMessagesCount == 500, 5000);
Assert.AreEqual(500, receivedMessagesCount);
}
@@ -325,7 +325,7 @@ namespace MQTTnet.Tests
for (var i = 0; i < 500; i++)
{
var so = new MqttClientSubscribeOptionsBuilder()
.WithTopicFilter(i.ToString(), MqttQualityOfServiceLevel.AtMostOnce).Build();
.WithTopicFilter(i.ToString()).Build();
await c1.SubscribeAsync(so).ConfigureAwait(false);
}
@@ -340,7 +340,7 @@ namespace MQTTnet.Tests
await c2.PublishAsync(messageBuilder.Build()).ConfigureAwait(false);
}
SpinWait.SpinUntil(() => receivedMessagesCount == 500, 2000);
SpinWait.SpinUntil(() => receivedMessagesCount == 500, 5000);
Assert.AreEqual(500, receivedMessagesCount);
}