Nov 7, 2024 · Nov 12, 2024 · Nov 13, 2024 · Nov 13, 2024 · Nov 13, 2024 · Nov 13, 2024
diff --git a/Source/MQTTnet.AspnetCore/Internal/MqttChannel.cs b/Source/MQTTnet.AspnetCore/Internal/MqttChannel.cs
 using MQTTnet.Packets;
 using System;
 using System.Buffers;
 using System.IO;
 using System.IO.Pipelines;
 using System.Net;
 using System.Net.Sockets;
 using System.Runtime.InteropServices;
 using System.Security.Cryptography.X509Certificates;
 using System.Threading;
 using System.Threading.Tasks;

    public async Task DisconnectAsync()
    {
        await _input.CompleteAsync().ConfigureAwait(false);
        await _output.CompleteAsync().ConfigureAwait(false);
        try
        {
            await _input.CompleteAsync().ConfigureAwait(false);
            await _output.CompleteAsync().ConfigureAwait(false);
        }
        catch (Exception exception)
        {
            if (!WrapAndThrowException(exception))
            {
                throw;
            }
        }
    }

    public virtual void Dispose()
    }

    public async Task<MqttPacket?> ReceivePacketAsync(CancellationToken cancellationToken)
    {
        try
        {
            return await ReceivePacketCoreAsync(cancellationToken).ConfigureAwait(false);
        }
        catch (OperationCanceledException)
        {
        }
        catch (ObjectDisposedException)
        {
        }
        catch (Exception exception)
        {
            if (!WrapAndThrowException(exception))
            {
                throw;
            }
        }

        return null;
    }

    private async Task<MqttPacket?> ReceivePacketCoreAsync(CancellationToken cancellationToken)
    {
        try
        {
                {
                    if (!buffer.IsEmpty)
                    {
                        if (PacketFormatterAdapter.TryDecode(buffer,_packetInspector, out var packet, out consumed, out observed, out var received))
                        if (PacketFormatterAdapter.TryDecode(buffer,_packetInspector, out var packet, out consumed, out observed, out var received))
                        {
                            BytesReceived += received;

                }
            }
        }
        catch (Exception exception)
        catch (Exception)
        {
            // completing the channel makes sure that there is no more data read after a protocol error
            _input.Complete(exception);
            _output.Complete(exception);
 await_input.CompleteAsync().ConfigureAwait(false);
 await_output.CompleteAsync().ConfigureAwait(false);

            throw;
        }
    }

    public async Task SendPacketAsync(MqttPacket packet, CancellationToken cancellationToken)
    {
        try
        {
            await SendPacketCoreAsync(packet, cancellationToken).ConfigureAwait(false);
        }
        catch (Exception exception)
        {
            if (!WrapAndThrowException(exception))
            {
                throw;
            }
        }
    }

    private async Task SendPacketCoreAsync(MqttPacket packet, CancellationToken cancellationToken)
    {
        using (await _writerLock.EnterAsync(cancellationToken).ConfigureAwait(false))
        {
        buffer.Payload.CopyTo(destination: span.Slice(offset));
        output.Advance(buffer.Length);
    }

    public static bool WrapAndThrowException(Exception exception)
    {
        if (exception is OperationCanceledException ||
            exception is MqttCommunicationTimedOutException ||
            exception is MqttCommunicationException ||
            exception is MqttProtocolViolationException)
        {
            return false;
        }

        if (exception is IOException && exception.InnerException is SocketException innerException)
        {
            exception = innerException;
        }

        if (exception is SocketException socketException)
        {
            if (socketException.SocketErrorCode == SocketError.OperationAborted)
            {
                throw new OperationCanceledException();
            }

            if (socketException.SocketErrorCode == SocketError.ConnectionAborted)
            {
                throw new MqttCommunicationException(socketException);
            }
        }

        if (exception is COMException comException)
        {
            const uint ErrorOperationAborted = 0x800703E3;
            if ((uint)comException.HResult == ErrorOperationAborted)
            {
                throw new OperationCanceledException();
            }
        }

        throw new MqttCommunicationException(exception);
    }
 }
diff --git a/Source/MQTTnet.AspnetCore/Internal/MqttClientChannelAdapter.cs b/Source/MQTTnet.AspnetCore/Internal/MqttClientChannelAdapter.cs

    public async Task ConnectAsync(CancellationToken cancellationToken)
    {
 _connection = _channelOptions switch
 try
        {
            MqttClientTcpOptions tcpOptions => await ClientConnectionContext.CreateAsync(tcpOptions, cancellationToken).ConfigureAwait(false),
            MqttClientWebSocketOptions webSocketOptions => await ClientConnectionContext.CreateAsync(webSocketOptions, cancellationToken).ConfigureAwait(false),
            _ => throw new NotSupportedException(),
        };
        _channel = new MqttChannel(_packetFormatterAdapter, _connection, _packetInspector, _allowPacketFragmentation);
            _connection = _channelOptions switch
            {
                MqttClientTcpOptions tcpOptions => await ClientConnectionContext.CreateAsync(tcpOptions, cancellationToken).ConfigureAwait(false),
                MqttClientWebSocketOptions webSocketOptions => await ClientConnectionContext.CreateAsync(webSocketOptions, cancellationToken).ConfigureAwait(false),
                _ => throw new NotSupportedException(),
            };
            _channel = new MqttChannel(_packetFormatterAdapter, _connection, _packetInspector, _allowPacketFragmentation);
        }
        catch (Exception ex)
        {
            if (!MqttChannel.WrapAndThrowException(ex))
            {
                throw;
            }
        }
    }

    public Task DisconnectAsync(CancellationToken cancellationToken)
