From 7e686d9aa5e463a2ee8f44946789ebea712eb945 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Thu, 15 Sep 2022 09:09:08 +0200 Subject: [PATCH] Refactor code. --- Source/MQTTnet/Client/MqttClient.cs | 4 +-- Source/MQTTnet/Internal/AsyncLock.cs | 29 ++++++++--------- .../Internal/MqttClientSessionsManager.cs | 31 +++++++++---------- 3 files changed, 29 insertions(+), 35 deletions(-) diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 52b774b2..00b0e3aa 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -91,9 +91,7 @@ namespace MQTTnet.Client add => _inspectPacketEvent.AddHandler(value); remove => _inspectPacketEvent.RemoveHandler(value); } - - [Obsolete( - "This property will be removed in the future. Checking for a working connection should be done via calling _PinAsync_. Also other successful traffic can be used to indicate connection status.")] + public bool IsConnected => (MqttClientConnectionStatus)_connectionStatus == MqttClientConnectionStatus.Connected; public MqttClientOptions Options { get; private set; } diff --git a/Source/MQTTnet/Internal/AsyncLock.cs b/Source/MQTTnet/Internal/AsyncLock.cs index 54b533ce..748ca0d8 100644 --- a/Source/MQTTnet/Internal/AsyncLock.cs +++ b/Source/MQTTnet/Internal/AsyncLock.cs @@ -10,8 +10,8 @@ namespace MQTTnet.Internal { public sealed class AsyncLock : IDisposable { - readonly object _syncRoot = new object(); readonly Task _releaser; + readonly object _syncRoot = new object(); SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1); @@ -20,6 +20,15 @@ namespace MQTTnet.Internal _releaser = Task.FromResult((IDisposable)new Releaser(this)); } + public void Dispose() + { + lock (_syncRoot) + { + _semaphore?.Dispose(); + _semaphore = null; + } + } + public Task WaitAsync(CancellationToken cancellationToken) { Task task; @@ -45,22 +54,10 @@ namespace MQTTnet.Internal } // Wait for the _WaitAsync_ method and return the releaser afterwards. - return task.ContinueWith( - (_, state) => (IDisposable)state, - _releaser.Result, - cancellationToken, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); + return task.ContinueWith((_, state) => (IDisposable)state, _releaser.Result, cancellationToken, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); } - public void Dispose() - { - lock (_syncRoot) - { - _semaphore?.Dispose(); - _semaphore = null; - } - } - - internal void Release() + void Release() { lock (_syncRoot) { @@ -83,4 +80,4 @@ namespace MQTTnet.Internal } } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 59dace26..8a2c2364 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -23,6 +23,8 @@ namespace MQTTnet.Server { readonly Dictionary _clients = new Dictionary(4096); + readonly AsyncLock _createConnectionSyncRoot = new AsyncLock(); + readonly MqttServerEventContainer _eventContainer; readonly MqttNetSourceLogger _logger; readonly MqttServerOptions _options; @@ -38,8 +40,6 @@ namespace MQTTnet.Server readonly object _sessionsManagementLock = new object(); readonly HashSet _subscriberSessions = new HashSet(); - readonly SemaphoreSlim _createConnectionSyncRoot = new SemaphoreSlim(1, 1); - public MqttClientSessionsManager( MqttServerOptions options, MqttRetainedMessagesManager retainedMessagesManager, @@ -77,7 +77,7 @@ namespace MQTTnet.Server public async Task DeleteSessionAsync(string clientId) { _logger.Verbose("Deleting session for client '{0}'.", clientId); - + MqttClient connection; lock (_clients) @@ -441,6 +441,11 @@ namespace MQTTnet.Server return GetClientSession(clientId).SubscriptionsManager.Unsubscribe(fakeUnsubscribePacket, CancellationToken.None); } + MqttClient CreateClient(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, MqttSession session) + { + return new MqttClient(connectPacket, channelAdapter, session, _options, _eventContainer, this, _rootLogger); + } + async Task CreateClientConnection( MqttConnectPacket connectPacket, MqttConnAckPacket connAckPacket, @@ -473,9 +478,8 @@ namespace MQTTnet.Server sessionShouldPersist = !connectPacket.CleanSession; } - - await _createConnectionSyncRoot.WaitAsync(CancellationToken.None).ConfigureAwait(false); - try + + using (await _createConnectionSyncRoot.WaitAsync(CancellationToken.None).ConfigureAwait(false)) { MqttSession session; lock (_sessionsManagementLock) @@ -528,24 +532,19 @@ namespace MQTTnet.Server if (_eventContainer.ClientConnectedEvent.HasHandlers) { - var eventArgs = new ClientDisconnectedEventArgs(existingClient.Id, MqttClientDisconnectType.Takeover, existingClient.Endpoint, existingClient.Session.Items); + var eventArgs = new ClientDisconnectedEventArgs( + existingClient.Id, + MqttClientDisconnectType.Takeover, + existingClient.Endpoint, + existingClient.Session.Items); await _eventContainer.ClientDisconnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false); } } } - finally - { - _createConnectionSyncRoot.Release(); - } return client; } - MqttClient CreateClient(MqttConnectPacket connectPacket, IMqttChannelAdapter channelAdapter, MqttSession session) - { - return new MqttClient(connectPacket, channelAdapter, session, _options, _eventContainer, this, _rootLogger); - } - MqttSession CreateSession(string clientId, IDictionary sessionItems, bool isPersistent) { _logger.Verbose("Created new session for client '{0}'.", clientId);