YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-27 08:42:37 +08:00
Code Issues Packages Projects Releases Wiki Activity

Improve exception handling in AsyncQueue.

Browse Source
This commit is contained in:
Christian Kratky
2020-10-30 20:43:39 +01:00
parent 58ad599783
commit f773257c86
1 changed files with 35 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Internal/AsyncQueue.cs
+35 -13
View File
@@ -7,15 +7,19 @@ namespace MQTTnet.Internal
{ {
public sealed class AsyncQueue<TItem> : IDisposable public sealed class AsyncQueue<TItem> : IDisposable
{ {
readonly object _syncRoot = new object();
SemaphoreSlim _semaphore = new SemaphoreSlim(0); SemaphoreSlim _semaphore = new SemaphoreSlim(0);
ConcurrentQueue<TItem> _queue = new ConcurrentQueue<TItem>(); ConcurrentQueue<TItem> _queue = new ConcurrentQueue<TItem>();
public int Count => _queue.Count; public int Count => _queue.Count;
public void Enqueue(TItem item) public void Enqueue(TItem item)
{ {
_queue.Enqueue(item); lock (_syncRoot)
_semaphore?.Release(); {
_queue.Enqueue(item);
_semaphore?.Release();
}
} }
public async Task<AsyncQueueDequeueResult<TItem>> TryDequeueAsync(CancellationToken cancellationToken) public async Task<AsyncQueueDequeueResult<TItem>> TryDequeueAsync(CancellationToken cancellationToken)
@@ -24,23 +28,38 @@ namespace MQTTnet.Internal
{ {
try try
{ {
if (_semaphore == null) Task task;
lock (_syncRoot)
{
if (_semaphore == null)
{
return new AsyncQueueDequeueResult<TItem>(false, default);
}
task = _semaphore.WaitAsync(cancellationToken);
}
await task.ConfigureAwait(false);
if (cancellationToken.IsCancellationRequested)
{ {
return new AsyncQueueDequeueResult<TItem>(false, default); return new AsyncQueueDequeueResult<TItem>(false, default);
} }
await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false); if (_queue.TryDequeue(out var item))
cancellationToken.ThrowIfCancellationRequested(); {
return new AsyncQueueDequeueResult<TItem>(true, item);
}
}
catch (ArgumentNullException)
{
// The semaphore throws this internally sometimes.
return new AsyncQueueDequeueResult<TItem>(false, default);
} }
catch (OperationCanceledException) catch (OperationCanceledException)
{ {
return new AsyncQueueDequeueResult<TItem>(false, default); return new AsyncQueueDequeueResult<TItem>(false, default);
} }
if (_queue.TryDequeue(out var item))
{
return new AsyncQueueDequeueResult<TItem>(true, item);
}
} }
return new AsyncQueueDequeueResult<TItem>(false, default); return new AsyncQueueDequeueResult<TItem>(false, default);
@@ -63,8 +82,11 @@ namespace MQTTnet.Internal
public void Dispose() public void Dispose()
{ {
_semaphore?.Dispose(); lock (_syncRoot)
_semaphore = null; {
_semaphore?.Dispose();
_semaphore = null;
}
} }
} }
} }