diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index e7670af7..8076bdd1 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -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) { diff --git a/Source/MQTTnet/Implementations/MqttTcpChannel.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.cs index 11dabe6e..270a193e 100644 --- a/Source/MQTTnet/Implementations/MqttTcpChannel.cs +++ b/Source/MQTTnet/Implementations/MqttTcpChannel.cs @@ -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 \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs b/Source/MQTTnet/Server/Internal/MqttClientConnection.cs index 3f929278..203b0a25 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientConnection.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientConnection.cs @@ -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 { diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 13947ac4..13a8b3cb 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -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); } } diff --git a/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs b/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs index 4b400378..640866c7 100644 --- a/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MQTTv5/Client_Tests.cs @@ -1,4 +1,5 @@ using System.Collections.Generic; +using System.Threading; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Client; diff --git a/Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs b/Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs index 36824ee2..b815b47b 100644 --- a/Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs +++ b/Tests/MQTTnet.Core.Tests/MQTTv5/Server_Tests.cs @@ -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); diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs index a7ea3b70..2beb23df 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestClientWrapper.cs @@ -52,7 +52,7 @@ namespace MQTTnet.Tests.Mockups remove => Implementation.ApplicationMessageReceivedAsync -= value; } - public Task ConnectAsync(IMqttClientOptions options, CancellationToken cancellationToken) + public Task 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 PublishAsync(MqttApplicationMessage applicationMessage, CancellationToken cancellationToken) + public Task 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 SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken) + public Task SubscribeAsync(MqttClientSubscribeOptions options, CancellationToken cancellationToken = default) { return Implementation.SubscribeAsync(options, cancellationToken); } - public Task UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken) + public Task UnsubscribeAsync(MqttClientUnsubscribeOptions options, CancellationToken cancellationToken = default) { return Implementation.UnsubscribeAsync(options, cancellationToken); } diff --git a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs index 69cb8710..d39c7819 100644 --- a/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs +++ b/Tests/MQTTnet.Core.Tests/Mockups/TestEnvironment.cs @@ -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 ConnectClient() + public Task ConnectClient() { return ConnectClient(Factory.CreateClientOptionsBuilder().WithProtocolVersion(_protocolVersion)); } @@ -115,7 +115,7 @@ namespace MQTTnet.Tests.Mockups return client; } - public async Task ConnectClient(MqttClientOptionsBuilder options) + public async Task ConnectClient(MqttClientOptionsBuilder options) { if (options == null) throw new ArgumentNullException(nameof(options)); @@ -127,7 +127,7 @@ namespace MQTTnet.Tests.Mockups return client; } - public async Task ConnectClient(IMqttClientOptions options) + public async Task 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 { diff --git a/Tests/MQTTnet.Core.Tests/Server/General.cs b/Tests/MQTTnet.Core.Tests/Server/General.cs index 87b906fb..e2e4a5ad 100644 --- a/Tests/MQTTnet.Core.Tests/Server/General.cs +++ b/Tests/MQTTnet.Core.Tests/Server/General.cs @@ -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});