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

Fix UnitTests.

Browse Source
This commit is contained in:
Christian
2021-10-29 22:35:32 +02:00
parent 6bad9f183c
commit 9d982eab5e
9 changed files with 46 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Adapter/MqttChannelAdapter.cs
+2 -2
View File
@@ -114,11 +114,11 @@ namespace MQTTnet.Adapter
var packetData = PacketFormatterAdapter.Encode(packet);
_packetInspectorHandler.BeginSendPacket(packetData);
_logger.Verbose("TX ({0} bytes) >>> {1}", packetData.Length, packet);
await packetData.WriteToAsync(_channel, cancellationToken).ConfigureAwait(false);
Interlocked.Add(ref _bytesReceived, packetData.Length);
_logger.Verbose("TX ({0} bytes) >>> {1}", packetData.Length, packet);
}
catch (Exception exception)
{
Source/MQTTnet/Implementations/MqttTcpChannel.cs
+15 -7
View File
@@ -24,7 +24,7 @@ namespace MQTTnet.Implementations
public MqttTcpChannel(IMqttClientOptions clientOptions)
{
_clientOptions = clientOptions ?? throw new ArgumentNullException(nameof(clientOptions));
_tcpOptions = (MqttClientTcpOptions)clientOptions.ChannelOptions;
_tcpOptions = (MqttClientTcpOptions) clientOptions.ChannelOptions;
IsSecureConnection = clientOptions.ChannelOptions?.TlsOptions?.UseTls == true;
}
@@ -61,7 +61,7 @@ namespace MQTTnet.Implementations
socket.ReceiveBufferSize = _tcpOptions.BufferSize;
socket.SendBufferSize = _tcpOptions.BufferSize;
socket.SendTimeout = (int)_clientOptions.CommunicationTimeout.TotalMilliseconds;
socket.SendTimeout = (int) _clientOptions.CommunicationTimeout.TotalMilliseconds;
socket.NoDelay = _tcpOptions.NoDelay;
if (_tcpOptions.DualMode.HasValue)
@@ -95,7 +95,8 @@ namespace MQTTnet.Implementations
await sslStream.AuthenticateAsClientAsync(sslOptions, cancellationToken).ConfigureAwait(false);
#else
await sslStream.AuthenticateAsClientAsync(_tcpOptions.Server, LoadCertificates(), _tcpOptions.TlsOptions.SslProtocol, !_tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
await sslStream.AuthenticateAsClientAsync(_tcpOptions.Server, LoadCertificates(), _tcpOptions.TlsOptions.SslProtocol,
!_tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors).ConfigureAwait(false);
#endif
}
catch
@@ -212,6 +213,9 @@ namespace MQTTnet.Implementations
// https://stackoverflow.com/questions/3601521/should-i-manually-dispose-the-socket-after-closing-it
try
{
#if !NETSTANDARD1_3
_stream?.Close();
#endif
_stream?.Dispose();
}
catch (ObjectDisposedException)
@@ -220,8 +224,10 @@ namespace MQTTnet.Implementations
catch (NullReferenceException)
{
}
_stream = null;
finally
{
_stream = null;
}
}
bool InternalUserCertificateValidationCallback(object sender, X509Certificate x509Certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
@@ -235,6 +241,7 @@ namespace MQTTnet.Implementations
{
return certificateValidationCallback(x509Certificate, chain, sslPolicyErrors, _clientOptions);
}
#endregion
var certificateValidationHandler = _tcpOptions?.TlsOptions?.CertificateValidationHandler;
@@ -256,7 +263,8 @@ namespace MQTTnet.Implementations
return true;
}
if (chain.ChainStatus.Any(c => c.Status == X509ChainStatusFlags.RevocationStatusUnknown || c.Status == X509ChainStatusFlags.Revoked || c.Status == X509ChainStatusFlags.OfflineRevocation))
if (chain.ChainStatus.Any(c =>
c.Status == X509ChainStatusFlags.RevocationStatusUnknown || c.Status == X509ChainStatusFlags.Revoked || c.Status == X509ChainStatusFlags.OfflineRevocation))
{
if (_tcpOptions?.TlsOptions?.IgnoreCertificateRevocationErrors != true)
{
@@ -292,4 +300,4 @@ namespace MQTTnet.Implementations
}
}
}
#endif
#endif
Source/MQTTnet/Server/Internal/MqttClientConnection.cs
+7
View File
@@ -138,6 +138,8 @@ namespace MQTTnet.Server
_= _sessionsManager.DispatchPublishPacket(Id, willApplicationMessage);
Session.WillMessageSent = true;
_logger.Info("Client '{0}': Published will message.", Id);
}
_logger.Info("Client '{0}': Connection stopped.", Id);
@@ -175,6 +177,11 @@ namespace MQTTnet.Server
while (!cancellationToken.IsCancellationRequested)
{
var packet = await ChannelAdapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false);
if (packet == null)
{
return;
}
var interceptingPacketEventArgs = new InterceptingPacketEventArgs
{
Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs
+3 -3
View File
@@ -518,15 +518,15 @@ namespace MQTTnet.Server
if (existingConnection != null)
{
existingConnection.IsTakenOver = true;
await existingConnection.StopAsync(MqttClientDisconnectReason.SessionTakenOver).ConfigureAwait(false);
await _eventContainer.ClientDisconnectedEvent.InvokeAsync(() => new MqttServerClientDisconnectedEventArgs
{
ClientId = existingConnection.Id,
DisconnectType = MqttClientDisconnectType.Takeover,
Endpoint = existingConnection.Endpoint
}).ConfigureAwait(false);
existingConnection.IsTakenOver = true;
await existingConnection.StopAsync(MqttClientDisconnectReason.SessionTakenOver).ConfigureAwait(false);
}
}
Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs
+1
View File
@@ -1,4 +1,5 @@
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using MQTTnet.Client;
Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs
+3 -4
View File
@@ -8,17 +8,15 @@ using System.Threading.Tasks;
namespace MQTTnet.Tests.MQTTv5
{
[TestClass]
public class Server_Tests
public class Server_Tests : BaseTestClass
{
public TestContext TestContext { get; set; }
[TestMethod]
public async Task Will_Message_Send()
{
using (var testEnvironment = new TestEnvironment(TestContext))
using (var testEnvironment = CreateTestEnvironment())
{
var receivedMessagesCount = 0;
await testEnvironment.StartServer();
var willMessage = new MqttApplicationMessageBuilder().WithTopic("My/last/will").WithAtMostOnceQoS().Build();
@@ -27,6 +25,7 @@ namespace MQTTnet.Tests.MQTTv5
var c1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder().WithProtocolVersion(MqttProtocolVersion.V500));
var receivedMessagesCount = 0;
c1.ApplicationMessageReceivedAsync += e =>
{
Interlocked.Increment(ref receivedMessagesCount);
Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs
+7 -7
View File
@@ -52,7 +52,7 @@ namespace MQTTnet.Tests.Mockups
remove => Implementation.ApplicationMessageReceivedAsync -= value;
}
public Task<MqttClientConnectResult> ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken)
public Task<MqttClientConnectResult> ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken = default)
{
if (TestContext != null)
{
@@ -68,7 +68,7 @@ namespace MQTTnet.Tests.Mockups
return Implementation.ConnectAsync(options, cancellationToken);
}
public Task DisconnectAsync(MqttClientDisconnectOptions options, CancellationToken cancellationToken)
public Task DisconnectAsync(MqttClientDisconnectOptions options, CancellationToken cancellationToken = default)
{
return Implementation.DisconnectAsync(options, cancellationToken);
}
@@ -78,27 +78,27 @@ namespace MQTTnet.Tests.Mockups
Implementation.Dispose();
}
public Task PingAsync(CancellationToken cancellationToken)
public Task PingAsync(CancellationToken cancellationToken = default)
{
return Implementation.PingAsync(cancellationToken);
}
public Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken)
public Task<MqttClientPublishResult> PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken = default)
{
return Implementation.PublishAsync(applicationMessage, cancellationToken);
}
public Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken)
public Task SendExtendedAuthenticationExchangeDataAsync(MqttExtendedAuthenticationExchangeData data, CancellationToken cancellationToken = default)
{
return Implementation.SendExtendedAuthenticationExchangeDataAsync(data, cancellationToken);
}
public Task<MqttClientSubscribeResult> SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken)
public Task<MqttClientSubscribeResult> SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken = default)
{
return Implementation.SubscribeAsync(options, cancellationToken);
}
public Task<MqttClientUnsubscribeResult> UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken)
public Task<MqttClientUnsubscribeResult> UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken = default)
{
return Implementation.UnsubscribeAsync(options, cancellationToken);
}
Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs
+5 -5
View File
@@ -83,7 +83,7 @@ namespace MQTTnet.Tests.Mockups
};
}
public IMqttClient CreateClient()
public TestClientWrapper CreateClient()
{
lock (_clients)
{
@@ -94,7 +94,7 @@ namespace MQTTnet.Tests.Mockups
}
}
public Task<IMqttClient> ConnectClient()
public Task<TestClientWrapper> ConnectClient()
{
return ConnectClient(Factory.CreateClientOptionsBuilder().WithProtocolVersion(_protocolVersion));
}
@@ -115,7 +115,7 @@ namespace MQTTnet.Tests.Mockups
return client;
}
public async Task<IMqttClient> ConnectClient(MqttClientOptionsBuilder options)
public async Task<TestClientWrapper> ConnectClient(MqttClientOptionsBuilder options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
@@ -127,7 +127,7 @@ namespace MQTTnet.Tests.Mockups
return client;
}
public async Task<IMqttClient> ConnectClient(IMqttClientOptions options)
public async Task<TestClientWrapper> ConnectClient(IMqttClientOptions options)
{
if (options == null) throw new ArgumentNullException(nameof(options));
@@ -267,7 +267,7 @@ namespace MQTTnet.Tests.Mockups
{
try
{
mqttClient.DisconnectAsync().GetAwaiter().GetResult();
//mqttClient.DisconnectAsync().GetAwaiter().GetResult();
}
catch
{
Tests/MQTTnet.Core.Tests/Server/General.cs
+3 -6
View File
@@ -190,7 +190,7 @@ namespace MQTTnet.Tests.Server
var clientOptions = new MqttClientOptionsBuilder().WithWillMessage(willMessage);
var c1 = await testEnvironment.ConnectClient();
var c1 = await testEnvironment.ConnectClient(new MqttClientOptionsBuilder());
var receivedMessagesCount = 0;
c1.ApplicationMessageReceivedAsync += e =>
@@ -200,10 +200,7 @@ namespace MQTTnet.Tests.Server
};
await c1.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("#").Build());
// clientOptions.WithClientId("Client2");
// var c2 = await testEnvironment.ConnectClient(clientOptions);
var c2 = await testEnvironment.ConnectClient(clientOptions);
c2.Dispose(); // Dispose will not send a DISCONNECT pattern first so the will message must be sent.
@@ -228,7 +225,7 @@ namespace MQTTnet.Tests.Server
Interlocked.Increment(ref receivedMessagesCount);
return Task.CompletedTask;
};
var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build();
await client.SubscribeAsync(new MqttTopicFilter {Topic = "a", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce});