mirror of
https://github.com/dotnet/MQTTnet.git
synced 2026-10-18 02:13:00 +08:00
fixed clients overlapping from one test to another by:
- adding client and server wrappers to verify the clientid starts with the testname - adding disposable baseclass - clear disconnect handler from client so it does not try to reconnect after it is disposed - moved code from dispose to cleanup in case it is called in a stop method as dispose may only be called once while start / stop may be called multiple times
This commit is contained in:
@@ -14,7 +14,7 @@ using MQTTnet.Packets;
|
||||
|
||||
namespace MQTTnet.Adapter
|
||||
{
|
||||
public class MqttChannelAdapter : IMqttChannelAdapter
|
||||
public class MqttChannelAdapter : Disposable, IMqttChannelAdapter
|
||||
{
|
||||
private const uint ErrorOperationAborted = 0x800703E3;
|
||||
private const int ReadBufferSize = 4096; // TODO: Move buffer size to config
|
||||
@@ -26,9 +26,7 @@ namespace MQTTnet.Adapter
|
||||
private readonly MqttPacketReader _packetReader;
|
||||
|
||||
private readonly byte[] _fixedHeaderBuffer = new byte[2];
|
||||
|
||||
private bool _isDisposed;
|
||||
|
||||
|
||||
private long _bytesReceived;
|
||||
private long _bytesSent;
|
||||
|
||||
@@ -269,19 +267,13 @@ namespace MQTTnet.Adapter
|
||||
}
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
protected override void Dispose(bool disposing)
|
||||
{
|
||||
_isDisposed = true;
|
||||
|
||||
_channel?.Dispose();
|
||||
}
|
||||
|
||||
private void ThrowIfDisposed()
|
||||
{
|
||||
if (_isDisposed)
|
||||
if (disposing)
|
||||
{
|
||||
throw new ObjectDisposedException(nameof(MqttChannelAdapter));
|
||||
_channel?.Dispose();
|
||||
}
|
||||
base.Dispose(disposing);
|
||||
}
|
||||
|
||||
private static bool IsWrappedException(Exception exception)
|
||||
|
||||
@@ -20,7 +20,7 @@ using MQTTnet.Protocol;
|
||||
|
||||
namespace MQTTnet.Client
|
||||
{
|
||||
public class MqttClient : IMqttClient
|
||||
public class MqttClient : Disposable, IMqttClient
|
||||
{
|
||||
private readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
|
||||
private readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher();
|
||||
@@ -63,6 +63,8 @@ namespace MQTTnet.Client
|
||||
|
||||
ThrowIfConnected("It is not allowed to connect with a server after the connection is established.");
|
||||
|
||||
ThrowIfDisposed();
|
||||
|
||||
MqttClientAuthenticateResult authenticateResult = null;
|
||||
|
||||
try
|
||||
@@ -79,13 +81,16 @@ namespace MQTTnet.Client
|
||||
var adapter = _adapterFactory.CreateClientAdapter(options, _logger);
|
||||
_adapter = adapter;
|
||||
|
||||
_logger.Verbose($"Trying to connect with server '{options.ChannelOptions}' (Timeout={options.CommunicationTimeout}).");
|
||||
await _adapter.ConnectAsync(options.CommunicationTimeout, cancellationToken).ConfigureAwait(false);
|
||||
_logger.Verbose("Connection with server established.");
|
||||
using (var combined = CancellationTokenSource.CreateLinkedTokenSource(backgroundCancellationToken, cancellationToken))
|
||||
{
|
||||
_logger.Verbose($"Trying to connect with server '{options.ChannelOptions}' (Timeout={options.CommunicationTimeout}).");
|
||||
await _adapter.ConnectAsync(options.CommunicationTimeout, combined.Token).ConfigureAwait(false);
|
||||
_logger.Verbose("Connection with server established.");
|
||||
|
||||
_packetReceiverTask = Task.Run(() => TryReceivePacketsAsync(backgroundCancellationToken), backgroundCancellationToken);
|
||||
_packetReceiverTask = Task.Run(() => TryReceivePacketsAsync(backgroundCancellationToken), backgroundCancellationToken);
|
||||
|
||||
authenticateResult = await AuthenticateAsync(adapter, options.WillMessage, cancellationToken).ConfigureAwait(false);
|
||||
authenticateResult = await AuthenticateAsync(adapter, options.WillMessage, combined.Token).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
_sendTracker.Restart();
|
||||
|
||||
@@ -161,6 +166,7 @@ namespace MQTTnet.Client
|
||||
{
|
||||
if (options == null) throw new ArgumentNullException(nameof(options));
|
||||
|
||||
ThrowIfDisposed();
|
||||
ThrowIfNotConnected();
|
||||
|
||||
var subscribePacket = _adapter.PacketFormatterAdapter.DataConverter.CreateSubscribePacket(options);
|
||||
@@ -174,6 +180,7 @@ namespace MQTTnet.Client
|
||||
{
|
||||
if (options == null) throw new ArgumentNullException(nameof(options));
|
||||
|
||||
ThrowIfDisposed();
|
||||
ThrowIfNotConnected();
|
||||
|
||||
var unsubscribePacket = _adapter.PacketFormatterAdapter.DataConverter.CreateUnsubscribePacket(options);
|
||||
@@ -189,6 +196,7 @@ namespace MQTTnet.Client
|
||||
|
||||
MqttTopicValidator.ThrowIfInvalid(applicationMessage.Topic);
|
||||
|
||||
ThrowIfDisposed();
|
||||
ThrowIfNotConnected();
|
||||
|
||||
var publishPacket = _adapter.PacketFormatterAdapter.DataConverter.CreatePublishPacket(applicationMessage);
|
||||
@@ -214,7 +222,7 @@ namespace MQTTnet.Client
|
||||
}
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
private void Cleanup()
|
||||
{
|
||||
_backgroundCancellationTokenSource?.Cancel(false);
|
||||
_backgroundCancellationTokenSource?.Dispose();
|
||||
@@ -224,6 +232,18 @@ namespace MQTTnet.Client
|
||||
_adapter = null;
|
||||
}
|
||||
|
||||
|
||||
protected override void Dispose(bool disposing)
|
||||
{
|
||||
if (disposing)
|
||||
{
|
||||
Cleanup();
|
||||
|
||||
DisconnectedHandler = null;
|
||||
}
|
||||
base.Dispose(disposing);
|
||||
}
|
||||
|
||||
private async Task<MqttClientAuthenticateResult> AuthenticateAsync(IMqttChannelAdapter channelAdapter, MqttApplicationMessage willApplicationMessage, CancellationToken cancellationToken)
|
||||
{
|
||||
var connectPacket = channelAdapter.PacketFormatterAdapter.DataConverter.CreateConnectPacket(
|
||||
@@ -288,7 +308,7 @@ namespace MQTTnet.Client
|
||||
}
|
||||
finally
|
||||
{
|
||||
Dispose();
|
||||
Cleanup();
|
||||
_cleanDisconnectInitiated = false;
|
||||
|
||||
_logger.Info("Disconnected.");
|
||||
|
||||
@@ -8,11 +8,12 @@ using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Adapter;
|
||||
using MQTTnet.Diagnostics;
|
||||
using MQTTnet.Internal;
|
||||
using MQTTnet.Server;
|
||||
|
||||
namespace MQTTnet.Implementations
|
||||
{
|
||||
public class MqttTcpServerAdapter : IMqttServerAdapter
|
||||
public class MqttTcpServerAdapter : Disposable, IMqttServerAdapter
|
||||
{
|
||||
private readonly List<MqttTcpServerListener> _listeners = new List<MqttTcpServerListener>();
|
||||
private readonly IMqttNetChildLogger _logger;
|
||||
@@ -72,11 +73,11 @@ namespace MQTTnet.Implementations
|
||||
|
||||
public Task StopAsync()
|
||||
{
|
||||
Dispose();
|
||||
Cleanup();
|
||||
return Task.FromResult(0);
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
private void Cleanup()
|
||||
{
|
||||
_cancellationTokenSource?.Cancel(false);
|
||||
_cancellationTokenSource?.Dispose();
|
||||
@@ -90,6 +91,15 @@ namespace MQTTnet.Implementations
|
||||
_listeners.Clear();
|
||||
}
|
||||
|
||||
protected override void Dispose(bool disposing)
|
||||
{
|
||||
if (disposing)
|
||||
{
|
||||
Cleanup();
|
||||
}
|
||||
base.Dispose(disposing);
|
||||
}
|
||||
|
||||
private void RegisterListeners(MqttServerTcpEndpointBaseOptions options, X509Certificate2 tlsCertificate, CancellationToken cancellationToken)
|
||||
{
|
||||
if (!options.BoundInterNetworkAddress.Equals(IPAddress.None))
|
||||
|
||||
@@ -6,10 +6,11 @@ using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Channel;
|
||||
using MQTTnet.Client.Options;
|
||||
using MQTTnet.Internal;
|
||||
|
||||
namespace MQTTnet.Implementations
|
||||
{
|
||||
public class MqttWebSocketChannel : IMqttChannel
|
||||
public class MqttWebSocketChannel : Disposable, IMqttChannel
|
||||
{
|
||||
private readonly MqttClientWebSocketOptions _options;
|
||||
|
||||
@@ -141,22 +142,26 @@ namespace MQTTnet.Implementations
|
||||
}
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
protected override void Dispose(bool disposing)
|
||||
{
|
||||
_sendLock?.Dispose();
|
||||
_sendLock = null;
|
||||
if (disposing)
|
||||
{
|
||||
_sendLock?.Dispose();
|
||||
_sendLock = null;
|
||||
|
||||
try
|
||||
{
|
||||
_webSocket?.Dispose();
|
||||
}
|
||||
catch (ObjectDisposedException)
|
||||
{
|
||||
}
|
||||
finally
|
||||
{
|
||||
_webSocket = null;
|
||||
try
|
||||
{
|
||||
_webSocket?.Dispose();
|
||||
}
|
||||
catch (ObjectDisposedException)
|
||||
{
|
||||
}
|
||||
finally
|
||||
{
|
||||
_webSocket = null;
|
||||
}
|
||||
}
|
||||
base.Dispose(disposing);
|
||||
}
|
||||
|
||||
private IWebProxy CreateProxy()
|
||||
|
||||
@@ -4,7 +4,7 @@ using System.Threading;
|
||||
|
||||
namespace MQTTnet.Internal
|
||||
{
|
||||
public class BlockingQueue<TItem> : IDisposable
|
||||
public class BlockingQueue<TItem> : Disposable
|
||||
{
|
||||
private readonly object _syncRoot = new object();
|
||||
private readonly LinkedList<TItem> _items = new LinkedList<TItem>();
|
||||
@@ -109,9 +109,13 @@ namespace MQTTnet.Internal
|
||||
}
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
protected override void Dispose(bool disposing)
|
||||
{
|
||||
_gate.Dispose();
|
||||
if (disposing)
|
||||
{
|
||||
_gate.Dispose();
|
||||
}
|
||||
base.Dispose(disposing);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,54 @@
|
||||
using System;
|
||||
|
||||
namespace MQTTnet.Internal
|
||||
{
|
||||
public class Disposable : IDisposable
|
||||
{
|
||||
protected void ThrowIfDisposed()
|
||||
{
|
||||
if (_isDisposed)
|
||||
{
|
||||
throw new ObjectDisposedException(GetType().Name);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
#region IDisposable Support
|
||||
|
||||
private bool _isDisposed = false; // To detect redundant calls
|
||||
|
||||
protected virtual void Dispose(bool disposing)
|
||||
{
|
||||
if (disposing)
|
||||
{
|
||||
// TODO: dispose managed state (managed objects).
|
||||
}
|
||||
|
||||
// TODO: free unmanaged resources (unmanaged objects) and override a finalizer below.
|
||||
// TODO: set large fields to null.
|
||||
}
|
||||
|
||||
// TODO: override a finalizer only if Dispose(bool disposing) above has code to free unmanaged resources.
|
||||
// ~Disposable()
|
||||
// {
|
||||
// // Do not change this code. Put cleanup code in Dispose(bool disposing) above.
|
||||
// Dispose(false);
|
||||
// }
|
||||
|
||||
// This code added to correctly implement the disposable pattern.
|
||||
public void Dispose()
|
||||
{
|
||||
if (_isDisposed)
|
||||
{
|
||||
return;
|
||||
}
|
||||
// Do not change this code. Put cleanup code in Dispose(bool disposing) above.
|
||||
Dispose(true);
|
||||
// TODO: uncomment the following line if the finalizer is overridden above.
|
||||
// GC.SuppressFinalize(this);
|
||||
|
||||
_isDisposed = true;
|
||||
}
|
||||
#endregion
|
||||
}
|
||||
}
|
||||
@@ -6,7 +6,7 @@ using System.Threading.Tasks;
|
||||
|
||||
namespace MQTTnet.Server
|
||||
{
|
||||
public class MqttClientSessionApplicationMessagesQueue : IDisposable
|
||||
public class MqttClientSessionApplicationMessagesQueue : Disposable
|
||||
{
|
||||
private readonly AsyncQueue<MqttQueuedApplicationMessage> _messageQueue = new AsyncQueue<MqttQueuedApplicationMessage>();
|
||||
|
||||
@@ -71,9 +71,14 @@ namespace MQTTnet.Server
|
||||
}
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
protected override void Dispose(bool disposing)
|
||||
{
|
||||
_messageQueue.Dispose();
|
||||
if (disposing)
|
||||
{
|
||||
_messageQueue.Dispose();
|
||||
}
|
||||
|
||||
base.Dispose(disposing);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -13,7 +13,7 @@ using MQTTnet.Server.Status;
|
||||
|
||||
namespace MQTTnet.Server
|
||||
{
|
||||
public class MqttClientSessionsManager : IDisposable
|
||||
public class MqttClientSessionsManager : Disposable
|
||||
{
|
||||
private readonly AsyncQueue<MqttEnqueuedApplicationMessage> _messageQueue = new AsyncQueue<MqttEnqueuedApplicationMessage>();
|
||||
|
||||
@@ -145,9 +145,13 @@ namespace MQTTnet.Server
|
||||
_logger.Verbose("Session for client '{0}' deleted.", clientId);
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
protected override void Dispose(bool disposing)
|
||||
{
|
||||
_messageQueue?.Dispose();
|
||||
if (disposing)
|
||||
{
|
||||
_messageQueue?.Dispose();
|
||||
}
|
||||
base.Dispose(disposing);
|
||||
}
|
||||
|
||||
private async Task TryProcessQueuedApplicationMessagesAsync(CancellationToken cancellationToken)
|
||||
|
||||
Reference in New Issue
Block a user