diff --git a/Source/MQTTnet.Tests/ASP/MqttConnectionContextTest.cs b/Source/MQTTnet.Tests/ASP/MqttConnectionContextTest.cs
            await Assert.ThrowsExceptionAsync<MqttProtocolViolationException>(() => ctx.ReceivePacketAsync(CancellationToken.None));

            // the first exception should complete the pipes so if someone tries to use the connection after that it should throw immidiatly
            await Assert.ThrowsExceptionAsync<InvalidOperationException>(() => ctx.ReceivePacketAsync(CancellationToken.None));
            await Assert.ThrowsExceptionAsync<MqttCommunicationException>(() => ctx.ReceivePacketAsync(CancellationToken.None));
        }

        // TODO: Fix test
diff --git a/Source/MQTTnet.Tests/Clients/LowLevelMqttClient/LowLevelMqttClient_Tests.cs b/Source/MQTTnet.Tests/Clients/LowLevelMqttClient/LowLevelMqttClient_Tests.cs
        [TestMethod]
        public async Task Loose_Connection()
        {
            using var testEnvironments =CreateMQTTnetTestEnvironment();
            using var testEnvironments =CreateMixedTestEnvironment();
            foreach (var testEnvironment in testEnvironments)
            {
                testEnvironment.IgnoreServerLogErrors = true;
diff --git a/Source/MQTTnet.Tests/MQTTv5/Server_Tests.cs b/Source/MQTTnet.Tests/MQTTv5/Server_Tests.cs
        }

        [TestMethod]
        public async Task Disconnect_with_Reason_MQTTnet()
        {
            using var testEnvironments = CreateMQTTnetTestEnvironment();
            await Disconnect_with_Reason(testEnvironments);
        }

        [TestMethod]
        public async Task Disconnect_with_Reason_AspNetCore()
        {
            using var testEnvironments = CreateAspNetCoreTestEnvironment();
            await Disconnect_with_Reason(testEnvironments);
        }

        private async Task Disconnect_with_Reason(TestEnvironmentCollection testEnvironments)
        public async Task Disconnect_with_Reason()
        {
            using var testEnvironments = CreateMixedTestEnvironment();
            foreach (var testEnvironment in testEnvironments)
            {
                var disconnectReasonTaskSource = new TaskCompletionSource<MqttClientDisconnectReason>();
diff --git a/Source/MQTTnet.Tests/Mockups/AspNetCoreTestEnvironment.cs b/Source/MQTTnet.Tests/Mockups/AspNetCoreTestEnvironment.cs
 using Microsoft.Extensions.Hosting;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.AspNetCore;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Formatter;
 using MQTTnet.Internal;
 using MQTTnet.LowLevelClient;
        private IMqttClientFactory CreateClientFactory()
        {
            var services = new ServiceCollection();
            var clientBuilder = services.AddMqttClient();
            UseMqttLogger(clientBuilder, "[CLIENT]=>");

            var logger = EnableLogger ? (IMqttNetLogger)ClientLogger : MqttNetNullLogger.Instance;
            services.AddSingleton(logger);
            services.AddMqttClient();

            return services.BuildServiceProvider().GetRequiredService<IMqttClientFactory>();
        }

        public override Task<MqttServer> StartServer(MqttServerOptionsBuilder optionsBuilder)
        {
            optionsBuilder.WithDefaultEndpoint();
            optionsBuilder.WithDefaultEndpointPort(ServerPort);
            optionsBuilder.WithMaxPendingMessagesPerClient(int.MaxValue);
            var serverOptions = optionsBuilder.Build();
            return StartServer(serverOptions);
            var appBuilder = WebApplication.CreateBuilder();
            appBuilder.Services.AddSingleton(serverOptions);

            var serverBuilder = appBuilder.Services.AddMqttServer();
            UseMqttLogger(serverBuilder, "[SERVER]=>");
            var logger = EnableLogger ? (IMqttNetLogger)ServerLogger : new MqttNetNullLogger();
            appBuilder.Services.AddSingleton(logger);
            appBuilder.Services.AddMqttServer();

            appBuilder.WebHost.UseKestrel(k => k.ListenMqtt());
            appBuilder.Host.ConfigureHostOptions(h => h.ShutdownTimeout = TimeSpan.FromMilliseconds(500d));

            _app = appBuilder.Build();
            await _app.StartAsync();

            Server = _app.Services.GetRequiredService<MqttServer>();
            ServerPort = serverOptions.DefaultEndpointOptions.Port;
                return CompletedTask.Instance;
            };

            var appStartedSource = new TaskCompletionSource();
            _app.Lifetime.ApplicationStarted.Register(() => appStartedSource.TrySetResult());

            await _app.StartAsync();
            await appStartedSource.Task;

            return Server;
        }

            return port;
        }

        private void UseMqttLogger(IMqttBuilder builder, string categoryNamePrefix)
        {
            if (EnableLogger)
            {
                builder.UseAspNetCoreMqttNetLogger(l => l.CategoryNamePrefix = categoryNamePrefix);
            }
            else
            {
                builder.UseMqttNetNullLogger();
            }
        }

        public override void Dispose()
        {
            base.Dispose();
diff --git a/Source/MQTTnet.Tests/Server/Keep_Alive_Tests.cs b/Source/MQTTnet.Tests/Server/Keep_Alive_Tests.cs
    public sealed class KeepAlive_Tests : BaseTestClass
    {
        [TestMethod]
        public async Task Disconnect_Client_DueTo_KeepAlive_MQTTnet()
        {
            using var testEnvironments = CreateMQTTnetTestEnvironment();
            await Disconnect_Client_DueTo_KeepAlive(testEnvironments);
        }

        [TestMethod]
        public async Task Disconnect_Client_DueTo_KeepAlive_AspNetCore()
        {
            using var testEnvironments = CreateAspNetCoreTestEnvironment();
            await Disconnect_Client_DueTo_KeepAlive(testEnvironments);
        }

        private async Task Disconnect_Client_DueTo_KeepAlive(TestEnvironmentCollection testEnvironments)
        public async Task Disconnect_Client_DueTo_KeepAlive()
        {
            using var testEnvironments = CreateMixedTestEnvironment();
            foreach (var testEnvironment in testEnvironments)
            {
                await testEnvironment.StartServer();
diff --git a/Source/MQTTnet.Tests/Server/Session_Tests.cs b/Source/MQTTnet.Tests/Server/Session_Tests.cs
        }

        [TestMethod]
        public async Task Session_Takeover_MQTTnet()
        {
            using var testEnvironments = CreateMQTTnetTestEnvironment();
            await Session_Takeover(testEnvironments);
        }

        [TestMethod]
        public async Task Session_Takeover_AspNetCore()
        {
            using var testEnvironments = CreateAspNetCoreTestEnvironment();
            await Session_Takeover(testEnvironments);
        }


        private async Task Session_Takeover(TestEnvironmentCollection testEnvironments)
        public async Task Session_Takeover()
        {
            using var testEnvironments = CreateMixedTestEnvironment();
            foreach (var testEnvironment in testEnvironments)
            {
                await testEnvironment.StartServer();
Original file line number	Diff line number	Diff line change
Expand Up		@@ -12,8 +12,11 @@
		using MQTTnet.Packets;
		using System;
		using System.Buffers;
		using System.IO;
		using System.IO.Pipelines;
		using System.Net;
		using System.Net.Sockets;
		using System.Runtime.InteropServices;
		using System.Security.Cryptography.X509Certificates;
		using System.Threading;
		using System.Threading.Tasks;
Expand DownExpand Up		@@ -106,8 +109,18 @@ private static bool IsTlsConnection(IHttpContextFeature? _httpContextFeature, IT

		public async Task DisconnectAsync()
		{
		await _input.CompleteAsync().ConfigureAwait(false);
		await _output.CompleteAsync().ConfigureAwait(false);
		try
		{
		await _input.CompleteAsync().ConfigureAwait(false);
		await _output.CompleteAsync().ConfigureAwait(false);
		}
		catch (Exception exception)
		{
		if (!WrapAndThrowException(exception))
		{
		throw;
		}
		}
		}

		public virtual void Dispose()
Expand All		@@ -116,6 +129,29 @@ public virtual void Dispose()
		}

		public async Task<MqttPacket?> ReceivePacketAsync(CancellationToken cancellationToken)
		{
		try
		{
		return await ReceivePacketCoreAsync(cancellationToken).ConfigureAwait(false);
		}
		catch (OperationCanceledException)
		{
		}
		catch (ObjectDisposedException)
		{
		}
		catch (Exception exception)
		{
		if (!WrapAndThrowException(exception))
		{
		throw;
		}
		}

		return null;
		}

		private async Task<MqttPacket?> ReceivePacketCoreAsync(CancellationToken cancellationToken)
		{
		try
		{
Expand DownExpand Up		@@ -143,7 +179,7 @@ public virtual void Dispose()
		{
		if (!buffer.IsEmpty)
		{
		if (PacketFormatterAdapter.TryDecode(buffer,_packetInspector, out var packet, out consumed, out observed, out var received))
		if (PacketFormatterAdapter.TryDecode(buffer,_packetInspector, out var packet, out consumed, out observed, out var received))
		{
		BytesReceived += received;

Expand All		@@ -168,11 +204,11 @@ public virtual void Dispose()
		}
		}
		}
		catch (Exception exception)
		catch (Exception)
		{
		// completing the channel makes sure that there is no more data read after a protocol error
		_input.Complete(exception);
		_output.Complete(exception);
		await_input.CompleteAsync().ConfigureAwait(false);
		await_output.CompleteAsync().ConfigureAwait(false);

		throw;
		}
Expand All		@@ -188,6 +224,21 @@ public void ResetStatistics()
		}

		public async Task SendPacketAsync(MqttPacket packet, CancellationToken cancellationToken)
		{
		try
		{
		await SendPacketCoreAsync(packet, cancellationToken).ConfigureAwait(false);
		}
		catch (Exception exception)
		{
		if (!WrapAndThrowException(exception))
		{
		throw;
		}
		}
		}

		private async Task SendPacketCoreAsync(MqttPacket packet, CancellationToken cancellationToken)
		{
		using (await _writerLock.EnterAsync(cancellationToken).ConfigureAwait(false))
		{
Expand DownExpand Up		@@ -241,4 +292,44 @@ static void WritePacketBuffer(PipeWriter output, MqttPacketBuffer buffer)
		buffer.Payload.CopyTo(destination: span.Slice(offset));
		output.Advance(buffer.Length);
		}

		public static bool WrapAndThrowException(Exception exception)
		{
		if (exception is OperationCanceledException \|\|
		exception is MqttCommunicationTimedOutException \|\|
		exception is MqttCommunicationException \|\|
		exception is MqttProtocolViolationException)
		{
		return false;
		}

		if (exception is IOException && exception.InnerException is SocketException innerException)
		{
		exception = innerException;
		}

		if (exception is SocketException socketException)
		{
		if (socketException.SocketErrorCode == SocketError.OperationAborted)
		{
		throw new OperationCanceledException();
		}

		if (socketException.SocketErrorCode == SocketError.ConnectionAborted)
		{
		throw new MqttCommunicationException(socketException);
		}
		}

		if (exception is COMException comException)
		{
		const uint ErrorOperationAborted = 0x800703E3;
		if ((uint)comException.HResult == ErrorOperationAborted)
		{
		throw new OperationCanceledException();
		}
		}

		throw new MqttCommunicationException(exception);
		}
		}
Original file line number	Diff line number	Diff line change
Expand Up		@@ -52,13 +52,23 @@ public MqttClientChannelAdapter(

		public async Task ConnectAsync(CancellationToken cancellationToken)
		{
		_connection = _channelOptions switch
		try
		{
		MqttClientTcpOptions tcpOptions => await ClientConnectionContext.CreateAsync(tcpOptions, cancellationToken).ConfigureAwait(false),
		MqttClientWebSocketOptions webSocketOptions => await ClientConnectionContext.CreateAsync(webSocketOptions, cancellationToken).ConfigureAwait(false),
		_ => throw new NotSupportedException(),
		};
		_channel = new MqttChannel(_packetFormatterAdapter, _connection, _packetInspector, _allowPacketFragmentation);
		_connection = _channelOptions switch
		{
		MqttClientTcpOptions tcpOptions => await ClientConnectionContext.CreateAsync(tcpOptions, cancellationToken).ConfigureAwait(false),
		MqttClientWebSocketOptions webSocketOptions => await ClientConnectionContext.CreateAsync(webSocketOptions, cancellationToken).ConfigureAwait(false),
		_ => throw new NotSupportedException(),
		};
		_channel = new MqttChannel(_packetFormatterAdapter, _connection, _packetInspector, _allowPacketFragmentation);
		}
		catch (Exception ex)
		{
		if (!MqttChannel.WrapAndThrowException(ex))
		{
		throw;
		}
		}
		}

		public Task DisconnectAsync(CancellationToken cancellationToken)
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -37,7 +37,7 @@ public async Task TestCorruptedConnectPacket()
		await Assert.ThrowsExceptionAsync<MqttProtocolViolationException>(() => ctx.ReceivePacketAsync(CancellationToken.None));

		// the first exception should complete the pipes so if someone tries to use the connection after that it should throw immidiatly
		await Assert.ThrowsExceptionAsync<InvalidOperationException>(() => ctx.ReceivePacketAsync(CancellationToken.None));
		await Assert.ThrowsExceptionAsync<MqttCommunicationException>(() => ctx.ReceivePacketAsync(CancellationToken.None));
		}

		// TODO: Fix test
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -80,7 +80,7 @@ public async Task Connect_To_Wrong_Host()
		[TestMethod]
		public async Task Loose_Connection()
		{
		using var testEnvironments =CreateMQTTnetTestEnvironment();
		using var testEnvironments =CreateMixedTestEnvironment();
		foreach (var testEnvironment in testEnvironments)
		{
		testEnvironment.IgnoreServerLogErrors = true;
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -182,21 +182,9 @@ public async Task Reconnect_with_different_SessionExpiryInterval()
		}

		[TestMethod]
		public async Task Disconnect_with_Reason_MQTTnet()
		{
		using var testEnvironments = CreateMQTTnetTestEnvironment();
		await Disconnect_with_Reason(testEnvironments);
		}

		[TestMethod]
		public async Task Disconnect_with_Reason_AspNetCore()
		{
		using var testEnvironments = CreateAspNetCoreTestEnvironment();
		await Disconnect_with_Reason(testEnvironments);
		}

		private async Task Disconnect_with_Reason(TestEnvironmentCollection testEnvironments)
		public async Task Disconnect_with_Reason()
		{
		using var testEnvironments = CreateMixedTestEnvironment();
		foreach (var testEnvironment in testEnvironments)
		{
		var disconnectReasonTaskSource = new TaskCompletionSource<MqttClientDisconnectReason>();
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -8,6 +8,7 @@
		using Microsoft.Extensions.Hosting;
		using Microsoft.VisualStudio.TestTools.UnitTesting;
		using MQTTnet.AspNetCore;
		using MQTTnet.Diagnostics.Logger;
		using MQTTnet.Formatter;
		using MQTTnet.Internal;
		using MQTTnet.LowLevelClient;
Expand DownExpand Up		@@ -47,8 +48,11 @@ protected override ILowLevelMqttClient CreateLowLevelClientCore()
		private IMqttClientFactory CreateClientFactory()
		{
		var services = new ServiceCollection();
		var clientBuilder = services.AddMqttClient();
		UseMqttLogger(clientBuilder, "[CLIENT]=>");

		var logger = EnableLogger ? (IMqttNetLogger)ClientLogger : MqttNetNullLogger.Instance;
		services.AddSingleton(logger);
		services.AddMqttClient();

		return services.BuildServiceProvider().GetRequiredService<IMqttClientFactory>();
		}

Expand All		@@ -67,6 +71,7 @@ public override Task<MqttServer> StartServer(Action<MqttServerOptionsBuilder> co
		public override Task<MqttServer> StartServer(MqttServerOptionsBuilder optionsBuilder)
		{
		optionsBuilder.WithDefaultEndpoint();
		optionsBuilder.WithDefaultEndpointPort(ServerPort);
		optionsBuilder.WithMaxPendingMessagesPerClient(int.MaxValue);
		var serverOptions = optionsBuilder.Build();
		return StartServer(serverOptions);
Expand All		@@ -88,14 +93,14 @@ private async Task<MqttServer> StartServer(MqttServerOptions serverOptions)
		var appBuilder = WebApplication.CreateBuilder();
		appBuilder.Services.AddSingleton(serverOptions);

		var serverBuilder = appBuilder.Services.AddMqttServer();
		UseMqttLogger(serverBuilder, "[SERVER]=>");
		var logger = EnableLogger ? (IMqttNetLogger)ServerLogger : new MqttNetNullLogger();
		appBuilder.Services.AddSingleton(logger);
		appBuilder.Services.AddMqttServer();

		appBuilder.WebHost.UseKestrel(k => k.ListenMqtt());
		appBuilder.Host.ConfigureHostOptions(h => h.ShutdownTimeout = TimeSpan.FromMilliseconds(500d));

		_app = appBuilder.Build();
		await _app.StartAsync();

		Server = _app.Services.GetRequiredService<MqttServer>();
		ServerPort = serverOptions.DefaultEndpointOptions.Port;
Expand All		@@ -115,6 +120,12 @@ private async Task<MqttServer> StartServer(MqttServerOptions serverOptions)
		return CompletedTask.Instance;
		};

		var appStartedSource = new TaskCompletionSource();
		_app.Lifetime.ApplicationStarted.Register(() => appStartedSource.TrySetResult());

		await _app.StartAsync();
		await appStartedSource.Task;

		return Server;
		}

Expand All		@@ -132,18 +143,6 @@ private static int GetServerPort()
		return port;
		}

		private void UseMqttLogger(IMqttBuilder builder, string categoryNamePrefix)
		{
		if (EnableLogger)
		{
		builder.UseAspNetCoreMqttNetLogger(l => l.CategoryNamePrefix = categoryNamePrefix);
		}
		else
		{
		builder.UseMqttNetNullLogger();
		}
		}

		public override void Dispose()
		{
		base.Dispose();
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -17,21 +17,9 @@ namespace MQTTnet.Tests.Server
		public sealed class KeepAlive_Tests : BaseTestClass
		{
		[TestMethod]
		public async Task Disconnect_Client_DueTo_KeepAlive_MQTTnet()
		{
		using var testEnvironments = CreateMQTTnetTestEnvironment();
		await Disconnect_Client_DueTo_KeepAlive(testEnvironments);
		}

		[TestMethod]
		public async Task Disconnect_Client_DueTo_KeepAlive_AspNetCore()
		{
		using var testEnvironments = CreateAspNetCoreTestEnvironment();
		await Disconnect_Client_DueTo_KeepAlive(testEnvironments);
		}

		private async Task Disconnect_Client_DueTo_KeepAlive(TestEnvironmentCollection testEnvironments)
		public async Task Disconnect_Client_DueTo_KeepAlive()
		{
		using var testEnvironments = CreateMixedTestEnvironment();
		foreach (var testEnvironment in testEnvironments)
		{
		await testEnvironment.StartServer();
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -228,22 +228,9 @@ public async Task Retry_If_Not_PubAck(MqttQualityOfServiceLevel qos)
		}

		[TestMethod]
		public async Task Session_Takeover_MQTTnet()
		{
		using var testEnvironments = CreateMQTTnetTestEnvironment();
		await Session_Takeover(testEnvironments);
		}

		[TestMethod]
		public async Task Session_Takeover_AspNetCore()
		{
		using var testEnvironments = CreateAspNetCoreTestEnvironment();
		await Session_Takeover(testEnvironments);
		}


		private async Task Session_Takeover(TestEnvironmentCollection testEnvironments)
		public async Task Session_Takeover()
		{
		using var testEnvironments = CreateMixedTestEnvironment();
		foreach (var testEnvironment in testEnvironments)
		{
		await testEnvironment.StartServer();
Expand Down