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

Fix Unit Tests

Browse Source
This commit is contained in:
christian
2024-09-07 11:22:43 +02:00
parent f0ea5c9ecd
commit 9177b6ae68
5 changed files with 66 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet.Tests/MqttTcpChannel_Tests.cs
+56 -50
View File
@@ -2,13 +2,13 @@
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Implementations;
using System;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Implementations;
namespace MQTTnet.Tests
{
@@ -18,62 +18,68 @@ namespace MQTTnet.Tests
[TestMethod]
public async Task Dispose_Channel_While_Used()
{
var ct = new CancellationTokenSource();
var serverSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp);
try
using (var ct = new CancellationTokenSource())
{
serverSocket.Bind(new IPEndPoint(IPAddress.Any, 50001));
serverSocket.Listen(0);
#pragma warning disable 4014
Task.Run(async () =>
#pragma warning restore 4014
using (var serverSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp))
{
while (!ct.IsCancellationRequested)
try
{
var client = await serverSocket.AcceptAsync();
var data = new byte[] { 128 };
await client.SendAsync(new ArraySegment<byte>(data), SocketFlags.None);
}
}, ct.Token);
serverSocket.Bind(new IPEndPoint(IPAddress.Any, 50001));
serverSocket.Listen(0);
var clientSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp);
await clientSocket.ConnectAsync("localhost", 50001, CancellationToken.None);
var tcpChannel = new MqttTcpChannel(clientSocket.GetStream(), "test", null);
await Task.Delay(100, ct.Token);
var buffer = new byte[1];
await tcpChannel.ReadAsync(buffer, 0, 1, ct.Token);
Assert.AreEqual(128, buffer[0]);
// This block should fail after dispose.
#pragma warning disable 4014
Task.Run(() =>
Task.Run(
async () =>
#pragma warning restore 4014
{
Task.Delay(200, ct.Token);
tcpChannel.Dispose();
}, ct.Token);
{
while (!ct.IsCancellationRequested)
{
var client = await serverSocket.AcceptAsync(ct.Token);
var data = new byte[] { 128 };
await client.SendAsync(new ArraySegment<byte>(data), SocketFlags.None);
}
},
ct.Token);
try
{
await tcpChannel.ReadAsync(buffer, 0, 1, CancellationToken.None);
var clientSocket = new CrossPlatformSocket(AddressFamily.InterNetwork, ProtocolType.Tcp);
await clientSocket.ConnectAsync("localhost", 50001, CancellationToken.None);
var tcpChannel = new MqttTcpChannel(clientSocket.GetStream(), "test", null);
await Task.Delay(100, ct.Token);
var buffer = new byte[1];
await tcpChannel.ReadAsync(buffer, 0, 1, ct.Token);
Assert.AreEqual(128, buffer[0]);
// This block should fail after dispose.
#pragma warning disable 4014
Task.Run(
() =>
#pragma warning restore 4014
{
Task.Delay(200, ct.Token);
tcpChannel.Dispose();
},
ct.Token);
try
{
await tcpChannel.ReadAsync(buffer, 0, 1, CancellationToken.None);
}
catch (Exception exception)
{
Assert.IsInstanceOfType(exception, typeof(SocketException));
Assert.AreEqual(SocketError.OperationAborted, ((SocketException)exception).SocketErrorCode);
}
}
finally
{
ct.Cancel(false);
}
}
catch (Exception exception)
{
Assert.IsInstanceOfType(exception, typeof(SocketException));
Assert.AreEqual(SocketError.OperationAborted, ((SocketException)exception).SocketErrorCode);
}
}
finally
{
ct.Cancel(false);
serverSocket.Dispose();
}
}
}
}
}