Refactor code.

This commit is contained in:
Christian
2022-09-15 09:09:08 +02:00
parent d23bbf7bf3
commit 7e686d9aa5
3 changed files with 29 additions and 35 deletions
+1 -3
View File
@@ -91,9 +91,7 @@ namespace MQTTnet.Client
add => _inspectPacketEvent.AddHandler(value); add => _inspectPacketEvent.AddHandler(value);
remove => _inspectPacketEvent.RemoveHandler(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 bool IsConnected => (MqttClientConnectionStatus)_connectionStatus == MqttClientConnectionStatus.Connected;
public MqttClientOptions Options { get; private set; } public MqttClientOptions Options { get; private set; }
+13 -16
View File
@@ -10,8 +10,8 @@ namespace MQTTnet.Internal
{ {
public sealed class AsyncLock : IDisposable public sealed class AsyncLock : IDisposable
{ {
readonly object _syncRoot = new object();
readonly Task<IDisposable> _releaser; readonly Task<IDisposable> _releaser;
readonly object _syncRoot = new object();
SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1); SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
@@ -20,6 +20,15 @@ namespace MQTTnet.Internal
_releaser = Task.FromResult((IDisposable)new Releaser(this)); _releaser = Task.FromResult((IDisposable)new Releaser(this));
} }
public void Dispose()
{
lock (_syncRoot)
{
_semaphore?.Dispose();
_semaphore = null;
}
}
public Task<IDisposable> WaitAsync(CancellationToken cancellationToken) public Task<IDisposable> WaitAsync(CancellationToken cancellationToken)
{ {
Task task; Task task;
@@ -45,22 +54,10 @@ namespace MQTTnet.Internal
} }
// Wait for the _WaitAsync_ method and return the releaser afterwards. // Wait for the _WaitAsync_ method and return the releaser afterwards.
return task.ContinueWith( return task.ContinueWith((_, state) => (IDisposable)state, _releaser.Result, cancellationToken, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
(_, state) => (IDisposable)state,
_releaser.Result,
cancellationToken, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
} }
public void Dispose() void Release()
{
lock (_syncRoot)
{
_semaphore?.Dispose();
_semaphore = null;
}
}
internal void Release()
{ {
lock (_syncRoot) lock (_syncRoot)
{ {
@@ -83,4 +80,4 @@ namespace MQTTnet.Internal
} }
} }
} }
} }
@@ -23,6 +23,8 @@ namespace MQTTnet.Server
{ {
readonly Dictionary<string, MqttClient> _clients = new Dictionary<string, MqttClient>(4096); readonly Dictionary<string, MqttClient> _clients = new Dictionary<string, MqttClient>(4096);
readonly AsyncLock _createConnectionSyncRoot = new AsyncLock();
readonly MqttServerEventContainer _eventContainer; readonly MqttServerEventContainer _eventContainer;
readonly MqttNetSourceLogger _logger; readonly MqttNetSourceLogger _logger;
readonly MqttServerOptions _options; readonly MqttServerOptions _options;
@@ -38,8 +40,6 @@ namespace MQTTnet.Server
readonly object _sessionsManagementLock = new object(); readonly object _sessionsManagementLock = new object();
readonly HashSet<MqttSession> _subscriberSessions = new HashSet<MqttSession>(); readonly HashSet<MqttSession> _subscriberSessions = new HashSet<MqttSession>();
readonly SemaphoreSlim _createConnectionSyncRoot = new SemaphoreSlim(1, 1);
public MqttClientSessionsManager( public MqttClientSessionsManager(
MqttServerOptions options, MqttServerOptions options,
MqttRetainedMessagesManager retainedMessagesManager, MqttRetainedMessagesManager retainedMessagesManager,
@@ -77,7 +77,7 @@ namespace MQTTnet.Server
public async Task DeleteSessionAsync(string clientId) public async Task DeleteSessionAsync(string clientId)
{ {
_logger.Verbose("Deleting session for client '{0}'.", clientId); _logger.Verbose("Deleting session for client '{0}'.", clientId);
MqttClient connection; MqttClient connection;
lock (_clients) lock (_clients)
@@ -441,6 +441,11 @@ namespace MQTTnet.Server
return GetClientSession(clientId).SubscriptionsManager.Unsubscribe(fakeUnsubscribePacket, CancellationToken.None); 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<MqttClient> CreateClientConnection( async Task<MqttClient> CreateClientConnection(
MqttConnectPacket connectPacket, MqttConnectPacket connectPacket,
MqttConnAckPacket connAckPacket, MqttConnAckPacket connAckPacket,
@@ -473,9 +478,8 @@ namespace MQTTnet.Server
sessionShouldPersist = !connectPacket.CleanSession; sessionShouldPersist = !connectPacket.CleanSession;
} }
await _createConnectionSyncRoot.WaitAsync(CancellationToken.None).ConfigureAwait(false); using (await _createConnectionSyncRoot.WaitAsync(CancellationToken.None).ConfigureAwait(false))
try
{ {
MqttSession session; MqttSession session;
lock (_sessionsManagementLock) lock (_sessionsManagementLock)
@@ -528,24 +532,19 @@ namespace MQTTnet.Server
if (_eventContainer.ClientConnectedEvent.HasHandlers) 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); await _eventContainer.ClientDisconnectedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
} }
} }
} }
finally
{
_createConnectionSyncRoot.Release();
}
return client; 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) MqttSession CreateSession(string clientId, IDictionary sessionItems, bool isPersistent)
{ {
_logger.Verbose("Created new session for client '{0}'.", clientId); _logger.Verbose("Created new session for client '{0}'.", clientId);